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

fix(core): report receiver failures to sender

Browse Source
This commit is contained in:
Hammed Abass
2026-07-24 14:52:09 +02:00
parent 0448137d84
commit 425500ecf2
25 changed files with 430 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
crates/vnidrop/src/approval.rs
View File

@@ -8,7 +8,10 @@ use uuid::Uuid;
use crate::{
access_policy::{AccessPolicy, APPROVAL_SESSION_TTL_MS},
event_hub::EventHub,
handshake::{DeliveryReceipt, DeliveryReceiptResponse, HandshakeResponse, RequestTransfer},
handshake::{
DeliveryFailureReceipt, DeliveryReceipt, DeliveryReceiptResponse, HandshakeResponse,
RequestTransfer,
},
repository::{ReceiverRequestInsert, Repository},
transfer_state::ReceiverRequestStatus,
util::now_ms,
@@ -70,6 +73,46 @@ impl ApprovalService {
}
}
pub(crate) async fn fail_delivery(
&self,
remote_endpoint_id: String,
receipt: DeliveryFailureReceipt,
) -> DeliveryReceiptResponse {
let token_hash = receipt_token_hash(&receipt.token);
match self
.repository
.fail_receiver_delivery(
&receipt.request_id,
receipt.transfer_id,
&remote_endpoint_id,
&token_hash,
&receipt.reason,
)
.await
{
Ok(()) => {
self.event_hub.emit_transfer(
receipt.transfer_id,
"send",
"delivery",
"receiver-failed",
json!({
"request_id": receipt.request_id,
"remote_endpoint_id": remote_endpoint_id,
"reason": receipt.reason,
}),
);
DeliveryReceiptResponse::Recorded
}
Err(error) => {
tracing::warn!(%error, "rejected receiver delivery failure");
DeliveryReceiptResponse::Rejected {
reason: "invalid-receipt".to_string(),
}
}
}
}
pub(crate) fn new(
repository: Repository,
event_hub: Arc<EventHub>,

25
crates/vnidrop/src/handshake.rs
View File

@@ -72,6 +72,14 @@ impl ProtocolHandler for HandshakeService {
.await;
let _ = tx.send(response).await;
}
HandshakeMessage::ReportDeliveryFailure(message) => {
let WithChannels { inner, tx, .. } = message;
let response = self
.approval
.fail_delivery(remote_endpoint_id.clone(), inner)
.await;
let _ = tx.send(response).await;
}
}
}
@@ -109,6 +117,13 @@ impl HandshakeClient {
) -> Result<DeliveryReceiptResponse, irpc::Error> {
self.inner.rpc(receipt).await
}
pub(crate) async fn report_delivery_failure(
&self,
receipt: DeliveryFailureReceipt,
) -> Result<DeliveryReceiptResponse, irpc::Error> {
self.inner.rpc(receipt).await
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -140,6 +155,14 @@ pub(crate) struct DeliveryReceipt {
pub(crate) token: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct DeliveryFailureReceipt {
pub(crate) request_id: String,
pub(crate) transfer_id: u64,
pub(crate) token: String,
pub(crate) reason: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub(crate) enum DeliveryReceiptResponse {
Recorded,
@@ -153,4 +176,6 @@ enum HandshakeProtocol {
RequestTransfer(RequestTransfer),
#[rpc(tx=oneshot::Sender<DeliveryReceiptResponse>)]
ReportDelivery(DeliveryReceipt),
#[rpc(tx=oneshot::Sender<DeliveryReceiptResponse>)]
ReportDeliveryFailure(DeliveryFailureReceipt),
}

110
crates/vnidrop/src/repository.rs
View File

@@ -20,7 +20,7 @@ use crate::{
util::now_ms,
};
const SCHEMA_VERSION: i64 = 6;
const SCHEMA_VERSION: i64 = 7;
#[derive(Debug, Clone)]
pub(crate) struct Repository {
@@ -86,6 +86,7 @@ pub(crate) struct PendingDeliveryReceipt {
pub(crate) request_id: String,
pub(crate) sender_transfer_id: u64,
pub(crate) token: String,
pub(crate) failure_reason: Option<String>,
}
pub(crate) struct PendingDeliveryReceiptInsert<'a> {
@@ -94,6 +95,7 @@ pub(crate) struct PendingDeliveryReceiptInsert<'a> {
pub(crate) request_id: &'a str,
pub(crate) sender_transfer_id: u64,
pub(crate) token: &'a str,
pub(crate) failure_reason: Option<&'a str>,
}
impl Repository {
@@ -294,12 +296,24 @@ impl Repository {
sender_blob_ticket TEXT NOT NULL,
sender_transfer_id INTEGER NOT NULL,
token TEXT NOT NULL,
failure_reason TEXT,
created_at INTEGER NOT NULL
);
"#,
)
.execute(&self.pool)
.await?;
let receipt_columns = sqlx::query("PRAGMA table_info(pending_delivery_receipts)")
.fetch_all(&self.pool)
.await?;
if !receipt_columns
.iter()
.any(|row| row.get::<String, _>(1) == "failure_reason")
{
sqlx::query("ALTER TABLE pending_delivery_receipts ADD COLUMN failure_reason TEXT")
.execute(&self.pool)
.await?;
}
sqlx::query(&format!("PRAGMA user_version = {SCHEMA_VERSION}"))
.execute(&self.pool)
@@ -561,13 +575,14 @@ impl Repository {
r#"
INSERT INTO pending_delivery_receipts (
request_id, local_transfer_id, sender_blob_ticket,
sender_transfer_id, token, created_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6)
sender_transfer_id, token, failure_reason, created_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
ON CONFLICT(request_id) DO UPDATE SET
local_transfer_id = excluded.local_transfer_id,
sender_blob_ticket = excluded.sender_blob_ticket,
sender_transfer_id = excluded.sender_transfer_id,
token = excluded.token
token = excluded.token,
failure_reason = excluded.failure_reason
"#,
)
.bind(receipt.request_id)
@@ -575,6 +590,7 @@ impl Repository {
.bind(receipt.sender_blob_ticket)
.bind(to_db_id(receipt.sender_transfer_id)?)
.bind(receipt.token)
.bind(receipt.failure_reason)
.bind(now_ms())
.execute(&mut *transaction)
.await?;
@@ -582,13 +598,46 @@ impl Repository {
Ok(())
}
pub(crate) async fn queue_failed_delivery_receipt(
&self,
receipt: PendingDeliveryReceiptInsert<'_>,
) -> Result<()> {
let Some(reason) = receipt.failure_reason else {
anyhow::bail!("failed delivery receipt requires a reason");
};
sqlx::query(
r#"
INSERT INTO pending_delivery_receipts (
request_id, local_transfer_id, sender_blob_ticket,
sender_transfer_id, token, failure_reason, created_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
ON CONFLICT(request_id) DO UPDATE SET
local_transfer_id = excluded.local_transfer_id,
sender_blob_ticket = excluded.sender_blob_ticket,
sender_transfer_id = excluded.sender_transfer_id,
token = excluded.token,
failure_reason = excluded.failure_reason
"#,
)
.bind(receipt.request_id)
.bind(to_db_id(receipt.local_transfer_id)?)
.bind(receipt.sender_blob_ticket)
.bind(to_db_id(receipt.sender_transfer_id)?)
.bind(receipt.token)
.bind(reason)
.bind(now_ms())
.execute(&self.pool)
.await?;
Ok(())
}
pub(crate) async fn list_pending_delivery_receipts(
&self,
) -> Result<Vec<PendingDeliveryReceipt>> {
let rows = sqlx::query(
r#"
SELECT local_transfer_id, sender_blob_ticket, request_id,
sender_transfer_id, token
sender_transfer_id, token, failure_reason
FROM pending_delivery_receipts
ORDER BY created_at ASC
"#,
@@ -603,6 +652,7 @@ impl Repository {
request_id: row.get("request_id"),
sender_transfer_id: row.get::<i64, _>("sender_transfer_id") as u64,
token: row.get("token"),
failure_reason: row.get("failure_reason"),
})
.collect())
}
@@ -883,6 +933,56 @@ impl Repository {
}
}
pub(crate) async fn fail_receiver_delivery(
&self,
id: &str,
transfer_id: u64,
remote_endpoint_id: &str,
token_hash: &str,
reason: &str,
) -> Result<()> {
let result = sqlx::query(
r#"
UPDATE receiver_requests
SET status = 'failed', reason = ?1
WHERE id = ?2 AND transfer_id = ?3 AND remote_endpoint_id = ?4 AND receipt_token_hash = ?5
AND status = 'accepted'
"#,
)
.bind(reason)
.bind(id)
.bind(to_db_id(transfer_id)?)
.bind(remote_endpoint_id)
.bind(token_hash)
.execute(&self.pool)
.await?;
if result.rows_affected() == 1 {
return Ok(());
}
let already_recorded = sqlx::query(
r#"
SELECT EXISTS(
SELECT 1 FROM receiver_requests
WHERE id = ?1 AND transfer_id = ?2 AND remote_endpoint_id = ?3
AND receipt_token_hash = ?4 AND status = 'failed'
)
"#,
)
.bind(id)
.bind(to_db_id(transfer_id)?)
.bind(remote_endpoint_id)
.bind(token_hash)
.fetch_one(&self.pool)
.await?
.get::<i64, _>(0)
!= 0;
if already_recorded {
Ok(())
} else {
anyhow::bail!("delivery failure did not match an accepted receiver request")
}
}
pub(crate) async fn expire_pending_receiver_requests(&self, reason: &str) -> Result<u64> {
let result = sqlx::query(
r#"

34
crates/vnidrop/src/runtime/delivery.rs
View File

@@ -4,7 +4,9 @@ use serde_json::json;
use super::{filter_peer_addr_for_relay_mode, CoreInner};
use crate::{
handshake::{DeliveryReceipt, DeliveryReceiptResponse, HandshakeService},
handshake::{
DeliveryFailureReceipt, DeliveryReceipt, DeliveryReceiptResponse, HandshakeService,
},
repository::PendingDeliveryReceipt,
ticket::parse_persisted_sender_address,
};
@@ -96,13 +98,29 @@ impl CoreInner {
}
};
let client = HandshakeService::client(self.endpoint.clone(), sender_addr);
let receipt = DeliveryReceipt {
request_id: pending.request_id.clone(),
transfer_id: pending.sender_transfer_id,
token: pending.token,
};
match tokio::time::timeout(DELIVERY_RECEIPT_TIMEOUT, client.report_delivery(receipt)).await
{
let request_id = pending.request_id.clone();
let response = tokio::time::timeout(DELIVERY_RECEIPT_TIMEOUT, async {
if let Some(reason) = pending.failure_reason {
client
.report_delivery_failure(DeliveryFailureReceipt {
request_id,
transfer_id: pending.sender_transfer_id,
token: pending.token,
reason,
})
.await
} else {
client
.report_delivery(DeliveryReceipt {
request_id,
transfer_id: pending.sender_transfer_id,
token: pending.token,
})
.await
}
})
.await;
match response {
Ok(Ok(DeliveryReceiptResponse::Recorded)) => {
if let Err(error) = self
.repository

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

@@ -1,7 +1,7 @@
use std::{
io,
path::{Path, PathBuf},
sync::Arc,
sync::{Arc, Mutex},
};
use anyhow::{Context, Result};
@@ -222,6 +222,10 @@ impl CoreInner {
self.persist_receive_start(transfer_id, &parsed, receiver_name.as_deref())
.await
.map_err(VnidropError::repository)?;
let persisted_sender_address =
encode_persisted_sender_address(parsed.blob_ticket.addr())
.context("failed to encode sender address for delivery receipt")?;
let delivery_receipt = Arc::new(Mutex::new(None));
// Cancellation is cooperative: it stops our receive future and marks
// local state while lower-level Iroh work unwinds naturally.
let (shutdown_tx, mut shutdown_rx) = oneshot::channel();
@@ -237,7 +241,14 @@ impl CoreInner {
);
let (result, cancelled) = tokio::select! {
result = self.receive_inner(transfer_id, parsed, target, receiver_name) => {
result = self.receive_inner(
transfer_id,
parsed,
target,
receiver_name,
persisted_sender_address.clone(),
delivery_receipt.clone(),
) => {
(result.map_err(VnidropError::transfer), false)
},
_ = &mut shutdown_rx => (Err(VnidropError::cancelled("transfer cancelled")), true),
@@ -267,6 +278,34 @@ impl CoreInner {
)
.await;
}
let receipt = delivery_receipt.lock().expect("delivery_receipt").take();
if let Some(receipt) = receipt {
let reason = if cancelled { "cancelled" } else { error.code() };
match self
.repository
.queue_failed_delivery_receipt(PendingDeliveryReceiptInsert {
local_transfer_id: transfer_id,
sender_blob_ticket: &persisted_sender_address,
request_id: &receipt.request_id,
sender_transfer_id: receipt.transfer_id,
token: &receipt.token,
failure_reason: Some(reason),
})
.await
{
Ok(()) => self.delivery_receipt_notify.notify_one(),
Err(queue_error) => {
tracing::warn!(%queue_error, "failed to queue delivery failure receipt");
self.emit_transfer(
transfer_id,
"receive",
"delivery",
"receipt-failed",
json!({ "reason": queue_error.to_string() }),
);
}
}
}
}
result.map_err(anyhow::Error::new)
}
@@ -277,6 +316,8 @@ impl CoreInner {
parsed: ParsedTransferTicket,
target: ReceiveTarget,
receiver_name: Option<String>,
persisted_sender_address: String,
pending_delivery_receipt: Arc<Mutex<Option<DeliveryReceipt>>>,
) -> Result<()> {
if let ReceiveTarget::Directory(output_dir) = &target {
tokio::fs::create_dir_all(output_dir)
@@ -284,8 +325,6 @@ impl CoreInner {
.map_err(VnidropError::filesystem)?;
}
let sender_addr = parsed.blob_ticket.addr().clone();
let persisted_sender_address = encode_persisted_sender_address(&sender_addr)
.context("failed to encode sender address for delivery receipt")?;
self.emit_transfer(transfer_id, "receive", "network", "connecting", json!({}));
// Every VniDrop ticket carries metadata and must complete the handshake.
@@ -297,6 +336,9 @@ impl CoreInner {
receiver_name.as_deref(),
)
.await?;
*pending_delivery_receipt
.lock()
.expect("pending_delivery_receipt") = Some(delivery_receipt.clone());
let connection = self
.endpoint
.connect(sender_addr.clone(), iroh_blobs::ALPN)
@@ -373,9 +415,14 @@ impl CoreInner {
request_id: &delivery_receipt.request_id,
sender_transfer_id: delivery_receipt.transfer_id,
token: &delivery_receipt.token,
failure_reason: None,
})
.await
.map_err(VnidropError::repository)?;
pending_delivery_receipt
.lock()
.expect("pending_delivery_receipt")
.take();
drop(download_tag);
self.emit_transfer(transfer_id, "receive", "lifecycle", "done", json!({}));
self.delivery_receipt_notify.notify_one();

67
crates/vnidrop/src/tests/repository.rs
View File

@@ -66,7 +66,7 @@ async fn received_artifacts_survive_history_deletion() {
async fn persists_transfers_and_events_across_reopen() {
let temp = tempfile::tempdir().unwrap();
let repository = Repository::open(temp.path()).await.unwrap();
assert_eq!(repository.schema_version().await.unwrap(), 6);
assert_eq!(repository.schema_version().await.unwrap(), 7);
repository
.insert_transfer(transfer(
7,
@@ -135,6 +135,7 @@ async fn receive_completion_persists_delivery_receipt_until_recorded() {
request_id: "request-93",
sender_transfer_id: 39,
token: "receipt-token",
failure_reason: None,
})
.await
.unwrap();
@@ -221,6 +222,68 @@ async fn receiver_request_can_only_be_resolved_once() {
assert!(requests[0].completed_at.is_some());
}
#[tokio::test]
async fn authenticated_delivery_failure_marks_accepted_receiver_failed() {
let temp = tempfile::tempdir().unwrap();
let repository = Repository::open(temp.path()).await.unwrap();
repository
.insert_receiver_request(ReceiverRequestInsert {
id: "request-failed",
transfer_id: 78,
remote_endpoint_id: "node-a",
transfer_name: "demo",
receiver_name: Some("receiver"),
receiver_device_name: None,
app_version: "0.1.0",
})
.await
.unwrap();
repository
.update_receiver_request_status("request-failed", ReceiverRequestStatus::Accepted, None)
.await
.unwrap();
repository
.set_receiver_receipt_token("request-failed", "token-hash")
.await
.unwrap();
repository
.fail_receiver_delivery(
"request-failed",
78,
"node-a",
"token-hash",
"destination_exists",
)
.await
.unwrap();
repository
.fail_receiver_delivery(
"request-failed",
78,
"node-a",
"token-hash",
"destination_exists",
)
.await
.unwrap();
assert!(repository
.fail_receiver_delivery(
"request-failed",
78,
"node-b",
"token-hash",
"destination_exists",
)
.await
.is_err());
let requests = repository.list_receiver_requests(78).await.unwrap();
assert_eq!(requests.len(), 1);
assert_eq!(requests[0].status, "failed");
assert_eq!(requests[0].reason.as_deref(), Some("destination_exists"));
}
#[tokio::test]
async fn startup_expiration_is_idempotent_for_pending_requests() {
let temp = tempfile::tempdir().unwrap();
@@ -582,7 +645,7 @@ async fn migrates_schema_v2_identity_without_losing_transfer() {
pool.close().await;
let repository = Repository::open(temp.path()).await.unwrap();
assert_eq!(repository.schema_version().await.unwrap(), 6);
assert_eq!(repository.schema_version().await.unwrap(), 7);
let stored = repository.list_transfers().await.unwrap().remove(0);
assert_eq!(stored.transfer_id, 7);
assert_eq!(stored.local_id, "legacy-7-send");

1
crates/vnidrop/src/tests/runtime.rs
View File

@@ -164,6 +164,7 @@ fn startup_processes_persisted_delivery_receipts() {
request_id: "request-94",
sender_transfer_id: 49,
token: "receipt-token",
failure_reason: None,
})
.await
.unwrap();

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

@@ -87,6 +87,7 @@ pub(crate) enum ReceiverRequestStatus {
Refused,
Expired,
Completed,
Failed,
}
impl ReceiverRequestStatus {
@@ -97,6 +98,7 @@ impl ReceiverRequestStatus {
Self::Refused => "refused",
Self::Expired => "expired",
Self::Completed => "completed",
Self::Failed => "failed",
}
}
}
@@ -111,6 +113,7 @@ impl TryFrom<&str> for ReceiverRequestStatus {
"refused" => Ok(Self::Refused),
"expired" => Ok(Self::Expired),
"completed" => Ok(Self::Completed),
"failed" => Ok(Self::Failed),
_ => bail!("unknown receiver request status: {value}"),
}
}

31
crates/vnidrop/tests/transfer.rs
View File

@@ -24,6 +24,30 @@ fn wait_for_sender_transfer_event(sender: &TestNode, transfer_id: u64, kind: &st
}
}
fn wait_for_receiver_status(
sender: &TestNode,
transfer_id: u64,
status: &str,
) -> vnidrop::ReceiverRequest {
let started = Instant::now();
loop {
if let Some(request) = sender
.core
.list_receiver_requests(transfer_id)
.unwrap()
.into_iter()
.find(|request| request.status == status)
{
return request;
}
assert!(
started.elapsed() < Duration::from_secs(5),
"timed out waiting for receiver status {status}"
);
std::thread::sleep(Duration::from_millis(10));
}
}
#[test]
fn transfers_file_between_two_cores() {
let source_dir = tempfile::tempdir().unwrap();
@@ -162,4 +186,11 @@ fn receive_refuses_to_overwrite_existing_destination() {
&& event.kind == "failed"
&& event.data_json.contains("\"code\":\"destination_exists\"")
}));
let failed = wait_for_receiver_status(&sender, share.transfer_id, "failed");
assert_eq!(failed.reason.as_deref(), Some("destination_exists"));
assert!(sender.sink.events().iter().any(|event| {
event.transfer_id == Some(share.transfer_id)
&& event.phase == "delivery"
&& event.kind == "receiver-failed"
}));
}