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): complete targeted transfers without invitation state

Browse Source
This commit is contained in:
Hammed Abass
2026-08-12 15:32:06 +02:00
parent 644c9bfda3
commit 32d69b9771
14 changed files with 953 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -23,7 +23,7 @@ pub fn experimental_saved_device_capabilities() -> ExperimentalSavedDeviceCapabi
ExperimentalSavedDeviceCapabilities {
domain_contract_version: 1,
relationship_protocol_version: 1,
targeted_transfer_protocol_version: 1,
targeted_transfer_protocol_version: 2,
}
}

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

@@ -124,6 +124,35 @@ impl VnidropCore {
})
}
pub(crate) fn suppress_targeted_completion_for_test(&self, suppress: bool) {
self.inner
.suppress_targeted_completion
.store(suppress, std::sync::atomic::Ordering::SeqCst);
}
pub(crate) fn targeted_payload_is_registered_for_test(
&self,
id: String,
) -> Result<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")))?;
let hash = row
.content_hash
.parse::<iroh_blobs::Hash>()
.map_err(|error| VnidropError::invalid_input(anyhow::anyhow!(error)))?;
Ok(self
.inner
.transfer_ids_for_hash(hash)
.await
.contains(&row.protocol_transfer_id))
})
}
pub(crate) fn initialize_with_test_secret_store_limits_and_network(
app_data_dir: String,
event_sink: Arc<dyn CoreEventSink>,

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

@@ -191,6 +191,10 @@ impl CoreInner {
task.abort();
let _ = task.await;
}
if let Some(task) = self.targeted_completion_task.lock().await.take() {
task.abort();
let _ = task.await;
}
if let Err(error) = self.router.shutdown().await {
self.emit_endpoint(
"shutdown",

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

@@ -122,10 +122,13 @@ 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) 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,
}
pub(super) struct ActiveTransfer {
@@ -445,9 +448,12 @@ impl CoreInner {
provider_task: TokioMutex::new(None),
delivery_receipt_notify: Notify::new(),
delivery_receipt_task: TokioMutex::new(None),
targeted_completion_task: TokioMutex::new(None),
shutdown_started: AtomicBool::new(false),
#[cfg(test)]
targeted_cancel_log: std::sync::Mutex::new(Vec::new()),
#[cfg(test)]
suppress_targeted_completion: AtomicBool::new(false),
});
// In-flight connecting/transferring transfers become Interrupted across restart.
@@ -470,6 +476,7 @@ impl CoreInner {
);
inner.spawn_provider_event_task(event_rx).await;
inner.spawn_delivery_receipt_task().await;
inner.spawn_targeted_completion_task().await;
if let Err(error) = inner.pairing_eligibility.reconcile().await {
tracing::warn!(%error, "failed to reconcile pairing eligibility");
}

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

@@ -213,8 +213,9 @@ impl CoreInner {
},
);
let result = tokio::select! {
result = self.download_targeted_payload(transfer_id, blob_ticket, target) => result,
biased;
_ = &mut cancelled => Err(VnidropError::cancelled("transfer cancelled").into()),
result = self.download_targeted_payload(transfer_id, blob_ticket, target) => result,
};
self.active_transfers
.lock()

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

@@ -17,8 +17,9 @@ use crate::{
targeted_transfer::{
auth_secret_material,
protocol::{
map_offer_refuse_reason, CancelTargetedOffer, DeliverTargetedAuthorization,
SubmitTargetedOffer, TargetedTransferProtocol, WireOfferResponse,
map_offer_refuse_reason, CancelTargetedOffer, CompleteTargetedTransfer,
CompletionResponse, DeliverTargetedAuthorization, SubmitTargetedOffer,
TargetedTransferProtocol, WireOfferResponse,
},
reconstruct_authorization, TargetedAuthorization, TargetedAuthorizationDraft,
TargetedTransferRole, TargetedTransferRow,
@@ -35,6 +36,71 @@ impl CoreInner {
std::time::Duration::from_millis(self.limits.offer_timeout_ms)
}
pub(super) async fn spawn_targeted_completion_task(self: &Arc<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_completions().await;
}
});
*self.targeted_completion_task.lock().await = Some(task);
}
async fn retry_targeted_completions(&self) {
let Ok(rows) = self.targeted_store().list_pending_completions().await else {
return;
};
for row in rows.into_iter().take(1) {
let Ok(Some(encoded)) = self.load_stored_authorization(&row).await else {
let _ = self
.targeted_store()
.defer_pending_completion(&row.id, now_ms() + 30_000)
.await;
continue;
};
let Ok(auth) = TargetedAuthorization::decode(&encoded) else {
let _ = self
.targeted_store()
.defer_pending_completion(&row.id, now_ms() + 30_000)
.await;
continue;
};
if self.acknowledge_targeted_completion(&auth).await.is_ok() {
let _ = self
.targeted_store()
.clear_pending_completion(&row.id)
.await;
} else {
let _ = self
.targeted_store()
.defer_pending_completion(&row.id, now_ms() + 5_000)
.await;
}
}
}
async fn release_completed_targeted_payloads(&self) {
let Ok(rows) = self.targeted_store().list_completed_sender_rows().await else {
return;
};
for row in rows {
if self
.try_teardown_targeted_payload(row.protocol_transfer_id, Some(&row.id))
.await
.is_ok()
{
let _ = self
.targeted_store()
.clear_pending_payload_release(&row.id)
.await;
}
}
}
pub(super) fn targeted_store(&self) -> crate::targeted_transfer::TargetedTransferStore {
self.targeted_transfers.clone()
}
@@ -628,6 +694,19 @@ impl CoreInner {
}
};
if let Err(error) = store
.set_state(
&transfer_uuid,
TargetedTransferState::AwaitingApproval,
TargetedTransferState::Approved,
)
.await
{
self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid))
.await;
return Err(error);
}
let deliver = match client
.deliver_authorization(DeliverTargetedAuthorization {
transfer_id: transfer_uuid.clone(),
@@ -649,11 +728,7 @@ impl CoreInner {
};
if deliver != crate::targeted_transfer::protocol::DeliverAuthorizationResponse::Stored {
let _ = store
.set_state(
&transfer_uuid,
TargetedTransferState::AwaitingApproval,
TargetedTransferState::Failed,
)
.set_state_from_any(&transfer_uuid, TargetedTransferState::Failed)
.await;
self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid))
.await;
@@ -662,14 +737,6 @@ impl CoreInner {
)));
}
store
.set_state(
&transfer_uuid,
TargetedTransferState::AwaitingApproval,
TargetedTransferState::Approved,
)
.await?;
store
.get(&transfer_uuid)
.await?
@@ -677,15 +744,31 @@ impl CoreInner {
}
async fn teardown_targeted_payload(&self, protocol_transfer_id: u64, id: Option<&str>) {
if let Err(error) = self
.try_teardown_targeted_payload(protocol_transfer_id, id)
.await
{
tracing::warn!(%error, "failed to release targeted payload");
}
}
async fn try_teardown_targeted_payload(
&self,
protocol_transfer_id: u64,
id: Option<&str>,
) -> Result<(), VnidropError> {
self.unregister_transfer_hashes(protocol_transfer_id).await;
self.access_policy
.remove_transfer(protocol_transfer_id)
.await;
if let Some(id) = id {
if let Err(error) = self.store.tags().delete(targeted_tag_name(id)).await {
tracing::warn!(%error, transfer_id = id, "failed to release targeted payload tag");
}
self.store
.tags()
.delete(targeted_tag_name(id))
.await
.map_err(VnidropError::transfer)?;
}
Ok(())
}
pub(super) async fn receive_targeted_transfer(
@@ -693,11 +776,10 @@ impl CoreInner {
transfer_id: String,
output_dir: String,
) -> Result<(), VnidropError> {
self.receive_targeted_to_target(
transfer_id,
ReceiveTarget::Directory(std::path::PathBuf::from(output_dir)),
)
.await
let output_dir =
crate::filesystem::platform_path(&output_dir).map_err(VnidropError::filesystem)?;
self.receive_targeted_to_target(transfer_id, ReceiveTarget::Directory(output_dir))
.await
}
pub(super) async fn receive_targeted_transfer_with_output_sink(
@@ -723,11 +805,10 @@ impl CoreInner {
id: String,
output_dir: String,
) -> Result<(), VnidropError> {
self.resume_targeted_to_target(
id,
ReceiveTarget::Directory(std::path::PathBuf::from(output_dir)),
)
.await
let output_dir =
crate::filesystem::platform_path(&output_dir).map_err(VnidropError::filesystem)?;
self.resume_targeted_to_target(id, ReceiveTarget::Directory(output_dir))
.await
}
pub(super) async fn resume_targeted_transfer_with_output_sink(
@@ -826,16 +907,11 @@ impl CoreInner {
)
.await?;
}
TargetedTransferState::Connecting => {
store
.set_state(
&auth.transfer_id,
TargetedTransferState::Connecting,
TargetedTransferState::Transferring,
)
.await?;
TargetedTransferState::Connecting | TargetedTransferState::Transferring => {
return Err(VnidropError::InvalidTransition {
reason: "targeted receive is already active".to_string(),
});
}
TargetedTransferState::Transferring => {}
other => {
return Err(VnidropError::InvalidTransition {
reason: format!(
@@ -855,17 +931,14 @@ impl CoreInner {
match receive_result {
Ok(()) => {
if let Ok(Some(row)) = store.get_row(&auth.transfer_id).await {
let _ = store
.set_verified_bytes(&auth.transfer_id, row.total_size)
.await;
let _ = store
.set_state(
&auth.transfer_id,
TargetedTransferState::Transferring,
TargetedTransferState::Completed,
)
.await;
let row = store.get_row(&auth.transfer_id).await?.ok_or_else(|| {
VnidropError::internal(anyhow::anyhow!("targeted transfer missing"))
})?;
store
.complete_receiver_and_enqueue(&auth.transfer_id, row.total_size)
.await?;
if self.acknowledge_targeted_completion(auth).await.is_ok() {
store.clear_pending_completion(&auth.transfer_id).await?;
}
Ok(())
}
@@ -888,6 +961,43 @@ impl CoreInner {
}
}
async fn acknowledge_targeted_completion(
&self,
auth: &TargetedAuthorization,
) -> Result<(), VnidropError> {
#[cfg(test)]
if self
.suppress_targeted_completion
.load(std::sync::atomic::Ordering::SeqCst)
{
return Err(VnidropError::device_unavailable(anyhow::anyhow!(
"completion delivery suppressed by test"
)));
}
let addr = self
.device_relationships
.peer_addr(&auth.sender_endpoint_id)
.await?;
let client = TargetedTransferProtocol::client(self.endpoint.clone(), addr);
let response = tokio::time::timeout(
self.connection_timeout(),
client.complete_transfer(CompleteTargetedTransfer {
transfer_id: auth.transfer_id.clone(),
verified_bytes: auth.total_size,
authorization: auth.encode()?,
}),
)
.await
.map_err(|_| VnidropError::device_unavailable(anyhow::anyhow!("completion timed out")))?
.map_err(|error| VnidropError::network(anyhow::anyhow!(error)))?;
if response != CompletionResponse::Recorded {
return Err(VnidropError::permission(anyhow::anyhow!(
"sender rejected targeted completion"
)));
}
Ok(())
}
async fn persist_authorization_secret(
&self,
transfer_id: &str,

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

@@ -218,6 +218,27 @@ impl TargetedOfferInbox {
.map(|entry| entry.offer.clone())
}
pub(crate) async fn authorization_matches_pending(
&self,
auth: &crate::targeted_transfer::TargetedAuthorization,
) -> bool {
self.pending
.lock()
.await
.get(&auth.transfer_id)
.is_some_and(|entry| {
let offer = &entry.offer;
offer.sender_endpoint_id == auth.sender_endpoint_id
&& offer.receiver_endpoint_id == auth.receiver_endpoint_id
&& offer.manifest_id == auth.manifest_id
&& offer.content_hash == auth.content_hash
&& offer.transfer_name == auth.transfer_name
&& offer.file_count == auth.file_count
&& offer.total_size == auth.total_size
&& offer.protocol_version == auth.protocol_version
})
}
pub(crate) async fn settled_authorization(&self, transfer_id: &str) -> Option<String> {
match self.settled.lock().await.get(transfer_id) {
Some(SettledOfferResult::Accepted {
@@ -370,6 +391,7 @@ fn offers_equivalent(left: &PendingTargetedOffer, right: &PendingTargetedOffer)
&& left.receiver_endpoint_id == right.receiver_endpoint_id
&& left.manifest_id == right.manifest_id
&& left.content_hash == right.content_hash
&& left.transfer_name == right.transfer_name
&& left.file_count == right.file_count
&& left.total_size == right.total_size
&& left.protocol_version == right.protocol_version

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

@@ -18,7 +18,7 @@ use serde::{Deserialize, Serialize};
use super::{
auth::TargetedAuthorization,
inbox::{TargetedOfferDecision, TargetedOfferInbox},
state_as_str, TargetedTransferStore,
state_as_str, TargetedTransferRole, TargetedTransferStore,
};
use crate::{
api::{
@@ -50,7 +50,7 @@ impl fmt::Debug for TargetedTransferProtocol {
}
impl TargetedTransferProtocol {
pub(crate) const ALPN: &'static [u8] = b"/vnidrop/targeted-transfer/1";
pub(crate) const ALPN: &'static [u8] = b"/vnidrop/targeted-transfer/2";
pub(crate) fn new(
relationships: std::sync::Arc<DeviceRelationshipService>,
@@ -248,18 +248,45 @@ impl TargetedTransferProtocol {
}
if let Ok(Some(row)) = self.store.get_row(&delivery.transfer_id).await {
if row.authorization_secret_handle.is_some()
&& row.protocol_transfer_id == auth.protocol_transfer_id
&& row.sender_endpoint_id == auth.sender_endpoint_id
&& row.receiver_endpoint_id == auth.receiver_endpoint_id
&& row.manifest_id == auth.manifest_id
&& row.content_hash == auth.content_hash
&& row.transfer_name == auth.transfer_name
&& row.file_count == auth.file_count
&& row.total_size == auth.total_size
&& row.blob_ticket.as_deref() == Some(auth.blob_ticket.as_str())
&& auth.protocol_version
== experimental_saved_device_capabilities().targeted_transfer_protocol_version
{
return DeliverAuthorizationResponse::Stored;
}
if row.authorization_secret_handle.is_some() {
return DeliverAuthorizationResponse::Rejected;
}
}
if !self.inbox.authorization_matches_pending(&auth).await {
return DeliverAuthorizationResponse::Rejected;
}
if self
.inbox
.deliver_authorization(&delivery.transfer_id, delivery.authorization)
.await
{
DeliverAuthorizationResponse::Stored
let deadline = tokio::time::Instant::now()
+ std::time::Duration::from_millis(self.limits.offer_timeout_ms);
loop {
if let Ok(Some(row)) = self.store.get_row(&delivery.transfer_id).await {
if row.authorization_secret_handle.is_some() {
return DeliverAuthorizationResponse::Stored;
}
}
if tokio::time::Instant::now() >= deadline {
return DeliverAuthorizationResponse::Rejected;
}
tokio::time::sleep(std::time::Duration::from_millis(25)).await;
}
} else {
DeliverAuthorizationResponse::Rejected
}
@@ -285,6 +312,65 @@ impl TargetedTransferProtocol {
}
CancelWireOfferResponse::Cancelled
}
async fn handle_completion(
&self,
remote_endpoint_id: &str,
completion: CompleteTargetedTransfer,
) -> CompletionResponse {
if self
.relationships
.require_saved(remote_endpoint_id)
.await
.is_err()
{
return CompletionResponse::Rejected;
}
let Ok(Some(row)) = self.store.get_row(&completion.transfer_id).await else {
return CompletionResponse::Rejected;
};
let Ok(auth) = TargetedAuthorization::decode(&completion.authorization) else {
return CompletionResponse::Rejected;
};
if row.role != TargetedTransferRole::Sender
|| row.receiver_endpoint_id != remote_endpoint_id
|| auth.receiver_endpoint_id != remote_endpoint_id
|| auth.sender_endpoint_id != self.local_endpoint_id
|| auth.transfer_id != row.id
|| auth.protocol_transfer_id != row.protocol_transfer_id
|| auth.manifest_id != row.manifest_id
|| auth.content_hash != row.content_hash
|| auth.transfer_name != row.transfer_name
|| auth.file_count != row.file_count
|| auth.total_size != row.total_size
|| auth.total_size != completion.verified_bytes
|| row.blob_ticket.as_deref() != Some(auth.blob_ticket.as_str())
|| auth.protocol_version
!= experimental_saved_device_capabilities().targeted_transfer_protocol_version
{
return CompletionResponse::Rejected;
}
if row.state == TargetedTransferState::Completed {
return CompletionResponse::Recorded;
}
if matches!(
row.state,
TargetedTransferState::Cancelled
| TargetedTransferState::Declined
| TargetedTransferState::Failed
| TargetedTransferState::Deleted
) {
return CompletionResponse::Rejected;
}
match self
.store
.mark_sender_completed(&completion.transfer_id)
.await
{
Ok(()) => CompletionResponse::Recorded,
Err(_) => CompletionResponse::Rejected,
}
}
}
impl ProtocolHandler for TargetedTransferProtocol {
@@ -320,6 +406,11 @@ impl ProtocolHandler for TargetedTransferProtocol {
let response = self.handle_cancel(&remote_endpoint_id, inner).await;
let _ = tx.send(response).await;
}
TargetedTransferMessage::CompleteTargetedTransfer(message) => {
let WithChannels { inner, tx, .. } = message;
let response = self.handle_completion(&remote_endpoint_id, inner).await;
let _ = tx.send(response).await;
}
}
}
connection.closed().await;
@@ -357,6 +448,13 @@ impl TargetedTransferClient {
) -> Result<CancelWireOfferResponse, irpc::Error> {
self.inner.rpc(cancel).await
}
pub(crate) async fn complete_transfer(
&self,
completion: CompleteTargetedTransfer,
) -> Result<CompletionResponse, irpc::Error> {
self.inner.rpc(completion).await
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -415,6 +513,19 @@ pub(crate) enum CancelWireOfferResponse {
Rejected,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct CompleteTargetedTransfer {
pub(crate) transfer_id: String,
pub(crate) verified_bytes: u64,
pub(crate) authorization: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub(crate) enum CompletionResponse {
Recorded,
Rejected,
}
#[rpc_requests(message = TargetedTransferMessage)]
#[derive(Debug, Serialize, Deserialize)]
#[allow(
@@ -430,6 +541,8 @@ enum TargetedTransferMessages {
DeliverTargetedAuthorization(DeliverTargetedAuthorization),
#[rpc(tx = oneshot::Sender<CancelWireOfferResponse>)]
CancelTargetedOffer(CancelTargetedOffer),
#[rpc(tx = oneshot::Sender<CompletionResponse>)]
CompleteTargetedTransfer(CompleteTargetedTransfer),
}
fn parse_offer_relay_urls(values: &[String]) -> Result<Vec<RelayUrl>, ()> {

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

@@ -36,6 +36,42 @@ pub(crate) async fn ensure_schema(pool: &SqlitePool) -> anyhow::Result<()> {
)
.execute(pool)
.await?;
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS targeted_completion_outbox (
transfer_id TEXT PRIMARY KEY,
created_at INTEGER NOT NULL,
next_attempt_at INTEGER NOT NULL,
FOREIGN KEY(transfer_id) REFERENCES targeted_transfers(id) ON DELETE CASCADE
);
"#,
)
.execute(pool)
.await?;
let completion_columns = sqlx::query("PRAGMA table_info(targeted_completion_outbox)")
.fetch_all(pool)
.await?;
if !completion_columns
.iter()
.any(|row| row.get::<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?;
@@ -190,27 +226,156 @@ impl TargetedTransferStore {
Ok(())
}
pub(crate) async fn set_verified_bytes(
pub(crate) async fn mark_sender_completed(&self, id: &str) -> Result<(), VnidropError> {
let mut transaction = self.pool.begin().await.map_err(VnidropError::repository)?;
let result = sqlx::query(
r#"
UPDATE targeted_transfers
SET state = 'completed', verified_bytes = total_size, updated_at = ?2
WHERE id = ?1 AND role = 'sender'
AND state IN ('approved', 'connecting', 'transferring', 'interrupted')
"#,
)
.bind(id)
.bind(now_ms())
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
if result.rows_affected() == 0 {
return Err(VnidropError::InvalidTransition {
reason: "sender transfer cannot be completed".to_string(),
});
}
sqlx::query(
"INSERT OR IGNORE INTO targeted_payload_release_outbox (transfer_id, created_at) VALUES (?1, ?2)",
)
.bind(id)
.bind(now_ms())
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
transaction
.commit()
.await
.map_err(VnidropError::repository)?;
Ok(())
}
pub(crate) async fn complete_receiver_and_enqueue(
&self,
id: &str,
verified_bytes: u64,
) -> Result<(), VnidropError> {
sqlx::query(
let mut transaction = self.pool.begin().await.map_err(VnidropError::repository)?;
let result = sqlx::query(
r#"
UPDATE targeted_transfers
SET verified_bytes = ?2, updated_at = ?3
WHERE id = ?1
SET state = 'completed', verified_bytes = ?2, updated_at = ?3
WHERE id = ?1 AND role = 'receiver' AND state = 'transferring'
"#,
)
.bind(id)
.bind(verified_bytes as i64)
.bind(now_ms())
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
if result.rows_affected() == 0 {
return Err(VnidropError::InvalidTransition {
reason: "receiver transfer cannot be completed".to_string(),
});
}
sqlx::query(
"INSERT OR IGNORE INTO targeted_completion_outbox (transfer_id, created_at, next_attempt_at) VALUES (?1, ?2, ?2)",
)
.bind(id)
.bind(now_ms())
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
transaction
.commit()
.await
.map_err(VnidropError::repository)?;
Ok(())
}
pub(crate) async fn list_pending_completions(
&self,
) -> Result<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(
&self,
id: &str,
@@ -237,6 +402,17 @@ impl TargetedTransferStore {
}
pub(crate) async fn clear_authorization(&self, id: &str) -> Result<(), VnidropError> {
let mut transaction = self.pool.begin().await.map_err(VnidropError::repository)?;
sqlx::query("DELETE FROM targeted_completion_outbox WHERE transfer_id = ?1")
.bind(id)
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
sqlx::query("DELETE FROM targeted_payload_release_outbox WHERE transfer_id = ?1")
.bind(id)
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
sqlx::query(
r#"
UPDATE targeted_transfers
@@ -249,9 +425,13 @@ impl TargetedTransferStore {
)
.bind(id)
.bind(now_ms())
.execute(&self.pool)
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
transaction
.commit()
.await
.map_err(VnidropError::repository)?;
Ok(())
}

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

@@ -69,5 +69,5 @@ fn public_api_exposes_saved_device_surface_without_prototype_contact_entry_point
let caps = crate::experimental_saved_device_capabilities();
assert_eq!(caps.domain_contract_version, 1);
assert_eq!(caps.relationship_protocol_version, 1);
assert_eq!(caps.targeted_transfer_protocol_version, 1);
assert_eq!(caps.targeted_transfer_protocol_version, 2);
}

62
crates/vnidrop/src/tests/control_plane.rs
View File

@@ -9,6 +9,7 @@ use crate::{
invitation::Repository,
secure_secret::FaultInjectingSecretStore,
targeted_transfer::inbox::{TargetedOfferDecision, TargetedOfferInbox},
targeted_transfer::{TargetedAuthorization, TargetedAuthorizationDraft},
CoreNetworkConfig, DeviceRelationshipState, ShareMetadataInput, ShareSource, SourceKind,
TransferAccessMode, VnidropCore, VnidropError,
};
@@ -17,6 +18,67 @@ struct RecordingSink {
events: Mutex<Vec<CoreEvent>>,
}
#[tokio::test]
async fn delivered_authorization_must_match_the_approved_offer_projection() {
let (inbox, _) = inbox_with_limits(1, 60_000, 5).await;
let offer = sample_offer("bound-offer", "sender-a");
let submit = {
let inbox = inbox.clone();
let offer = offer.clone();
tokio::spawn(async move { inbox.submit(offer).await })
};
let started = std::time::Instant::now();
while inbox.list().await.is_empty() {
assert!(started.elapsed() < std::time::Duration::from_secs(2));
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
}
let exact = TargetedAuthorization::issue(TargetedAuthorizationDraft {
transfer_id: offer.transfer_id.clone(),
protocol_transfer_id: 42,
sender_endpoint_id: offer.sender_endpoint_id.clone(),
receiver_endpoint_id: offer.receiver_endpoint_id.clone(),
manifest_id: offer.manifest_id.clone(),
content_hash: offer.content_hash.clone(),
file_count: offer.file_count,
total_size: offer.total_size,
protocol_version: offer.protocol_version,
transfer_name: offer.transfer_name.clone(),
blob_ticket: "blob-a".to_string(),
})
.unwrap();
assert!(inbox.authorization_matches_pending(&exact).await);
let substituted = TargetedAuthorization::issue(TargetedAuthorizationDraft {
manifest_id: "replacement-manifest".to_string(),
content_hash: "replacement-hash".to_string(),
transfer_name: "replacement.txt".to_string(),
total_size: 99,
blob_ticket: "blob-b".to_string(),
..TargetedAuthorizationDraft {
transfer_id: offer.transfer_id,
protocol_transfer_id: 42,
sender_endpoint_id: offer.sender_endpoint_id,
receiver_endpoint_id: offer.receiver_endpoint_id,
manifest_id: offer.manifest_id,
content_hash: offer.content_hash,
file_count: offer.file_count,
total_size: offer.total_size,
protocol_version: offer.protocol_version,
transfer_name: offer.transfer_name,
blob_ticket: "blob-a".to_string(),
}
})
.unwrap();
assert!(!inbox.authorization_matches_pending(&substituted).await);
assert!(
inbox.list().await.len() == 1,
"rejection keeps the approved offer pending"
);
inbox.discard("bound-offer").await;
let _ = submit.await;
}
impl CoreEventSink for RecordingSink {
fn on_event(&self, event: CoreEvent) {
self.events.lock().unwrap().push(event);

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

@@ -35,6 +35,8 @@ async fn open_all_returns_all_domain_stores_and_schemas() {
"protected_secret_refs",
"blocked_endpoints",
"targeted_transfers",
"targeted_completion_outbox",
"targeted_payload_release_outbox",
"transfers",
] {
let row = sqlx::query(&format!(
@@ -51,6 +53,38 @@ async fn open_all_returns_all_domain_stores_and_schemas() {
}
}
#[tokio::test]
async fn open_all_migrates_targeted_completion_retry_schedule() {
let temp = tempfile::tempdir().unwrap();
let db = temp.path().join("vnidrop.sqlite3");
let options = sqlx::sqlite::SqliteConnectOptions::new()
.filename(&db)
.create_if_missing(true);
let pool = sqlx::SqlitePool::connect_with(options).await.unwrap();
sqlx::query(
r#"
CREATE TABLE targeted_completion_outbox (
transfer_id TEXT PRIMARY KEY,
created_at INTEGER NOT NULL
)
"#,
)
.execute(&pool)
.await
.unwrap();
drop(pool);
let _stores = persistence::open_all(temp.path()).await.unwrap();
let pool = open_profile_pool(temp.path()).await;
let columns = sqlx::query("PRAGMA table_info(targeted_completion_outbox)")
.fetch_all(&pool)
.await
.unwrap();
assert!(columns
.iter()
.any(|row| row.get::<String, _>("name") == "next_attempt_at"));
}
#[tokio::test]
async fn open_all_migrates_name_columns_without_losing_existing_local_labels() {
let temp = tempfile::tempdir().unwrap();

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

@@ -412,6 +412,7 @@ fn explicit_approval_gates_content_and_binds_authorization_to_receiver() {
let bob_invitation_count = bob.core().list_transfers().unwrap().len();
let alice_eligibility_count = alice.core().list_pairing_eligibilities().unwrap().len();
let bob_eligibility_count = bob.core().list_pairing_eligibilities().unwrap().len();
let bob_artifacts_before = bob.core().list_received_artifacts().unwrap();
let source_dir = tempfile::tempdir().unwrap();
let source_path = source_dir.path().join("payload.txt");
@@ -508,6 +509,10 @@ fn explicit_approval_gates_content_and_binds_authorization_to_receiver() {
bob.core().list_pairing_eligibilities().unwrap().len(),
bob_eligibility_count
);
assert_eq!(
bob.core().list_received_artifacts().unwrap().len(),
bob_artifacts_before.len()
);
let charlie_output = tempfile::tempdir().unwrap();
let leaked = charlie.core().receive_targeted_transfer(
@@ -788,6 +793,127 @@ fn approved_transfer_resumes_after_restart_without_reapproval() {
.unwrap();
assert_eq!(completed.state, TargetedTransferState::Completed);
assert_eq!(completed.verified_bytes, b"resume me please".len() as u64);
let sender_completed = alice
.core()
.get_targeted_transfer(completed.id)
.unwrap()
.unwrap();
assert_eq!(sender_completed.state, TargetedTransferState::Completed);
let started = Instant::now();
while alice
.core()
.targeted_payload_is_registered_for_test(sender_completed.id.clone())
.unwrap()
{
assert!(
started.elapsed() < Duration::from_secs(10),
"completed sender payload access was not released"
);
std::thread::sleep(Duration::from_millis(25));
}
}
#[test]
fn completion_retries_after_receiver_restart_without_failing_published_receive() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 11_025);
let transfer = approve_one(&alice, &bob, b"eventual completion", "payload.txt");
bob.core().suppress_targeted_completion_for_test(true);
let output = tempfile::tempdir().unwrap();
bob.core()
.receive_targeted_transfer(
transfer.id.clone(),
output.path().to_string_lossy().into_owned(),
)
.unwrap();
assert_eq!(
std::fs::read(output.path().join("payload.txt")).unwrap(),
b"eventual completion"
);
assert_eq!(
bob.core()
.get_targeted_transfer(transfer.id.clone())
.unwrap()
.unwrap()
.state,
TargetedTransferState::Completed
);
assert_eq!(
alice
.core()
.get_targeted_transfer(transfer.id.clone())
.unwrap()
.unwrap()
.state,
TargetedTransferState::Approved
);
let bob = bob.restart();
let started = Instant::now();
loop {
if alice
.core()
.get_targeted_transfer(transfer.id.clone())
.unwrap()
.is_some_and(|entry| entry.state == TargetedTransferState::Completed)
{
break;
}
assert!(
started.elapsed() < Duration::from_secs(15),
"completion outbox was not retried after restart"
);
std::thread::sleep(Duration::from_millis(50));
}
assert_eq!(
bob.core()
.get_targeted_transfer(transfer.id)
.unwrap()
.unwrap()
.state,
TargetedTransferState::Completed
);
}
#[test]
fn targeted_path_receive_preserves_no_overwrite_and_resumes_elsewhere() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 11_026);
let transfer = approve_one(&alice, &bob, b"new payload", "payload.txt");
let occupied = tempfile::tempdir().unwrap();
std::fs::write(occupied.path().join("payload.txt"), b"keep me").unwrap();
assert!(bob
.core()
.receive_targeted_transfer(
transfer.id.clone(),
occupied.path().to_string_lossy().into_owned(),
)
.is_err());
assert_eq!(
std::fs::read(occupied.path().join("payload.txt")).unwrap(),
b"keep me"
);
assert_eq!(
bob.core()
.get_targeted_transfer(transfer.id.clone())
.unwrap()
.unwrap()
.state,
TargetedTransferState::Interrupted
);
let clean = tempfile::tempdir().unwrap();
bob.core()
.resume_targeted_transfer(transfer.id, clean.path().to_string_lossy().into_owned())
.unwrap();
assert_eq!(
std::fs::read(clean.path().join("payload.txt")).unwrap(),
b"new payload"
);
}
#[test]
@@ -1258,6 +1384,131 @@ struct MemoryOutputSink {
files: Mutex<std::collections::HashMap<String, Vec<u8>>>,
}
#[derive(Default)]
struct FailingFinishSink {
starts: std::sync::atomic::AtomicUsize,
finishes: std::sync::atomic::AtomicUsize,
aborts: std::sync::atomic::AtomicUsize,
}
#[derive(Default)]
struct GatedSink {
gate: (Mutex<bool>, std::sync::Condvar),
entered: std::sync::atomic::AtomicBool,
finishes: std::sync::atomic::AtomicUsize,
aborts: std::sync::atomic::AtomicUsize,
}
impl GatedSink {
fn wait_until_entered(&self) {
let started = Instant::now();
while !self.entered.load(std::sync::atomic::Ordering::SeqCst) {
assert!(started.elapsed() < Duration::from_secs(10));
std::thread::sleep(Duration::from_millis(10));
}
}
fn release(&self) {
let (lock, wake) = &self.gate;
*lock.lock().unwrap() = true;
wake.notify_all();
}
}
impl ReceiveOutputSink for GatedSink {
fn start_file(&self, _relative_path: String) -> Result<(), VnidropError> {
Ok(())
}
fn write_chunk(&self, _relative_path: String, _bytes: Vec<u8>) -> Result<(), VnidropError> {
self.entered
.store(true, std::sync::atomic::Ordering::SeqCst);
let (lock, wake) = &self.gate;
let mut released = lock.lock().unwrap();
while !*released {
released = wake.wait(released).unwrap();
}
Ok(())
}
fn finish_file(&self, _relative_path: String) -> Result<(), VnidropError> {
self.finishes
.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Ok(())
}
fn abort_file(&self, _relative_path: String, _reason: String) -> Result<(), VnidropError> {
self.aborts
.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Ok(())
}
}
impl ReceiveOutputSinkV2 for GatedSink {
fn start_file(&self, relative_path: String) -> Result<(), VnidropError> {
ReceiveOutputSink::start_file(self, relative_path)
}
fn write_chunk(&self, relative_path: String, bytes: Vec<u8>) -> Result<(), VnidropError> {
ReceiveOutputSink::write_chunk(self, relative_path, bytes)
}
fn finish_file(&self, relative_path: String) -> Result<PublishedOutput, VnidropError> {
ReceiveOutputSink::finish_file(self, relative_path.clone())?;
Ok(PublishedOutput {
locator_kind: ReceivedLocatorKind::FilesystemPath,
locator: format!("memory://{relative_path}"),
})
}
fn abort_file(&self, relative_path: String, reason: String) -> Result<(), VnidropError> {
ReceiveOutputSink::abort_file(self, relative_path, reason)
}
}
impl ReceiveOutputSink for FailingFinishSink {
fn start_file(&self, _relative_path: String) -> Result<(), VnidropError> {
self.starts
.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Ok(())
}
fn write_chunk(&self, _relative_path: String, _bytes: Vec<u8>) -> Result<(), VnidropError> {
Ok(())
}
fn finish_file(&self, _relative_path: String) -> Result<(), VnidropError> {
self.finishes
.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Err(VnidropError::filesystem(anyhow::anyhow!("finish failed")))
}
fn abort_file(&self, _relative_path: String, _reason: String) -> Result<(), VnidropError> {
self.aborts
.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Ok(())
}
}
impl ReceiveOutputSinkV2 for FailingFinishSink {
fn start_file(&self, relative_path: String) -> Result<(), VnidropError> {
ReceiveOutputSink::start_file(self, relative_path)
}
fn write_chunk(&self, relative_path: String, bytes: Vec<u8>) -> Result<(), VnidropError> {
ReceiveOutputSink::write_chunk(self, relative_path, bytes)
}
fn finish_file(&self, relative_path: String) -> Result<PublishedOutput, VnidropError> {
ReceiveOutputSink::finish_file(self, relative_path)?;
unreachable!()
}
fn abort_file(&self, relative_path: String, reason: String) -> Result<(), VnidropError> {
ReceiveOutputSink::abort_file(self, relative_path, reason)
}
}
impl MemoryOutputSink {
fn file(&self, relative_path: &str) -> Vec<u8> {
self.files.lock().unwrap()[relative_path].clone()
@@ -1350,6 +1601,86 @@ fn targeted_receive_and_resume_through_output_sinks() {
);
}
#[test]
fn targeted_sink_finish_failure_is_the_only_terminal_callback() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 11_075);
for use_v2 in [false, true] {
let transfer = approve_one(&alice, &bob, b"sink failure", "sink.txt");
let sink = Arc::new(FailingFinishSink::default());
let result = if use_v2 {
bob.core()
.receive_targeted_transfer_with_output_sink_v2(transfer.id, sink.clone())
} else {
bob.core()
.receive_targeted_transfer_with_output_sink(transfer.id, sink.clone())
};
assert!(result.is_err());
assert_eq!(sink.starts.load(std::sync::atomic::Ordering::SeqCst), 1);
assert_eq!(sink.finishes.load(std::sync::atomic::Ordering::SeqCst), 1);
assert_eq!(sink.aborts.load(std::sync::atomic::Ordering::SeqCst), 0);
}
}
#[test]
fn targeted_receive_rejects_a_concurrent_second_pull() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 11_076);
let transfer = approve_one(&alice, &bob, b"concurrent pull", "sink.txt");
let sink = Arc::new(GatedSink::default());
let bob_core = bob.core();
let transfer_id = transfer.id.clone();
let sink_for_thread = sink.clone();
let receive = std::thread::spawn(move || {
bob_core.receive_targeted_transfer_with_output_sink(transfer_id, sink_for_thread)
});
sink.wait_until_entered();
let second = bob.core().resume_targeted_transfer(
transfer.id,
tempfile::tempdir()
.unwrap()
.path()
.to_string_lossy()
.into_owned(),
);
assert!(matches!(
second,
Err(VnidropError::InvalidTransition { .. })
));
sink.release();
receive.join().unwrap().unwrap();
}
#[test]
fn targeted_cancel_aborts_each_sink_exactly_once() {
for use_v2 in [false, true] {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, if use_v2 { 11_078 } else { 11_077 });
let transfer = approve_one(&alice, &bob, b"cancel sink", "sink.txt");
let sink = Arc::new(GatedSink::default());
let bob_core = bob.core();
let transfer_id = transfer.id.clone();
let sink_for_thread = sink.clone();
let receive = std::thread::spawn(move || {
if use_v2 {
bob_core.receive_targeted_transfer_with_output_sink_v2(transfer_id, sink_for_thread)
} else {
bob_core.receive_targeted_transfer_with_output_sink(transfer_id, sink_for_thread)
}
});
sink.wait_until_entered();
bob.core().cancel_targeted_transfer(transfer.id).unwrap();
sink.release();
assert!(receive.join().unwrap().is_err());
assert_eq!(sink.finishes.load(std::sync::atomic::Ordering::SeqCst), 0);
assert_eq!(sink.aborts.load(std::sync::atomic::Ordering::SeqCst), 1);
}
}
#[test]
fn decline_returns_typed_declined_outcome() {
let alice = ProtectedNode::new();

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

@@ -14,7 +14,7 @@ fn saved_device_protocols_are_explicitly_experimental_and_versioned() {
ExperimentalSavedDeviceCapabilities {
domain_contract_version: 1,
relationship_protocol_version: 1,
targeted_transfer_protocol_version: 1,
targeted_transfer_protocol_version: 2,
}
);
}