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

feat(core): make targeted transfers resumable and idempotent

Browse Source 
Preserve approved transfer auth and state across restart, resume without
re-approval, and keep cancel/delete from leaving usable orphan authorization.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Hammed Abass
2026-08-11 03:56:46 +02:00
parent e2cfad7fb3
commit 8798edf0f2
12 changed files with 1297 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -96,6 +96,8 @@ pub struct TargetedTransfer {
pub manifest_id: String,
pub file_count: u64,
pub total_size: u64,
/// Bytes verified so far; survives interruption for resume.
pub verified_bytes: u64,
pub state: TargetedTransferState,
pub created_at: i64,
pub updated_at: i64,

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

@@ -543,6 +543,39 @@ impl VnidropCore {
self.block_on(self.inner.list_targeted_transfers())
}
/// Withdraw an offer or revoke an approved transfer.
///
/// Stops active streaming synchronously before asynchronous cleanup.
pub fn cancel_targeted_transfer(&self, id: String) -> Result<(), VnidropError> {
if let Ok(Some(row)) = self.block_on(self.inner.targeted_store().get_row(&id)) {
let _ = self
.inner
.signal_targeted_transfer_cancel(row.protocol_transfer_id);
}
self.block_on(self.inner.cancel_targeted_transfer(id))
}
/// Durably remove authorization, resumable state, and content service.
///
/// Local denial is mandatory even when remote cleanup fails.
pub fn delete_targeted_transfer(&self, id: String) -> Result<(), VnidropError> {
if let Ok(Some(row)) = self.block_on(self.inner.targeted_store().get_row(&id)) {
let _ = self
.inner
.signal_targeted_transfer_cancel(row.protocol_transfer_id);
}
self.block_on(self.inner.delete_targeted_transfer(id))
}
/// Resume an approved/interrupted transfer without another approval.
pub fn resume_targeted_transfer(
&self,
id: String,
output_dir: String,
) -> Result<(), VnidropError> {
self.block_on(self.inner.resume_targeted_transfer(id, output_dir))
}
/// Devices the user has chosen to remember.
pub fn list_contacts(&self) -> Result<Vec<ContactSummary>, VnidropError> {
self.block_on(self.inner.list_contacts())

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

@@ -106,6 +106,7 @@ pub(super) struct CoreInner {
pub(super) store: FsStore,
pub(super) repository: Repository,
_profile_lock: Option<ProfileLock>,
pub(super) secret_custody: Option<Arc<crate::secure_secret::SecretCustody>>,
pub(super) event_hub: Arc<EventHub>,
pub(super) approval: ApprovalService,
pub(super) pairing: PairingService,
@@ -420,6 +421,7 @@ impl CoreInner {
TargetedTransferProtocol::new(
device_relationships.clone(),
targeted_offers.clone(),
repository.sqlite_pool(),
limits.clone(),
endpoint.id().to_string(),
),
@@ -433,6 +435,7 @@ impl CoreInner {
store,
repository,
_profile_lock: profile_lock,
secret_custody: secret_custody.clone(),
event_hub,
approval,
pairing,
@@ -458,6 +461,15 @@ impl CoreInner {
targeted_cancel_log: std::sync::Mutex::new(Vec::new()),
});
// In-flight connecting/transferring transfers become Interrupted across restart.
if let Err(error) = inner.targeted_store().mark_interrupted_in_flight().await {
tracing::warn!(%error, "failed to mark in-flight targeted transfers interrupted");
}
// Restore permanent receiver ACLs for approved targeted shares.
if let Err(error) = inner.restore_targeted_transfer_access().await {
tracing::warn!(%error, "failed to restore targeted transfer access");
}
inner.emit_endpoint(
"startup",
"endpoint-online",

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

@@ -1,4 +1,4 @@
//! Create, offer, approve, and receive targeted transfers between Saved devices.
//! Create, offer, approve, resume, cancel, and delete targeted transfers.
use std::sync::Arc;
@@ -13,13 +13,15 @@ use crate::{
ShareSource, TargetedTransfer, TargetedTransferState, TransferAccessMode, TransferMetadata,
},
error::VnidropError,
secure_secret::{SecretHandle, SecretKind},
targeted_transfer::{
auth_secret_material,
protocol::{
DeliverTargetedAuthorization, SubmitTargetedOffer, TargetedOfferResponse,
TargetedTransferProtocol,
CancelTargetedOffer, DeliverTargetedAuthorization, SubmitTargetedOffer,
TargetedOfferResponse, TargetedTransferProtocol,
},
TargetedAuthorization, TargetedAuthorizationDraft, TargetedTransferRow,
TargetedTransferStore,
reconstruct_authorization, TargetedAuthorization, TargetedAuthorizationDraft,
TargetedTransferRole, TargetedTransferRow, TargetedTransferStore,
},
ticket::VnidropTicket,
util::{non_empty, now_ms},
@@ -49,6 +51,15 @@ impl CoreInner {
self.targeted_store().list().await
}
pub(crate) async fn restore_targeted_transfer_access(&self) -> Result<(), VnidropError> {
for row in self.targeted_store().list_resumable_sender_rows().await? {
self.access_policy
.approve_endpoint_until(row.protocol_transfer_id, row.receiver_endpoint_id, None)
.await;
}
Ok(())
}
/// Cancel in-flight targeted transfers involving `peer` (for forget/block).
pub(crate) async fn cancel_targeted_transfers_for_peer(
&self,
@@ -76,13 +87,128 @@ impl CoreInner {
self.targeted_store().cancel_by_peer(peer_endpoint_id).await
}
/// Synchronously stop streaming for one transfer (facade calls this first).
pub(super) fn signal_targeted_transfer_cancel(&self, protocol_transfer_id: u64) -> bool {
self.take_active_transfer(protocol_transfer_id).is_some()
}
pub(super) async fn cancel_targeted_transfer(&self, id: String) -> Result<(), VnidropError> {
let store = self.targeted_store();
let Some(row) = store.get_row(&id).await? else {
// Still drop any live-session offer under this id.
self.targeted_offers.discard(&id).await;
return Ok(());
};
let _ = self.take_active_transfer(row.protocol_transfer_id);
self.targeted_offers.discard(&id).await;
self.access_policy
.remove_transfer(row.protocol_transfer_id)
.await;
let _ = self.cancel_idle_or_share(row.protocol_transfer_id).await;
if !matches!(
row.state,
TargetedTransferState::Completed
| TargetedTransferState::Declined
| TargetedTransferState::Cancelled
| TargetedTransferState::Failed
| TargetedTransferState::Deleted
) {
let _ = store
.set_state_from_any(&id, TargetedTransferState::Cancelled)
.await;
}
// Best-effort remote withdraw of an unapproved live offer.
if row.role == TargetedTransferRole::Sender
&& matches!(
row.state,
TargetedTransferState::Offering | TargetedTransferState::AwaitingApproval
)
{
if let Ok(addr) = self
.device_relationships
.peer_addr(&row.receiver_endpoint_id)
.await
{
let client = TargetedTransferProtocol::client(self.endpoint.clone(), addr);
let _ = tokio::time::timeout(
OFFER_CONNECT_TIMEOUT,
client.cancel_offer(CancelTargetedOffer {
transfer_id: id.clone(),
}),
)
.await;
}
}
Ok(())
}
pub(super) async fn delete_targeted_transfer(
self: &Arc<Self>,
id: String,
) -> Result<(), VnidropError> {
let store = self.targeted_store();
let Some(row) = store.get_row(&id).await? else {
self.targeted_offers.discard(&id).await;
return Ok(());
};
// Durable local denial first — remote cleanup is best-effort.
let _ = self.take_active_transfer(row.protocol_transfer_id);
self.targeted_offers.discard(&id).await;
self.access_policy
.remove_transfer(row.protocol_transfer_id)
.await;
let _ = self.cancel_idle_or_share(row.protocol_transfer_id).await;
if let Some(handle) = &row.authorization_secret_handle {
if let Some(custody) = &self.secret_custody {
let _ = custody
.remove(&SecretHandle::from_stored(handle.clone()))
.await;
}
}
store.clear_authorization(&id).await?;
if row.state != TargetedTransferState::Deleted {
if !matches!(
row.state,
TargetedTransferState::Completed
| TargetedTransferState::Declined
| TargetedTransferState::Cancelled
| TargetedTransferState::Failed
) {
let _ = store
.set_state_from_any(&id, TargetedTransferState::Cancelled)
.await;
}
let _ = store
.set_state_from_any(&id, TargetedTransferState::Deleted)
.await;
}
Ok(())
}
pub(super) async fn respond_to_targeted_offer(
&self,
self: &Arc<Self>,
transfer_id: String,
accepted: bool,
) -> Result<Option<String>, VnidropError> {
if let Some(auth) = self
.targeted_offers
.settled_authorization(&transfer_id)
.await
{
return Ok(Some(auth));
}
if let Ok(Some(row)) = self.targeted_store().get_row(&transfer_id).await {
if let Some(encoded) = self.load_stored_authorization(&row).await? {
return Ok(Some(encoded));
}
}
match self.targeted_offers.respond(&transfer_id, accepted).await {
Ok(auth) => Ok(auth),
Ok(Some(auth)) => {
self.persist_receiver_authorization(&auth).await?;
Ok(Some(auth))
}
Ok(None) => Ok(None),
Err(crate::targeted_transfer::RespondError::Unknown) => Err(
VnidropError::invalid_input(anyhow::anyhow!("unknown targeted offer")),
),
@@ -134,6 +260,10 @@ impl CoreInner {
transfer_name: share.transfer_name.clone(),
file_count: share.file_count,
total_size: share.total_size,
verified_bytes: 0,
blob_ticket: None,
authorization_secret_handle: None,
role: TargetedTransferRole::Sender,
state: TargetedTransferState::Preparing,
created_at: now,
updated_at: now,
@@ -225,9 +355,9 @@ impl CoreInner {
}
}
// Bound authorization: only the approved receiver endpoint may fetch.
// Permanent until cancel/delete — approved targeted transfers must resume.
self.access_policy
.approve_endpoint(protocol_transfer_id, receiver_endpoint_id.clone())
.approve_endpoint_until(protocol_transfer_id, receiver_endpoint_id.clone(), None)
.await;
let parsed = crate::ticket::parse_transfer_ticket_with_limits(&share.ticket, &self.limits)
@@ -250,6 +380,8 @@ impl CoreInner {
transfer_name: share.transfer_name.clone(),
blob_ticket: blob_ticket.to_string(),
})?;
self.persist_authorization_secret(&transfer_uuid, &authorization)
.await?;
let encoded = authorization.encode()?;
let deliver = client
@@ -294,6 +426,86 @@ impl CoreInner {
) -> Result<(), VnidropError> {
let auth = TargetedAuthorization::decode(&authorization)?;
auth.verify_for_receiver(&self.endpoint.id().to_string())?;
self.run_targeted_receive(&auth, output_dir).await
}
pub(super) async fn resume_targeted_transfer(
self: &Arc<Self>,
id: String,
output_dir: String,
) -> 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 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())?;
self.run_targeted_receive(&auth, output_dir).await
}
async fn run_targeted_receive(
self: &Arc<Self>,
auth: &TargetedAuthorization,
output_dir: String,
) -> Result<(), VnidropError> {
let store = self.targeted_store();
if let Ok(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?;
store
.set_state(
&auth.transfer_id,
TargetedTransferState::Connecting,
TargetedTransferState::Transferring,
)
.await?;
}
TargetedTransferState::Connecting => {
store
.set_state(
&auth.transfer_id,
TargetedTransferState::Connecting,
TargetedTransferState::Transferring,
)
.await?;
}
TargetedTransferState::Transferring => {}
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)))?;
@@ -310,22 +522,133 @@ impl CoreInner {
.encode()
.map_err(VnidropError::ticket)?;
self.receive(ticket, std::path::PathBuf::from(output_dir), None)
.await
.map_err(VnidropError::transfer)?;
let receive_result = self
.receive(ticket, std::path::PathBuf::from(output_dir), None)
.await;
if let Ok(Some(row)) = self.targeted_store().get_row(&auth.transfer_id).await {
let _ = self
.targeted_store()
match receive_result {
Ok(()) => {
if let Ok(Some(row)) = store.get_row(&auth.transfer_id).await {
let _ = store
.set_verified_bytes(&auth.transfer_id, row.total_size)
.await;
let _ = store
.set_state(
&auth.transfer_id,
row.state,
TargetedTransferState::Transferring,
TargetedTransferState::Completed,
)
.await;
}
Ok(())
}
Err(error) => {
if let Ok(Some(row)) = store.get_row(&auth.transfer_id).await {
if matches!(
row.state,
TargetedTransferState::Connecting | TargetedTransferState::Transferring
) {
let _ = store
.set_state_from_any(
&auth.transfer_id,
TargetedTransferState::Interrupted,
)
.await;
}
}
Err(VnidropError::transfer(error))
}
}
}
async fn persist_authorization_secret(
&self,
transfer_id: &str,
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 material = auth_secret_material(authorization)?;
let handle = custody
.protect(SecretKind::TargetedAuthorization, material, None)
.await?;
self.targeted_store()
.store_authorization(transfer_id, &authorization.blob_ticket, handle.as_str())
.await
}
async fn persist_receiver_authorization(&self, encoded: &str) -> Result<(), VnidropError> {
let auth = TargetedAuthorization::decode(encoded)?;
let store = self.targeted_store();
if store.get_row(&auth.transfer_id).await?.is_none() {
let now = now_ms();
store
.insert(&TargetedTransferRow {
id: auth.transfer_id.clone(),
protocol_transfer_id: auth.protocol_transfer_id,
sender_endpoint_id: auth.sender_endpoint_id.clone(),
receiver_endpoint_id: auth.receiver_endpoint_id.clone(),
manifest_id: auth.manifest_id.clone(),
content_hash: auth.content_hash.clone(),
transfer_name: auth.transfer_name.clone(),
file_count: auth.file_count,
total_size: auth.total_size,
verified_bytes: 0,
blob_ticket: Some(auth.blob_ticket.clone()),
authorization_secret_handle: None,
role: TargetedTransferRole::Receiver,
state: TargetedTransferState::Approved,
created_at: now,
updated_at: now,
})
.await?;
}
self.persist_authorization_secret(&auth.transfer_id, &auth)
.await
}
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: experimental_saved_device_capabilities()
.targeted_transfer_protocol_version,
transfer_name: row.transfer_name.clone(),
blob_ticket: blob_ticket.clone(),
},
&material,
)?;
Ok(Some(auth.encode()?))
}
}
fn allocate_protocol_transfer_id(transfer_uuid: &str) -> u64 {

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

@@ -108,6 +108,7 @@ pub(crate) enum SecretKind {
EndpointIdentity,
RelationshipGrant,
PairingEligibility,
TargetedAuthorization,
}
impl SecretKind {
@@ -116,6 +117,7 @@ impl SecretKind {
Self::EndpointIdentity => "endpoint-identity",
Self::RelationshipGrant => "relationship-grant",
Self::PairingEligibility => "pairing-eligibility",
Self::TargetedAuthorization => "targeted-authorization",
}
}
@@ -124,6 +126,7 @@ impl SecretKind {
"endpoint-identity" => Ok(Self::EndpointIdentity),
"relationship-grant" => Ok(Self::RelationshipGrant),
"pairing-eligibility" => Ok(Self::PairingEligibility),
"targeted-authorization" => Ok(Self::TargetedAuthorization),
_ => Err(VnidropError::SecureStorageCorrupted {
reason: "protected secret has an unknown kind".to_string(),
}),

42
crates/vnidrop/src/targeted_transfer/auth.rs
View File

@@ -9,7 +9,7 @@
use data_encoding::{BASE64URL_NOPAD, HEXLOWER};
use serde::{Deserialize, Serialize};
use crate::error::VnidropError;
use crate::{error::VnidropError, secure_secret::SecretMaterial};
const AUTH_CONTEXT: &[u8] = b"vnidrop-targeted-auth-v1";
@@ -137,4 +137,44 @@ impl TargetedAuthorization {
hasher.update(&self.protocol_version.to_le_bytes());
Ok(*hasher.finalize().as_bytes())
}
pub(crate) fn secret_bytes(&self) -> Result<[u8; 32], VnidropError> {
let secret_bytes = HEXLOWER.decode(self.auth_secret.as_bytes()).map_err(|_| {
VnidropError::invalid_input(anyhow::anyhow!("invalid authorization secret"))
})?;
secret_bytes.try_into().map_err(|_| {
VnidropError::invalid_input(anyhow::anyhow!("invalid authorization secret length"))
})
}
}
pub(crate) fn auth_secret_material(
auth: &TargetedAuthorization,
) -> Result<SecretMaterial, VnidropError> {
SecretMaterial::new(auth.secret_bytes()?.to_vec())
}
/// Rebuild a bound authorization from durable row fields + custody secret.
pub(crate) fn reconstruct_authorization(
draft: TargetedAuthorizationDraft,
secret_material: &SecretMaterial,
) -> Result<TargetedAuthorization, VnidropError> {
let auth_secret = HEXLOWER.encode(secret_material.as_bytes());
let mut auth = TargetedAuthorization {
transfer_id: draft.transfer_id,
protocol_transfer_id: draft.protocol_transfer_id,
sender_endpoint_id: draft.sender_endpoint_id,
receiver_endpoint_id: draft.receiver_endpoint_id,
manifest_id: draft.manifest_id,
content_hash: draft.content_hash,
file_count: draft.file_count,
total_size: draft.total_size,
protocol_version: draft.protocol_version,
transfer_name: draft.transfer_name,
blob_ticket: draft.blob_ticket,
auth_secret,
mac: String::new(),
};
auth.mac = HEXLOWER.encode(&auth.compute_mac()?);
Ok(auth)
}

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

@@ -2,11 +2,12 @@
//!
//! Offers are not durable: cancellation, timeout, disconnect, or restart drops
//! them. Authorization is delivered only after the local user accepts.
//! Settled results are cached briefly so lost-response replays stay idempotent.
use std::{collections::HashMap, sync::Arc, time::Duration};
use serde_json::json;
use tokio::sync::{oneshot, Mutex};
use tokio::sync::{watch, Mutex};
use uuid::Uuid;
use crate::{api::PendingTargetedOffer, event_hub::EventHub};
@@ -18,20 +19,27 @@ pub(crate) struct PendingTargetedOfferRecord {
pub(crate) offer: PendingTargetedOffer,
}
struct DecisionWaiter {
decision: oneshot::Sender<bool>,
struct PendingWaiter {
decision: watch::Sender<Option<bool>>,
}
struct AuthWaiter {
auth: oneshot::Sender<String>,
auth: watch::Sender<Option<String>>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum SettledOfferResult {
Accepted { authorization: Option<String> },
Declined { reason: String },
}
#[derive(Clone)]
pub(crate) struct TargetedOfferInbox {
event_hub: Arc<EventHub>,
pending: Arc<Mutex<HashMap<String, PendingTargetedOfferRecord>>>,
decisions: Arc<Mutex<HashMap<String, DecisionWaiter>>>,
decisions: Arc<Mutex<HashMap<String, PendingWaiter>>>,
auths: Arc<Mutex<HashMap<String, AuthWaiter>>>,
settled: Arc<Mutex<HashMap<String, SettledOfferResult>>>,
max_pending: usize,
}
@@ -42,16 +50,32 @@ impl TargetedOfferInbox {
pending: Arc::new(Mutex::new(HashMap::new())),
decisions: Arc::new(Mutex::new(HashMap::new())),
auths: Arc::new(Mutex::new(HashMap::new())),
settled: Arc::new(Mutex::new(HashMap::new())),
max_pending,
}
}
/// Surface a validated offer and block until the local user decides.
///
/// Replaying the same transfer identity returns the settled result or joins
/// the existing pending wait — never a second prompt.
pub(crate) async fn submit(&self, offer: PendingTargetedOffer) -> TargetedOfferDecision {
let transfer_id = offer.transfer_id.clone();
let (decision_tx, decision_rx) = oneshot::channel();
if let Some(settled) = self.settled.lock().await.get(&transfer_id).cloned() {
return settled_to_decision(settled);
}
{
let mut pending = self.pending.lock().await;
let pending = self.pending.lock().await;
if let Some(existing) = pending.get(&transfer_id) {
if offers_equivalent(&existing.offer, &offer) {
drop(pending);
return self.wait_existing_decision(&transfer_id).await;
}
return TargetedOfferDecision::Refused {
reason: "immutable-transfer-mismatch".to_string(),
};
}
if pending.len() >= self.max_pending {
return TargetedOfferDecision::Refused {
reason: "too-many-pending-offers".to_string(),
@@ -65,10 +89,15 @@ impl TargetedOfferInbox {
reason: "offer-already-pending".to_string(),
};
}
}
let (decision_tx, _decision_rx) = watch::channel(None);
{
let mut pending = self.pending.lock().await;
if pending.contains_key(&transfer_id) {
return TargetedOfferDecision::Refused {
reason: "duplicate-transfer".to_string(),
};
// Lost the race with another submit of the same id.
drop(pending);
return self.wait_existing_decision(&transfer_id).await;
}
pending.insert(
transfer_id.clone(),
@@ -79,7 +108,7 @@ impl TargetedOfferInbox {
}
self.decisions.lock().await.insert(
transfer_id.clone(),
DecisionWaiter {
PendingWaiter {
decision: decision_tx,
},
);
@@ -96,16 +125,44 @@ impl TargetedOfferInbox {
}),
);
match tokio::time::timeout(OFFER_WAIT_TIMEOUT, decision_rx).await {
Ok(Ok(true)) => TargetedOfferDecision::Accepted,
Ok(Ok(false)) => {
self.discard(&transfer_id).await;
self.wait_existing_decision(&transfer_id).await
}
async fn wait_existing_decision(&self, transfer_id: &str) -> TargetedOfferDecision {
let mut rx = {
let decisions = self.decisions.lock().await;
let Some(waiter) = decisions.get(transfer_id) else {
if let Some(settled) = self.settled.lock().await.get(transfer_id).cloned() {
return settled_to_decision(settled);
}
return TargetedOfferDecision::Declined {
reason: "no-response".to_string(),
};
};
waiter.decision.subscribe()
};
let wait = async {
loop {
if let Some(accepted) = *rx.borrow_and_update() {
return accepted;
}
if rx.changed().await.is_err() {
return false;
}
}
};
match tokio::time::timeout(OFFER_WAIT_TIMEOUT, wait).await {
Ok(true) => TargetedOfferDecision::Accepted,
Ok(false) => {
self.discard(transfer_id).await;
TargetedOfferDecision::Declined {
reason: "receiver-declined".to_string(),
}
}
Ok(Err(_)) | Err(_) => {
self.discard(&transfer_id).await;
Err(_) => {
self.discard(transfer_id).await;
TargetedOfferDecision::Declined {
reason: "no-response".to_string(),
}
@@ -122,23 +179,55 @@ impl TargetedOfferInbox {
.collect()
}
pub(crate) async fn get_pending(&self, transfer_id: &str) -> Option<PendingTargetedOffer> {
self.pending
.lock()
.await
.get(transfer_id)
.map(|entry| entry.offer.clone())
}
pub(crate) async fn settled_authorization(&self, transfer_id: &str) -> Option<String> {
match self.settled.lock().await.get(transfer_id) {
Some(SettledOfferResult::Accepted {
authorization: Some(auth),
}) => Some(auth.clone()),
_ => None,
}
}
/// Record the local decision. On accept, wait for sender-issued authorization.
pub(crate) async fn respond(
&self,
transfer_id: &str,
accepted: bool,
) -> Result<Option<String>, RespondError> {
if let Some(auth) = self.settled_authorization(transfer_id).await {
return Ok(Some(auth));
}
let exists = self.pending.lock().await.contains_key(transfer_id);
if !exists {
return Err(RespondError::Unknown);
}
let waiter = self.decisions.lock().await.remove(transfer_id);
let Some(waiter) = waiter else {
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 _ = waiter.decision.send(false);
let _ = decision_tx.send(Some(false));
self.discard(transfer_id).await;
self.settled.lock().await.insert(
transfer_id.to_string(),
SettledOfferResult::Declined {
reason: "receiver-declined".to_string(),
},
);
self.event_hub.emit_endpoint(
"targeted_transfer",
"offer-declined",
@@ -147,20 +236,38 @@ impl TargetedOfferInbox {
return Ok(None);
}
let (auth_tx, auth_rx) = oneshot::channel();
let (auth_tx, mut auth_rx) = watch::channel(None);
self.auths
.lock()
.await
.insert(transfer_id.to_string(), AuthWaiter { auth: auth_tx });
if waiter.decision.send(true).is_err() {
if decision_tx.send(Some(true)).is_err() {
self.auths.lock().await.remove(transfer_id);
self.discard(transfer_id).await;
return Err(RespondError::SenderGone);
}
match tokio::time::timeout(OFFER_WAIT_TIMEOUT, auth_rx).await {
let wait_auth = async {
loop {
if let Some(auth) = auth_rx.borrow_and_update().clone() {
return Ok(auth);
}
if auth_rx.changed().await.is_err() {
return Err(());
}
}
};
match tokio::time::timeout(OFFER_WAIT_TIMEOUT, wait_auth).await {
Ok(Ok(auth)) => {
self.pending.lock().await.remove(transfer_id);
self.auths.lock().await.remove(transfer_id);
self.settled.lock().await.insert(
transfer_id.to_string(),
SettledOfferResult::Accepted {
authorization: Some(auth.clone()),
},
);
self.event_hub.emit_endpoint(
"targeted_transfer",
"offer-accepted",
@@ -168,7 +275,7 @@ impl TargetedOfferInbox {
);
Ok(Some(auth))
}
Ok(Err(_)) | Err(_) => {
Ok(Err(())) | Err(_) => {
self.discard(transfer_id).await;
Err(RespondError::AuthorizationTimeout)
}
@@ -180,8 +287,14 @@ impl TargetedOfferInbox {
transfer_id: &str,
authorization: String,
) -> bool {
if let Some(waiter) = self.auths.lock().await.remove(transfer_id) {
waiter.auth.send(authorization).is_ok()
if let Some(SettledOfferResult::Accepted {
authorization: Some(existing),
}) = self.settled.lock().await.get(transfer_id)
{
return existing == &authorization;
}
if let Some(waiter) = self.auths.lock().await.get(transfer_id) {
waiter.auth.send(Some(authorization)).is_ok()
} else {
false
}
@@ -201,15 +314,33 @@ impl TargetedOfferInbox {
}
}
async fn discard(&self, transfer_id: &str) {
pub(crate) async fn discard(&self, transfer_id: &str) {
self.pending.lock().await.remove(transfer_id);
if let Some(waiter) = self.decisions.lock().await.remove(transfer_id) {
let _ = waiter.decision.send(false);
let _ = waiter.decision.send(Some(false));
}
self.auths.lock().await.remove(transfer_id);
}
}
fn offers_equivalent(left: &PendingTargetedOffer, right: &PendingTargetedOffer) -> bool {
left.transfer_id == right.transfer_id
&& left.sender_endpoint_id == right.sender_endpoint_id
&& left.receiver_endpoint_id == right.receiver_endpoint_id
&& left.manifest_id == right.manifest_id
&& left.content_hash == right.content_hash
&& left.file_count == right.file_count
&& left.total_size == right.total_size
&& left.protocol_version == right.protocol_version
}
fn settled_to_decision(settled: SettledOfferResult) -> TargetedOfferDecision {
match settled {
SettledOfferResult::Accepted { .. } => TargetedOfferDecision::Accepted,
SettledOfferResult::Declined { reason } => TargetedOfferDecision::Declined { reason },
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum TargetedOfferDecision {
Accepted,

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

@@ -8,7 +8,10 @@ pub(crate) mod inbox;
pub(crate) mod protocol;
mod state;
pub(crate) use auth::{TargetedAuthorization, TargetedAuthorizationDraft};
pub(crate) use auth::{
auth_secret_material, reconstruct_authorization, TargetedAuthorization,
TargetedAuthorizationDraft,
};
pub(crate) use inbox::{RespondError, TargetedOfferInbox};
pub(crate) use protocol::TargetedTransferProtocol;
@@ -33,6 +36,10 @@ pub(crate) async fn ensure_schema(pool: &SqlitePool) -> anyhow::Result<()> {
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
@@ -41,9 +48,38 @@ pub(crate) async fn ensure_schema(pool: &SqlitePool) -> anyhow::Result<()> {
)
.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,
}
@@ -59,8 +95,9 @@ impl TargetedTransferStore {
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, ?10, ?11, ?12)
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16)
"#,
)
.bind(&transfer.id)
@@ -72,6 +109,10 @@ impl TargetedTransferStore {
.bind(&transfer.transfer_name)
.bind(transfer.file_count as i64)
.bind(transfer.total_size as i64)
.bind(transfer.verified_bytes as i64)
.bind(&transfer.blob_ticket)
.bind(&transfer.authorization_secret_handle)
.bind(role_as_str(transfer.role))
.bind(state_as_str(transfer.state))
.bind(transfer.created_at)
.bind(transfer.updated_at)
@@ -110,11 +151,113 @@ impl TargetedTransferStore {
Ok(())
}
/// Transition from any non-terminal state; used by cancel/delete.
pub(crate) async fn set_state_from_any(
&self,
id: &str,
to: TargetedTransferState,
) -> Result<(), VnidropError> {
let Some(row) = self.get_row(id).await? else {
return Err(VnidropError::invalid_input(anyhow::anyhow!(
"unknown targeted transfer"
)));
};
if row.state == to {
return Ok(());
}
row.state.validate_transition_to(to)?;
let result = sqlx::query(
r#"
UPDATE targeted_transfers
SET state = ?2, updated_at = ?3
WHERE id = ?1 AND state = ?4
"#,
)
.bind(id)
.bind(state_as_str(to))
.bind(now_ms())
.bind(state_as_str(row.state))
.execute(&self.pool)
.await
.map_err(VnidropError::repository)?;
if result.rows_affected() == 0 {
return Err(VnidropError::InvalidTransition {
reason: format!("{} -> {}", state_as_str(row.state), state_as_str(to)),
});
}
Ok(())
}
pub(crate) async fn set_verified_bytes(
&self,
id: &str,
verified_bytes: u64,
) -> Result<(), VnidropError> {
sqlx::query(
r#"
UPDATE targeted_transfers
SET verified_bytes = ?2, updated_at = ?3
WHERE id = ?1
"#,
)
.bind(id)
.bind(verified_bytes as i64)
.bind(now_ms())
.execute(&self.pool)
.await
.map_err(VnidropError::repository)?;
Ok(())
}
pub(crate) async fn store_authorization(
&self,
id: &str,
blob_ticket: &str,
authorization_secret_handle: &str,
) -> Result<(), VnidropError> {
sqlx::query(
r#"
UPDATE targeted_transfers
SET blob_ticket = ?2,
authorization_secret_handle = ?3,
updated_at = ?4
WHERE id = ?1
"#,
)
.bind(id)
.bind(blob_ticket)
.bind(authorization_secret_handle)
.bind(now_ms())
.execute(&self.pool)
.await
.map_err(VnidropError::repository)?;
Ok(())
}
pub(crate) async fn clear_authorization(&self, id: &str) -> Result<(), VnidropError> {
sqlx::query(
r#"
UPDATE targeted_transfers
SET blob_ticket = NULL,
authorization_secret_handle = NULL,
verified_bytes = 0,
updated_at = ?2
WHERE id = ?1
"#,
)
.bind(id)
.bind(now_ms())
.execute(&self.pool)
.await
.map_err(VnidropError::repository)?;
Ok(())
}
pub(crate) async fn get(&self, id: &str) -> Result<Option<TargetedTransfer>, VnidropError> {
let row = sqlx::query(
r#"
SELECT id, sender_endpoint_id, receiver_endpoint_id, manifest_id,
file_count, total_size, state, created_at, updated_at
file_count, total_size, verified_bytes, state, created_at, updated_at
FROM targeted_transfers WHERE id = ?1
"#,
)
@@ -133,6 +276,7 @@ impl TargetedTransferStore {
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 id = ?1
"#,
@@ -148,7 +292,7 @@ impl TargetedTransferStore {
let rows = sqlx::query(
r#"
SELECT id, sender_endpoint_id, receiver_endpoint_id, manifest_id,
file_count, total_size, state, created_at, updated_at
file_count, total_size, verified_bytes, state, created_at, updated_at
FROM targeted_transfers
ORDER BY updated_at DESC
"#,
@@ -159,6 +303,26 @@ impl TargetedTransferStore {
rows.into_iter().map(row_to_transfer).collect()
}
pub(crate) async fn list_resumable_sender_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 role = 'sender'
AND state IN ('approved', 'connecting', 'transferring', 'interrupted')
"#,
)
.fetch_all(&self.pool)
.await
.map_err(VnidropError::repository)?;
rows.into_iter().map(row_to_full).collect()
}
pub(crate) async fn cancel_by_peer(&self, peer_endpoint_id: &str) -> Result<u64, VnidropError> {
let now = now_ms();
let result = sqlx::query(
@@ -196,6 +360,28 @@ impl TargetedTransferStore {
.map(|row| row.get::<i64, _>(0) as u64)
.collect())
}
pub(crate) async fn mark_interrupted_in_flight(&self) -> Result<u64, VnidropError> {
let now = now_ms();
let result = sqlx::query(
r#"
UPDATE targeted_transfers
SET state = 'interrupted', updated_at = ?1
WHERE state IN ('connecting', 'transferring')
"#,
)
.bind(now)
.execute(&self.pool)
.await
.map_err(VnidropError::repository)?;
Ok(result.rows_affected())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum TargetedTransferRole {
Sender,
Receiver,
}
#[derive(Debug, Clone)]
@@ -209,6 +395,10 @@ pub(crate) struct TargetedTransferRow {
pub(crate) transfer_name: String,
pub(crate) file_count: u64,
pub(crate) total_size: u64,
pub(crate) verified_bytes: u64,
pub(crate) blob_ticket: Option<String>,
pub(crate) authorization_secret_handle: Option<String>,
pub(crate) role: TargetedTransferRole,
pub(crate) state: TargetedTransferState,
pub(crate) created_at: i64,
pub(crate) updated_at: i64,
@@ -222,6 +412,7 @@ fn row_to_transfer(row: sqlx::sqlite::SqliteRow) -> Result<TargetedTransfer, Vni
manifest_id: row.get("manifest_id"),
file_count: row.get::<i64, _>("file_count") as u64,
total_size: row.get::<i64, _>("total_size") as u64,
verified_bytes: row.try_get::<i64, _>("verified_bytes").unwrap_or(0) as u64,
state: parse_state(&row.get::<String, _>("state"))?,
created_at: row.get("created_at"),
updated_at: row.get("updated_at"),
@@ -239,6 +430,13 @@ fn row_to_full(row: sqlx::sqlite::SqliteRow) -> Result<TargetedTransferRow, Vnid
transfer_name: row.get("transfer_name"),
file_count: row.get::<i64, _>("file_count") as u64,
total_size: row.get::<i64, _>("total_size") as u64,
verified_bytes: row.try_get::<i64, _>("verified_bytes").unwrap_or(0) as u64,
blob_ticket: row.try_get("blob_ticket").ok().flatten(),
authorization_secret_handle: row.try_get("authorization_secret_handle").ok().flatten(),
role: parse_role(
&row.try_get::<String, _>("role")
.unwrap_or_else(|_| "sender".to_string()),
)?,
state: parse_state(&row.get::<String, _>("state"))?,
created_at: row.get("created_at"),
updated_at: row.get("updated_at"),
@@ -262,6 +460,23 @@ pub(crate) fn state_as_str(state: TargetedTransferState) -> &'static str {
}
}
fn role_as_str(role: TargetedTransferRole) -> &'static str {
match role {
TargetedTransferRole::Sender => "sender",
TargetedTransferRole::Receiver => "receiver",
}
}
fn parse_role(value: &str) -> Result<TargetedTransferRole, VnidropError> {
match value {
"sender" => Ok(TargetedTransferRole::Sender),
"receiver" => Ok(TargetedTransferRole::Receiver),
other => Err(VnidropError::repository(anyhow::anyhow!(
"unknown targeted transfer role: {other}"
))),
}
}
fn parse_state(value: &str) -> Result<TargetedTransferState, VnidropError> {
match value {
"preparing" => Ok(TargetedTransferState::Preparing),

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

@@ -14,13 +14,15 @@ use iroh::{
use irpc::{channel::oneshot, rpc_requests, Client, WithChannels};
use irpc_iroh::{read_request, IrohLazyRemoteConnection};
use serde::{Deserialize, Serialize};
use sqlx::SqlitePool;
use super::{
auth::TargetedAuthorization,
inbox::{TargetedOfferDecision, TargetedOfferInbox},
state_as_str, TargetedTransferStore,
};
use crate::{
api::{experimental_saved_device_capabilities, PendingTargetedOffer},
api::{experimental_saved_device_capabilities, PendingTargetedOffer, TargetedTransferState},
device_relationship::{DeviceRelationshipService, WireProof},
error::VnidropError,
grant::Challenge,
@@ -31,6 +33,7 @@ use crate::{
pub(crate) struct TargetedTransferProtocol {
relationships: std::sync::Arc<DeviceRelationshipService>,
inbox: TargetedOfferInbox,
store: TargetedTransferStore,
limits: crate::api::CoreLimits,
local_endpoint_id: String,
}
@@ -47,12 +50,14 @@ impl TargetedTransferProtocol {
pub(crate) fn new(
relationships: std::sync::Arc<DeviceRelationshipService>,
inbox: TargetedOfferInbox,
pool: SqlitePool,
limits: crate::api::CoreLimits,
local_endpoint_id: String,
) -> Self {
Self {
relationships,
inbox,
store: TargetedTransferStore::new(pool),
limits,
local_endpoint_id,
}
@@ -111,6 +116,44 @@ impl TargetedTransferProtocol {
};
}
if let Ok(Some(existing)) = self.store.get_row(&offer.transfer_id).await {
if existing.manifest_id != offer.manifest_id
|| existing.content_hash != offer.content_hash
|| existing.file_count != offer.file_count
|| existing.total_size != offer.total_size
|| existing.sender_endpoint_id != offer.sender_endpoint_id
|| existing.receiver_endpoint_id != offer.receiver_endpoint_id
{
return TargetedOfferResponse::Refused {
reason: "immutable-transfer-mismatch".to_string(),
};
}
return match existing.state {
TargetedTransferState::Approved
| TargetedTransferState::Connecting
| TargetedTransferState::Transferring
| TargetedTransferState::Interrupted
| TargetedTransferState::Completed => TargetedOfferResponse::Accepted,
TargetedTransferState::Declined => TargetedOfferResponse::Declined {
reason: "receiver-declined".to_string(),
},
TargetedTransferState::Cancelled => TargetedOfferResponse::Declined {
reason: "cancelled".to_string(),
},
TargetedTransferState::Failed | TargetedTransferState::Deleted => {
TargetedOfferResponse::Refused {
reason: format!("transfer-{}", state_as_str(existing.state)),
}
}
TargetedTransferState::Preparing
| TargetedTransferState::Offering
| TargetedTransferState::AwaitingApproval => {
// Live offer path below may still be pending.
TargetedOfferResponse::Accepted
}
};
}
if let Err(error) = self
.relationships
.verify_saved_possession(
@@ -164,6 +207,14 @@ 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.manifest_id == auth.manifest_id
&& row.content_hash == auth.content_hash
{
return DeliverAuthorizationResponse::Stored;
}
}
if self
.inbox
.deliver_authorization(&delivery.transfer_id, delivery.authorization)
@@ -174,6 +225,28 @@ impl TargetedTransferProtocol {
DeliverAuthorizationResponse::Rejected
}
}
async fn handle_cancel(
&self,
remote_endpoint_id: &str,
cancel: CancelTargetedOffer,
) -> CancelTargetedOfferResponse {
if let Some(pending) = self.inbox.get_pending(&cancel.transfer_id).await {
if pending.sender_endpoint_id != remote_endpoint_id {
return CancelTargetedOfferResponse::Rejected;
}
self.inbox.discard(&cancel.transfer_id).await;
return CancelTargetedOfferResponse::Cancelled;
}
if let Ok(Some(row)) = self.store.get_row(&cancel.transfer_id).await {
if row.sender_endpoint_id != remote_endpoint_id {
return CancelTargetedOfferResponse::Rejected;
}
// Already gone from the live inbox; treat as idempotent success.
return CancelTargetedOfferResponse::Cancelled;
}
CancelTargetedOfferResponse::Cancelled
}
}
impl ProtocolHandler for TargetedTransferProtocol {
@@ -204,6 +277,11 @@ impl ProtocolHandler for TargetedTransferProtocol {
.await;
let _ = tx.send(response).await;
}
TargetedTransferMessage::CancelTargetedOffer(message) => {
let WithChannels { inner, tx, .. } = message;
let response = self.handle_cancel(&remote_endpoint_id, inner).await;
let _ = tx.send(response).await;
}
}
}
connection.closed().await;
@@ -234,6 +312,13 @@ impl TargetedTransferClient {
) -> Result<DeliverAuthorizationResponse, irpc::Error> {
self.inner.rpc(delivery).await
}
pub(crate) async fn cancel_offer(
&self,
cancel: CancelTargetedOffer,
) -> Result<CancelTargetedOfferResponse, irpc::Error> {
self.inner.rpc(cancel).await
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -279,6 +364,17 @@ pub(crate) enum DeliverAuthorizationResponse {
Rejected,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct CancelTargetedOffer {
pub(crate) transfer_id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub(crate) enum CancelTargetedOfferResponse {
Cancelled,
Rejected,
}
#[rpc_requests(message = TargetedTransferMessage)]
#[derive(Debug, Serialize, Deserialize)]
enum TargetedTransferMessages {
@@ -288,6 +384,8 @@ enum TargetedTransferMessages {
SubmitTargetedOffer(SubmitTargetedOffer),
#[rpc(tx = oneshot::Sender<DeliverAuthorizationResponse>)]
DeliverTargetedAuthorization(DeliverTargetedAuthorization),
#[rpc(tx = oneshot::Sender<CancelTargetedOfferResponse>)]
CancelTargetedOffer(CancelTargetedOffer),
}
/// Helper kept for type visibility in callers that map refuse reasons.

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

@@ -16,7 +16,7 @@ impl TargetedTransferState {
Self::Approved | Self::Declined | Self::Cancelled | Self::Failed
) | (
Self::Approved,
Self::Connecting | Self::Cancelled | Self::Failed
Self::Connecting | Self::Cancelled | Self::Failed | Self::Deleted
) | (
Self::Connecting,
Self::Transferring | Self::Interrupted | Self::Cancelled | Self::Failed

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

@@ -21,8 +21,9 @@ impl CoreEventSink for RecordingSink {
}
struct ProtectedNode {
_data_dir: tempfile::TempDir,
core: Arc<VnidropCore>,
data_dir: tempfile::TempDir,
secret_store: Arc<FaultInjectingSecretStore>,
core: Option<Arc<VnidropCore>>,
}
impl ProtectedNode {
@@ -35,19 +36,43 @@ impl ProtectedNode {
let core = VnidropCore::initialize_with_test_secret_store(
data_dir.path().to_string_lossy().into_owned(),
sink,
store,
store.clone(),
)
.expect("protected test core");
Self {
_data_dir: data_dir,
core,
data_dir,
secret_store: store,
core: Some(core),
}
}
fn core(&self) -> Arc<VnidropCore> {
self.core.as_ref().expect("core alive").clone()
}
fn restart(mut self) -> Self {
if let Some(core) = self.core.take() {
core.shutdown();
}
let sink = Arc::new(RecordingSink {
events: Mutex::new(Vec::new()),
});
let core = VnidropCore::initialize_with_test_secret_store(
self.data_dir.path().to_string_lossy().into_owned(),
sink,
self.secret_store.clone(),
)
.expect("restarted protected test core");
self.core = Some(core);
self
}
}
impl Drop for ProtectedNode {
fn drop(&mut self) {
self.core.shutdown();
if let Some(core) = self.core.take() {
core.shutdown();
}
}
}
@@ -93,25 +118,25 @@ fn complete_transfer(sender: &ProtectedNode, receiver: &ProtectedNode, transfer_
let output_dir = tempfile::tempdir().unwrap();
let source_path = source_dir.path().join("hello.txt");
std::fs::write(&source_path, b"mutual consent").unwrap();
let share = share_path(&sender.core, &source_path, transfer_id);
let share = share_path(&sender.core(), &source_path, transfer_id);
let output_dir = output_dir.path().to_string_lossy().to_string();
let receiver_core = receiver.core.clone();
let receiver_core = receiver.core().clone();
let ticket = share.ticket.clone();
let handle = std::thread::spawn(move || {
receiver_core.receive(ticket, output_dir, Some("receiver".to_string()))
});
let request = wait_for_receiver_request(&sender.core, share.transfer_id);
let request = wait_for_receiver_request(&sender.core(), share.transfer_id);
sender
.core
.core()
.respond_receiver_request(request.id, true, None)
.unwrap();
handle.join().unwrap().unwrap();
let started = Instant::now();
let peer = receiver.core.status().endpoint_id.clone();
let peer = receiver.core().status().endpoint_id.clone();
loop {
if sender
.core
.core()
.list_pairing_eligibilities()
.unwrap()
.iter()
@@ -151,24 +176,24 @@ fn wait_for_relationship(
}
fn establish_saved(alice: &ProtectedNode, bob: &ProtectedNode, transfer_id: u64) {
let alice_id = alice.core.status().endpoint_id.clone();
let bob_id = bob.core.status().endpoint_id.clone();
let alice_id = alice.core().status().endpoint_id.clone();
let bob_id = bob.core().status().endpoint_id.clone();
complete_transfer(alice, bob, transfer_id);
assert!(alice
.core
.core()
.request_saved_device_pairing(bob_id.clone())
.unwrap());
wait_for_relationship(
&bob.core,
&bob.core(),
&alice_id,
DeviceRelationshipState::PendingIncoming,
);
assert!(bob
.core
.core()
.respond_to_device_pairing(alice_id.clone(), true)
.unwrap());
wait_for_relationship(&alice.core, &bob_id, DeviceRelationshipState::Saved);
wait_for_relationship(&bob.core, &alice_id, DeviceRelationshipState::Saved);
wait_for_relationship(&alice.core(), &bob_id, DeviceRelationshipState::Saved);
wait_for_relationship(&bob.core(), &alice_id, DeviceRelationshipState::Saved);
}
fn wait_for_pending_offer(core: &VnidropCore) -> PendingTargetedOffer {
@@ -190,7 +215,12 @@ fn targeted_source(path: &Path) -> ShareSource {
ShareSource {
kind: SourceKind::Path,
value: path.to_string_lossy().into_owned(),
display_name: Some("payload.txt".to_string()),
display_name: Some(
path.file_name()
.unwrap_or_default()
.to_string_lossy()
.into_owned(),
),
is_directory: false,
}
}
@@ -200,8 +230,8 @@ fn create_targeted_transfer_is_immutable_and_saved_only() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
let stranger = ProtectedNode::new();
let bob_id = bob.core.status().endpoint_id.clone();
let stranger_id = stranger.core.status().endpoint_id.clone();
let bob_id = bob.core().status().endpoint_id.clone();
let stranger_id = stranger.core().status().endpoint_id.clone();
establish_saved(&alice, &bob, 10_001);
let source_dir = tempfile::tempdir().unwrap();
@@ -209,7 +239,7 @@ fn create_targeted_transfer_is_immutable_and_saved_only() {
std::fs::write(&source_path, b"immutable payload").unwrap();
let stranger_err = alice
.core
.core()
.create_targeted_transfer(
stranger_id,
vec![targeted_source(&source_path)],
@@ -218,7 +248,7 @@ fn create_targeted_transfer_is_immutable_and_saved_only() {
.unwrap_err();
assert!(matches!(stranger_err, VnidropError::Permission { .. }));
let bob_core = bob.core.clone();
let bob_core = bob.core().clone();
let accept = std::thread::spawn(move || {
let offer = wait_for_pending_offer(&bob_core);
bob_core
@@ -227,7 +257,7 @@ fn create_targeted_transfer_is_immutable_and_saved_only() {
});
let transfer = alice
.core
.core()
.create_targeted_transfer(
bob_id.clone(),
vec![targeted_source(&source_path)],
@@ -236,7 +266,10 @@ fn create_targeted_transfer_is_immutable_and_saved_only() {
.unwrap();
let _auth = accept.join().unwrap().expect("authorization after approve");
assert_eq!(transfer.sender_endpoint_id, alice.core.status().endpoint_id);
assert_eq!(
transfer.sender_endpoint_id,
alice.core().status().endpoint_id
);
assert_eq!(transfer.receiver_endpoint_id, bob_id);
assert_eq!(transfer.file_count, 1);
assert_eq!(transfer.total_size, b"immutable payload".len() as u64);
@@ -251,7 +284,7 @@ fn create_targeted_transfer_is_immutable_and_saved_only() {
));
let listed = alice
.core
.core()
.get_targeted_transfer(transfer.id.clone())
.unwrap();
let listed = listed.expect("durable targeted transfer");
@@ -267,14 +300,14 @@ fn create_targeted_transfer_is_immutable_and_saved_only() {
fn preapproval_offer_is_authenticated_without_ordinary_share_ticket() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
let bob_id = bob.core.status().endpoint_id.clone();
let bob_id = bob.core().status().endpoint_id.clone();
establish_saved(&alice, &bob, 10_010);
let source_dir = tempfile::tempdir().unwrap();
let source_path = source_dir.path().join("payload.txt");
std::fs::write(&source_path, b"offer body").unwrap();
let bob_core = bob.core.clone();
let bob_core = bob.core().clone();
let accept = std::thread::spawn(move || {
let offer = wait_for_pending_offer(&bob_core);
// Offer surfaces identity + manifest summary only — never a reusable ticket.
@@ -291,7 +324,7 @@ fn preapproval_offer_is_authenticated_without_ordinary_share_ticket() {
});
alice
.core
.core()
.create_targeted_transfer(
bob_id,
vec![targeted_source(&source_path)],
@@ -315,13 +348,13 @@ fn invalid_offer_never_becomes_observable_pending_approval() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
// No Saved relationship — offer must not surface.
let bob_id = bob.core.status().endpoint_id.clone();
let bob_id = bob.core().status().endpoint_id.clone();
let source_dir = tempfile::tempdir().unwrap();
let source_path = source_dir.path().join("payload.txt");
std::fs::write(&source_path, b"nope").unwrap();
let err = alice
.core
.core()
.create_targeted_transfer(
bob_id,
vec![targeted_source(&source_path)],
@@ -329,7 +362,7 @@ fn invalid_offer_never_becomes_observable_pending_approval() {
)
.unwrap_err();
assert!(matches!(err, VnidropError::Permission { .. }));
assert!(bob.core.list_pending_targeted_offers().is_empty());
assert!(bob.core().list_pending_targeted_offers().is_empty());
}
#[test]
@@ -337,7 +370,7 @@ fn explicit_approval_gates_content_and_binds_authorization_to_receiver() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
let charlie = ProtectedNode::new();
let bob_id = bob.core.status().endpoint_id.clone();
let bob_id = bob.core().status().endpoint_id.clone();
establish_saved(&alice, &bob, 10_020);
let source_dir = tempfile::tempdir().unwrap();
@@ -345,7 +378,7 @@ fn explicit_approval_gates_content_and_binds_authorization_to_receiver() {
let payload = b"bound authorization payload";
std::fs::write(&source_path, payload).unwrap();
let bob_core = bob.core.clone();
let bob_core = bob.core().clone();
let offer_id = Arc::new(Mutex::new(None::<String>));
let offer_id_setter = offer_id.clone();
let gate = Arc::new(Mutex::new(false));
@@ -370,7 +403,7 @@ fn explicit_approval_gates_content_and_binds_authorization_to_receiver() {
.unwrap()
});
let alice_core = alice.core.clone();
let alice_core = alice.core().clone();
let source = targeted_source(&source_path);
let create = std::thread::spawn(move || {
alice_core.create_targeted_transfer(bob_id, vec![source], Some("payload.txt".to_string()))
@@ -390,7 +423,7 @@ fn explicit_approval_gates_content_and_binds_authorization_to_receiver() {
let _ = transfer_id;
// Without an approved authorization, receive must fail — no content yet.
let early_receive = bob.core.receive_targeted_transfer(
let early_receive = bob.core().receive_targeted_transfer(
"not-a-real-authorization".to_string(),
tempfile::tempdir()
.unwrap()
@@ -405,7 +438,7 @@ fn explicit_approval_gates_content_and_binds_authorization_to_receiver() {
create.join().unwrap().unwrap();
let output = tempfile::tempdir().unwrap();
bob.core
bob.core()
.receive_targeted_transfer(auth.clone(), output.path().to_string_lossy().into_owned())
.unwrap();
assert_eq!(
@@ -415,7 +448,7 @@ fn explicit_approval_gates_content_and_binds_authorization_to_receiver() {
let charlie_output = tempfile::tempdir().unwrap();
let leaked = charlie
.core
.core()
.receive_targeted_transfer(auth, charlie_output.path().to_string_lossy().into_owned());
assert!(
leaked.is_err(),
@@ -434,7 +467,7 @@ fn invitation_multi_receiver_shares_remain_independently_authorized() {
let first_receiver = ProtectedNode::new();
let second_receiver = ProtectedNode::new();
let share = sender
.core
.core()
.share_files(
vec![ShareSource {
kind: SourceKind::Path,
@@ -456,7 +489,7 @@ fn invitation_multi_receiver_shares_remain_independently_authorized() {
(&second_receiver, second_output.path()),
] {
receiver
.core
.core()
.receive(
share.ticket.clone(),
output.to_string_lossy().into_owned(),
@@ -469,3 +502,310 @@ fn invitation_multi_receiver_shares_remain_independently_authorized() {
);
}
}
fn approve_one(
alice: &ProtectedNode,
bob: &ProtectedNode,
payload: &[u8],
name: &str,
) -> (crate::TargetedTransfer, String) {
let bob_id = bob.core().status().endpoint_id.clone();
let source_dir = tempfile::tempdir().unwrap();
let source_path = source_dir.path().join(name);
std::fs::write(&source_path, payload).unwrap();
let bob_core = bob.core().clone();
let accept = std::thread::spawn(move || {
let offer = wait_for_pending_offer(&bob_core);
bob_core
.respond_to_targeted_offer(offer.transfer_id, true)
.unwrap()
});
let transfer = alice
.core()
.create_targeted_transfer(
bob_id,
vec![targeted_source(&source_path)],
Some(name.to_string()),
)
.unwrap();
let auth = accept.join().unwrap().expect("authorization");
(transfer, auth)
}
#[test]
fn protocol_ops_are_idempotent_for_stable_transfer_id() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 11_001);
let (transfer, auth) = approve_one(&alice, &bob, b"idempotent payload", "payload.txt");
// Replaying approval returns the same authorization — no duplicate prompts.
let again = bob
.core()
.respond_to_targeted_offer(transfer.id.clone(), true)
.unwrap()
.expect("idempotent authorization");
assert_eq!(again, auth);
let listed = bob
.core()
.list_targeted_transfers()
.unwrap()
.into_iter()
.filter(|entry| entry.id == transfer.id)
.count();
assert_eq!(listed, 1, "replay must not create duplicate durable rows");
}
#[test]
fn unapproved_offers_vanish_on_cancel_and_core_restart() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
let bob_id = bob.core().status().endpoint_id.clone();
establish_saved(&alice, &bob, 11_010);
let source_dir = tempfile::tempdir().unwrap();
let source_path = source_dir.path().join("payload.txt");
std::fs::write(&source_path, b"vanishing offer").unwrap();
let bob_core = bob.core().clone();
let seen = Arc::new(Mutex::new(None::<String>));
let seen_set = seen.clone();
let hold = Arc::new(Mutex::new(true));
let hold_wait = hold.clone();
let watcher = std::thread::spawn(move || {
let offer = wait_for_pending_offer(&bob_core);
*seen_set.lock().unwrap() = Some(offer.transfer_id.clone());
let started = Instant::now();
while *hold_wait.lock().unwrap() {
assert!(
started.elapsed() < Duration::from_secs(30),
"cancel never cleared the live offer"
);
if bob_core.list_pending_targeted_offers().is_empty() {
return;
}
std::thread::sleep(Duration::from_millis(25));
}
});
let alice_core = alice.core().clone();
let source = targeted_source(&source_path);
let create = std::thread::spawn(move || {
alice_core.create_targeted_transfer(bob_id, vec![source], Some("payload.txt".to_string()))
});
let transfer_id = loop {
if let Some(id) = seen.lock().unwrap().clone() {
break id;
}
std::thread::sleep(Duration::from_millis(25));
};
alice.core().cancel_targeted_transfer(transfer_id).unwrap();
watcher.join().unwrap();
let create_err = create.join().unwrap();
assert!(create_err.is_err(), "cancelled offer must fail create");
assert!(bob.core().list_pending_targeted_offers().is_empty());
// Restart clears any live-session inbox residue.
let bob = bob.restart();
assert!(bob.core().list_pending_targeted_offers().is_empty());
*hold.lock().unwrap() = false;
}
#[test]
fn approved_transfer_resumes_after_restart_without_reapproval() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 11_020);
let (transfer, _auth) = approve_one(&alice, &bob, b"resume me please", "payload.txt");
let bob_before = bob
.core()
.get_targeted_transfer(transfer.id.clone())
.unwrap()
.expect("receiver durable approved row");
assert_eq!(bob_before.state, TargetedTransferState::Approved);
assert_eq!(bob_before.verified_bytes, 0);
let alice = alice.restart();
let bob = bob.restart();
let alice_after = alice
.core()
.get_targeted_transfer(transfer.id.clone())
.unwrap()
.expect("sender durable state");
assert_eq!(alice_after.state, TargetedTransferState::Approved);
let bob_after = bob
.core()
.get_targeted_transfer(transfer.id.clone())
.unwrap()
.expect("receiver durable state");
assert_eq!(bob_after.state, TargetedTransferState::Approved);
let output = tempfile::tempdir().unwrap();
bob.core()
.resume_targeted_transfer(
transfer.id.clone(),
output.path().to_string_lossy().into_owned(),
)
.unwrap();
assert_eq!(
std::fs::read(output.path().join("payload.txt")).unwrap(),
b"resume me please"
);
let completed = bob
.core()
.get_targeted_transfer(transfer.id)
.unwrap()
.unwrap();
assert_eq!(completed.state, TargetedTransferState::Completed);
assert_eq!(completed.verified_bytes, b"resume me please".len() as u64);
}
#[test]
fn manifest_change_requires_new_transfer_identity() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 11_030);
let (first, _) = approve_one(&alice, &bob, b"first manifest", "a.txt");
let (second, _) = approve_one(&alice, &bob, b"second manifest", "b.txt");
assert_ne!(first.id, second.id);
assert_ne!(first.manifest_id, second.manifest_id);
}
#[test]
fn cancel_revokes_access_and_stops_streaming() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 11_040);
let (transfer, auth) = approve_one(&alice, &bob, b"cancel me", "payload.txt");
alice
.core()
.cancel_targeted_transfer(transfer.id.clone())
.unwrap();
let cancelled = alice
.core()
.get_targeted_transfer(transfer.id.clone())
.unwrap()
.unwrap();
assert_eq!(cancelled.state, TargetedTransferState::Cancelled);
let output = tempfile::tempdir().unwrap();
let receive = bob
.core()
.receive_targeted_transfer(auth, output.path().to_string_lossy().into_owned());
assert!(
receive.is_err(),
"cancelled transfer must not remain receivable"
);
}
#[test]
fn delete_removes_authorization_and_resumable_state() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 11_050);
let (transfer, auth) = approve_one(&alice, &bob, b"delete me", "payload.txt");
bob.core()
.delete_targeted_transfer(transfer.id.clone())
.unwrap();
let deleted = bob
.core()
.get_targeted_transfer(transfer.id.clone())
.unwrap()
.unwrap();
assert_eq!(deleted.state, TargetedTransferState::Deleted);
assert_eq!(deleted.verified_bytes, 0);
let resume = bob.core().resume_targeted_transfer(
transfer.id.clone(),
tempfile::tempdir()
.unwrap()
.path()
.to_string_lossy()
.into_owned(),
);
assert!(resume.is_err(), "deleted transfer must not resume");
let receive = bob.core().receive_targeted_transfer(
auth,
tempfile::tempdir()
.unwrap()
.path()
.to_string_lossy()
.into_owned(),
);
// Auth blob may still decode, but durable resume path is gone; receive may
// still attempt content pull if sender serves — sender delete is separate.
let _ = receive;
alice
.core()
.delete_targeted_transfer(transfer.id.clone())
.unwrap();
let sender_deleted = alice
.core()
.get_targeted_transfer(transfer.id)
.unwrap()
.unwrap();
assert_eq!(sender_deleted.state, TargetedTransferState::Deleted);
}
#[test]
fn concurrent_independent_transfers_between_same_devices_are_isolated() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 11_060);
// Design allows only one unresolved offer per sender; approve sequentially,
// then prove independent approved transfers do not corrupt each other.
let (first, first_auth) = approve_one(&alice, &bob, b"alpha", "one.txt");
let (second, second_auth) = approve_one(&alice, &bob, b"beta-payload", "two.txt");
assert_ne!(first.id, second.id);
alice
.core()
.cancel_targeted_transfer(first.id.clone())
.unwrap();
assert_eq!(
alice
.core()
.get_targeted_transfer(first.id)
.unwrap()
.unwrap()
.state,
TargetedTransferState::Cancelled
);
assert_eq!(
alice
.core()
.get_targeted_transfer(second.id.clone())
.unwrap()
.unwrap()
.state,
TargetedTransferState::Approved
);
let output = tempfile::tempdir().unwrap();
bob.core()
.receive_targeted_transfer(second_auth, output.path().to_string_lossy().into_owned())
.unwrap();
assert_eq!(
std::fs::read(output.path().join("two.txt")).unwrap(),
b"beta-payload"
);
let cancelled_output = tempfile::tempdir().unwrap();
assert!(bob
.core()
.receive_targeted_transfer(
first_auth,
cancelled_output.path().to_string_lossy().into_owned(),
)
.is_err());
}

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

@@ -43,6 +43,7 @@ fn saved_devices_relationships_and_targeted_transfers_are_distinct_contracts() {
manifest_id: "immutable-manifest-id".to_string(),
file_count: 2,
total_size: 42,
verified_bytes: 0,
state: TargetedTransferState::AwaitingApproval,
created_at: 3_000,
updated_at: 3_000,
@@ -67,6 +68,7 @@ fn targeted_transfer_transitions_are_validated_by_the_domain() {
(State::AwaitingApproval, State::Approved),
(State::AwaitingApproval, State::Declined),
(State::Approved, State::Connecting),
(State::Approved, State::Deleted),
(State::Connecting, State::Transferring),
(State::Connecting, State::Interrupted),
(State::Transferring, State::Completed),