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

feat(core): track delivery receipts and transfer deletion

Browse Source
This commit is contained in:
Hammed Abass
2026-07-11 09:30:56 +02:00
parent 1e6404abfa
commit dc862bc01c
11 changed files with 471 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@@ -5056,6 +5056,7 @@ version = "0.1.0"
dependencies = [
"anyhow",
"async-channel",
"blake3",
"bytes",
"data-encoding",
"futures",

1
crates/vnidrop/Cargo.toml
View File

@@ -11,6 +11,7 @@ crate-type = ["cdylib", "staticlib", "rlib"]
anyhow = "1.0.102"
async-channel = "2.5.0"
bytes = "1.11.1"
blake3 = "1.8.3"
data-encoding = "2.11.0"
futures = "0.3"
futures-lite = "2.6.1"

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

@@ -244,4 +244,5 @@ pub struct ReceiverRequest {
pub reason: Option<String>,
pub requested_at: i64,
pub responded_at: Option<i64>,
pub completed_at: Option<i64>,
}

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

@@ -8,7 +8,7 @@ use uuid::Uuid;
use crate::{
access_policy::AccessPolicy,
event_hub::EventHub,
handshake::{HandshakeResponse, RequestTransfer},
handshake::{DeliveryReceipt, DeliveryReceiptResponse, HandshakeResponse, RequestTransfer},
repository::{ReceiverRequestInsert, Repository},
transfer_state::ReceiverRequestStatus,
util::now_ms,
@@ -35,6 +35,41 @@ pub(crate) struct ApprovalService {
}
impl ApprovalService {
pub(crate) async fn complete_delivery(
&self,
remote_endpoint_id: String,
receipt: DeliveryReceipt,
) -> DeliveryReceiptResponse {
let token_hash = receipt_token_hash(&receipt.token);
match self
.repository
.complete_receiver_delivery(
&receipt.request_id,
receipt.transfer_id,
&remote_endpoint_id,
&token_hash,
)
.await
{
Ok(()) => {
self.event_hub.emit_transfer(
receipt.transfer_id,
"send",
"delivery",
"receiver-completed",
json!({ "request_id": receipt.request_id, "remote_endpoint_id": remote_endpoint_id }),
);
DeliveryReceiptResponse::Recorded
}
Err(error) => {
tracing::warn!(%error, "rejected receiver delivery receipt");
DeliveryReceiptResponse::Rejected {
reason: "invalid-receipt".to_string(),
}
}
}
}
pub(crate) fn new(
repository: Repository,
event_hub: Arc<EventHub>,
@@ -149,7 +184,41 @@ impl ApprovalService {
remote_endpoint_id: String,
request: RequestTransfer,
) -> HandshakeResponse {
let request_id = Uuid::new_v4().to_string();
if self
.repository
.insert_receiver_request(ReceiverRequestInsert {
id: &request_id,
transfer_id: request.transfer_id,
remote_endpoint_id: &remote_endpoint_id,
transfer_name: &request.transfer_name,
receiver_name: request.receiver_name.as_deref(),
receiver_device_name: request.receiver_device_name.as_deref(),
app_version: &request.app_version,
})
.await
.is_err()
|| self
.repository
.update_receiver_request_status(&request_id, ReceiverRequestStatus::Accepted, None)
.await
.is_err()
{
return self
.deny(request.transfer_id, remote_endpoint_id, "repository-error")
.await;
}
let token = Uuid::new_v4().to_string();
if self
.repository
.set_receiver_receipt_token(&request_id, &receipt_token_hash(&token))
.await
.is_err()
{
return self
.deny(request.transfer_id, remote_endpoint_id, "repository-error")
.await;
}
let expires_at = now_ms() + APPROVAL_TTL_MS;
self.event_hub.emit_transfer(
request.transfer_id,
@@ -161,7 +230,11 @@ impl ApprovalService {
"expires_at": expires_at,
}),
);
HandshakeResponse::Approved { token, expires_at }
HandshakeResponse::Approved {
request_id,
token,
expires_at,
}
}
async fn wait_for_sender_decision(
@@ -242,7 +315,21 @@ impl ApprovalService {
"expires_at": expires_at,
}),
);
HandshakeResponse::Approved { token, expires_at }
if let Err(error) = self
.repository
.set_receiver_receipt_token(&decision.request_id, &receipt_token_hash(&token))
.await
{
tracing::error!(%error, "failed to attach delivery receipt token");
return HandshakeResponse::Denied {
reason: "repository-error".to_string(),
};
}
HandshakeResponse::Approved {
request_id: decision.request_id,
token,
expires_at,
}
}
Ok(Ok(decision)) => {
self.deny(
@@ -290,3 +377,7 @@ impl ApprovalService {
HandshakeResponse::Denied { reason }
}
}
fn receipt_token_hash(token: &str) -> String {
blake3::hash(token.as_bytes()).to_hex().to_string()
}

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

@@ -24,7 +24,7 @@ impl fmt::Debug for HandshakeService {
}
impl HandshakeService {
pub(crate) const ALPN: &'static [u8] = b"/vnidrop/handshake/1";
pub(crate) const ALPN: &'static [u8] = b"/vnidrop/handshake/2";
pub(crate) fn new(approval: crate::approval::ApprovalService) -> Self {
Self { approval }
@@ -64,6 +64,14 @@ impl ProtocolHandler for HandshakeService {
let response = self.handle_request(remote_endpoint_id.clone(), inner).await;
let _ = tx.send(response).await;
}
HandshakeMessage::ReportDelivery(message) => {
let WithChannels { inner, tx, .. } = message;
let response = self
.approval
.complete_delivery(remote_endpoint_id.clone(), inner)
.await;
let _ = tx.send(response).await;
}
}
}
@@ -94,6 +102,13 @@ impl HandshakeClient {
})
.await
}
pub(crate) async fn report_delivery(
&self,
receipt: DeliveryReceipt,
) -> Result<DeliveryReceiptResponse, irpc::Error> {
self.inner.rpc(receipt).await
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -108,8 +123,27 @@ pub(crate) struct RequestTransfer {
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub(crate) enum HandshakeResponse {
Approved { token: String, expires_at: i64 },
Denied { reason: String },
Approved {
request_id: String,
token: String,
expires_at: i64,
},
Denied {
reason: String,
},
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct DeliveryReceipt {
pub(crate) request_id: String,
pub(crate) transfer_id: u64,
pub(crate) token: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub(crate) enum DeliveryReceiptResponse {
Recorded,
Rejected { reason: String },
}
#[rpc_requests(message = HandshakeMessage)]
@@ -117,4 +151,6 @@ pub(crate) enum HandshakeResponse {
enum HandshakeProtocol {
#[rpc(tx=oneshot::Sender<HandshakeResponse>)]
RequestTransfer(RequestTransfer),
#[rpc(tx=oneshot::Sender<DeliveryReceiptResponse>)]
ReportDelivery(DeliveryReceipt),
}

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

@@ -20,7 +20,7 @@ use crate::{
util::now_ms,
};
const SCHEMA_VERSION: i64 = 3;
const SCHEMA_VERSION: i64 = 4;
#[derive(Debug, Clone)]
pub(crate) struct Repository {
@@ -202,12 +202,34 @@ impl Repository {
reason TEXT,
requested_at INTEGER NOT NULL,
responded_at INTEGER
,receipt_token_hash TEXT
,completed_at INTEGER
);
"#,
)
.execute(&self.pool)
.await?;
let receiver_columns = sqlx::query("PRAGMA table_info(receiver_requests)")
.fetch_all(&self.pool)
.await?;
if !receiver_columns
.iter()
.any(|row| row.get::<String, _>(1) == "receipt_token_hash")
{
sqlx::query("ALTER TABLE receiver_requests ADD COLUMN receipt_token_hash TEXT")
.execute(&self.pool)
.await?;
}
if !receiver_columns
.iter()
.any(|row| row.get::<String, _>(1) == "completed_at")
{
sqlx::query("ALTER TABLE receiver_requests ADD COLUMN completed_at INTEGER")
.execute(&self.pool)
.await?;
}
sqlx::query(
"CREATE INDEX IF NOT EXISTS idx_receiver_requests_transfer_id ON receiver_requests(transfer_id, requested_at DESC);",
)
@@ -583,6 +605,70 @@ impl Repository {
Ok(())
}
pub(crate) async fn set_receiver_receipt_token(
&self,
id: &str,
token_hash: &str,
) -> Result<()> {
let result = sqlx::query(
"UPDATE receiver_requests SET receipt_token_hash = ?1 WHERE id = ?2 AND status = 'accepted'",
)
.bind(token_hash)
.bind(id)
.execute(&self.pool)
.await?;
require_one_changed(result.rows_affected(), "attach receiver receipt token")
}
pub(crate) async fn complete_receiver_delivery(
&self,
id: &str,
transfer_id: u64,
remote_endpoint_id: &str,
token_hash: &str,
) -> Result<()> {
let result = sqlx::query(
r#"
UPDATE receiver_requests
SET status = 'completed', completed_at = ?1
WHERE id = ?2 AND transfer_id = ?3 AND remote_endpoint_id = ?4 AND receipt_token_hash = ?5
AND status = 'accepted'
"#,
)
.bind(now_ms())
.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 = 'completed'
)
"#,
)
.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 receipt 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#"
@@ -606,7 +692,7 @@ impl Repository {
r#"
SELECT id, transfer_id, remote_endpoint_id, transfer_name,
receiver_name, receiver_device_name, app_version, status,
reason, requested_at, responded_at
reason, requested_at, responded_at, completed_at
FROM receiver_requests
WHERE transfer_id = ?1
ORDER BY requested_at DESC
@@ -652,6 +738,27 @@ impl Repository {
rows.into_iter().map(row_to_transfer).collect()
}
pub(crate) async fn delete_transfer(&self, transfer_id: u64) -> Result<()> {
self.maybe_fail_write()?;
let transfer_id = to_db_id(transfer_id)?;
let mut transaction = self.pool.begin().await?;
sqlx::query("DELETE FROM receiver_requests WHERE transfer_id = ?1")
.bind(transfer_id)
.execute(&mut *transaction)
.await?;
sqlx::query("DELETE FROM transfer_events WHERE transfer_id = ?1")
.bind(transfer_id)
.execute(&mut *transaction)
.await?;
let deleted = sqlx::query("DELETE FROM transfers WHERE transfer_id = ?1")
.bind(transfer_id)
.execute(&mut *transaction)
.await?;
require_one_changed(deleted.rows_affected(), "delete transfer")?;
transaction.commit().await?;
Ok(())
}
pub(crate) async fn list_events(
&self,
transfer_id: Option<u64>,
@@ -754,5 +861,6 @@ fn row_to_receiver_request(row: sqlx::sqlite::SqliteRow) -> ReceiverRequest {
reason: row.get("reason"),
requested_at: row.get("requested_at"),
responded_at: row.get("responded_at"),
completed_at: row.get("completed_at"),
}
}

118
crates/vnidrop/src/runtime.rs
View File

@@ -44,12 +44,12 @@ use crate::{
read_stream_from_blocking_reader, validated_relative_string, wait_for_writer,
write_stream_to_blocking_writer, AtomicOutputFile, TransferImport,
},
handshake::{HandshakeResponse, HandshakeService},
handshake::{DeliveryReceipt, DeliveryReceiptResponse, HandshakeResponse, HandshakeService},
logging::init_logging,
repository::{Repository, TransferUpsert},
secret::load_or_create_secret,
ticket::{parse_transfer_ticket_with_limits, ParsedTransferTicket, VnidropTicket},
transfer_state::{TransferDirection, TransferStatus},
transfer_state::{ReceiverRequestStatus, TransferDirection, TransferStatus},
util::{non_empty, unique_transfer_id},
};
@@ -244,6 +244,12 @@ impl VnidropCore {
.map_err(VnidropError::transfer)
}
pub fn delete_transfer(&self, transfer_id: u64) -> Result<(), VnidropError> {
self.runtime
.block_on(self.inner.delete_transfer(transfer_id))
.map_err(VnidropError::transfer)
}
pub fn set_transfer_access_mode(
&self,
transfer_id: u64,
@@ -725,20 +731,25 @@ impl CoreInner {
if let ReceiveTarget::Directory(output_dir) = &target {
tokio::fs::create_dir_all(output_dir).await?;
}
let sender_addr = parsed.blob_ticket.addr().clone();
self.emit_transfer(transfer_id, "receive", "network", "connecting", json!({}));
if let Some(metadata) = &parsed.metadata {
let delivery_receipt = if let Some(metadata) = &parsed.metadata {
Some(
self.request_transfer_approval(
transfer_id,
parsed.blob_ticket.addr().clone(),
sender_addr.clone(),
metadata,
receiver_name.as_deref(),
)
.await?;
}
.await?,
)
} else {
None
};
let connection = self
.endpoint
.connect(parsed.blob_ticket.addr().clone(), iroh_blobs::ALPN)
.connect(sender_addr.clone(), iroh_blobs::ALPN)
.await?;
self.emit_transfer(transfer_id, "receive", "network", "connected", json!({}));
@@ -801,6 +812,33 @@ impl CoreInner {
)
.await?;
self.emit_transfer(transfer_id, "receive", "lifecycle", "done", json!({}));
if let Some(receipt) = delivery_receipt {
let sender_transfer_id = receipt.transfer_id;
let client = HandshakeService::client(self.endpoint.clone(), sender_addr);
match client.report_delivery(receipt).await {
Ok(DeliveryReceiptResponse::Recorded) => self.emit_transfer(
transfer_id,
"receive",
"delivery",
"receipt-recorded",
json!({ "sender_transfer_id": sender_transfer_id }),
),
Ok(DeliveryReceiptResponse::Rejected { reason }) => self.emit_transfer(
transfer_id,
"receive",
"delivery",
"receipt-rejected",
json!({ "reason": reason }),
),
Err(error) => self.emit_transfer(
transfer_id,
"receive",
"delivery",
"receipt-failed",
json!({ "reason": error.to_string() }),
),
}
}
Ok(())
}
@@ -905,6 +943,58 @@ impl CoreInner {
anyhow::bail!("transfer not found")
}
async fn delete_transfer(&self, transfer_id: u64) -> Result<()> {
if self
.active_transfers
.lock()
.await
.contains_key(&transfer_id)
{
anyhow::bail!("an active transfer must finish or be cancelled before deletion");
}
let transfer = self
.repository
.list_transfers()
.await?
.into_iter()
.find(|transfer| transfer.transfer_id == transfer_id)
.ok_or_else(|| anyhow::anyhow!("transfer not found"))?;
// Revoke a live share durably before removing its history. If the
// subsequent delete fails, a restart must never expose it again.
if transfer.status == TransferStatus::Sharing.as_str() {
self.repository
.transition_transfer_status(
transfer_id,
TransferStatus::Sharing,
TransferStatus::Stopped,
)
.await?;
}
for request in self.repository.list_receiver_requests(transfer_id).await? {
if request.status == ReceiverRequestStatus::Requested.as_str() {
let _ = self
.approval
.respond(
request.id,
false,
Some("transfer deleted by sender".to_string()),
)
.await;
}
}
self.active_shares.lock().await.remove(&transfer_id);
self.hash_to_transfer
.lock()
.await
.retain(|_, id| *id != transfer_id);
self.access_policy.remove_transfer(transfer_id).await;
self.repository.delete_transfer(transfer_id).await
}
async fn set_transfer_access_mode(
&self,
transfer_id: u64,
@@ -948,7 +1038,7 @@ impl CoreInner {
addr: iroh::EndpointAddr,
metadata: &TransferMetadata,
receiver_name: Option<&str>,
) -> Result<()> {
) -> Result<DeliveryReceipt> {
self.emit_transfer(
local_transfer_id,
"receive",
@@ -966,7 +1056,11 @@ impl CoreInner {
.await
.map_err(|error| anyhow::anyhow!("handshake request failed: {error}"))?
{
HandshakeResponse::Approved { expires_at, .. } => {
HandshakeResponse::Approved {
request_id,
token,
expires_at,
} => {
self.emit_transfer(
local_transfer_id,
"receive",
@@ -977,7 +1071,11 @@ impl CoreInner {
"expires_at": expires_at,
}),
);
Ok(())
Ok(DeliveryReceipt {
request_id,
transfer_id: metadata.transfer_id,
token,
})
}
HandshakeResponse::Denied { reason } => {
anyhow::bail!("transfer request was denied by sender: {reason}")

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

@@ -27,7 +27,7 @@ fn transfer(
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(), 3);
assert_eq!(repository.schema_version().await.unwrap(), 4);
repository
.insert_transfer(transfer(
7,
@@ -96,6 +96,22 @@ async fn receiver_request_can_only_be_resolved_once() {
.update_receiver_request_status("request-1", ReceiverRequestStatus::Accepted, None)
.await
.unwrap();
repository
.set_receiver_receipt_token("request-1", "token-hash")
.await
.unwrap();
repository
.complete_receiver_delivery("request-1", 77, "node-a", "token-hash")
.await
.unwrap();
repository
.complete_receiver_delivery("request-1", 77, "node-a", "token-hash")
.await
.unwrap();
assert!(repository
.complete_receiver_delivery("request-1", 77, "node-b", "token-hash")
.await
.is_err());
assert!(repository
.update_receiver_request_status("request-1", ReceiverRequestStatus::Refused, Some("late"),)
@@ -108,9 +124,10 @@ async fn receiver_request_can_only_be_resolved_once() {
let requests = repository.list_receiver_requests(77).await.unwrap();
assert_eq!(requests.len(), 1);
assert_eq!(requests[0].status, "accepted");
assert_eq!(requests[0].status, "completed");
assert_eq!(requests[0].receiver_name.as_deref(), Some("receiver"));
assert!(requests[0].responded_at.is_some());
assert!(requests[0].completed_at.is_some());
}
#[tokio::test]
@@ -474,7 +491,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(), 3);
assert_eq!(repository.schema_version().await.unwrap(), 4);
let stored = repository.list_transfers().await.unwrap().remove(0);
assert_eq!(stored.transfer_id, 7);
assert_eq!(stored.local_id, "legacy-7-send");
@@ -511,6 +528,63 @@ async fn event_reads_respect_configured_history_limit() {
assert_eq!(events[0].id, "event-2");
assert_eq!(events[1].id, "event-1");
}
#[tokio::test]
async fn deleting_transfer_removes_related_history_transactionally() {
let temp = tempfile::tempdir().unwrap();
let repository = Repository::open(temp.path()).await.unwrap();
repository
.insert_transfer(transfer(
88,
TransferDirection::Send,
TransferStatus::Stopped,
))
.await
.unwrap();
repository
.insert_receiver_request(ReceiverRequestInsert {
id: "request-delete",
transfer_id: 88,
remote_endpoint_id: "receiver",
transfer_name: "demo",
receiver_name: None,
receiver_device_name: None,
app_version: "1.0",
})
.await
.unwrap();
repository
.insert_event(
&CoreEvent {
id: "event-delete".to_string(),
timestamp: 1,
scope: "transfer".to_string(),
transfer_id: Some(88),
direction: Some("send".to_string()),
phase: "test".to_string(),
kind: "created".to_string(),
data_json: "{}".to_string(),
},
500,
)
.await
.unwrap();
repository.delete_transfer(88).await.unwrap();
assert!(repository.list_transfers().await.unwrap().is_empty());
assert!(repository
.list_events(Some(88), 500)
.await
.unwrap()
.is_empty());
assert!(repository
.list_receiver_requests(88)
.await
.unwrap()
.is_empty());
assert!(repository.delete_transfer(88).await.is_err());
}
use std::str::FromStr;
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};

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

@@ -86,6 +86,7 @@ pub(crate) enum ReceiverRequestStatus {
Accepted,
Refused,
Expired,
Completed,
}
impl ReceiverRequestStatus {
@@ -95,6 +96,7 @@ impl ReceiverRequestStatus {
Self::Accepted => "accepted",
Self::Refused => "refused",
Self::Expired => "expired",
Self::Completed => "completed",
}
}
}
@@ -108,6 +110,7 @@ impl TryFrom<&str> for ReceiverRequestStatus {
"accepted" => Ok(Self::Accepted),
"refused" => Ok(Self::Refused),
"expired" => Ok(Self::Expired),
"completed" => Ok(Self::Completed),
_ => bail!("unknown receiver request status: {value}"),
}
}

16
crates/vnidrop/tests/approval.rs
View File

@@ -47,11 +47,14 @@ fn public_share_receives_without_sender_approval() {
std::fs::read(output_dir.path().join("public.txt")).unwrap(),
b"public content"
);
assert!(sender
let deliveries = sender
.core
.list_receiver_requests(share.transfer_id)
.unwrap()
.is_empty());
.unwrap();
assert_eq!(deliveries.len(), 1);
assert_eq!(deliveries[0].receiver_name.as_deref(), Some("Receiver"));
assert_eq!(deliveries[0].status, "completed");
assert!(deliveries[0].completed_at.is_some());
}
#[test]
@@ -93,6 +96,13 @@ fn approval_required_denies_then_allows_receiver() {
std::fs::read(allowed_output.path().join("private.txt")).unwrap(),
b"approved content"
);
let completed = sender
.core
.list_receiver_requests(share.transfer_id)
.unwrap();
assert!(completed
.iter()
.any(|request| request.status == "completed"));
}
#[test]

20
crates/vnidrop/tests/lifecycle.rs
View File

@@ -54,6 +54,26 @@ fn cancelling_share_updates_status_and_events() {
.any(|event| event.kind == "share-stopped"));
}
#[test]
fn deleting_share_revokes_it_and_removes_persisted_history() {
let source_dir = tempfile::tempdir().unwrap();
let core_dir = tempfile::tempdir().unwrap();
let source_path = source_dir.path().join("delete.txt");
std::fs::write(&source_path, b"delete me").unwrap();
let sender = CoreGuard::start(core_dir.path(), Arc::new(RecordingSink::default()));
let share = share_path(&sender, &source_path, 101, "delete.txt", false);
sender.delete_transfer(share.transfer_id).unwrap();
assert!(sender.list_transfers().unwrap().is_empty());
assert_eq!(sender.status().active_shares, 0);
drop(sender);
let restarted = CoreGuard::start(core_dir.path(), Arc::new(RecordingSink::default()));
assert!(restarted.list_transfers().unwrap().is_empty());
assert_eq!(restarted.status().active_shares, 0);
}
#[test]
fn persisted_share_is_recovered_and_can_be_stopped_after_restart() {
let source_dir = tempfile::tempdir().unwrap();