diff --git a/crates/vnidrop/src/api.rs b/crates/vnidrop/src/api.rs index b2470ed..f59f099 100644 --- a/crates/vnidrop/src/api.rs +++ b/crates/vnidrop/src/api.rs @@ -477,6 +477,29 @@ pub struct ContactSummary { pub can_send: bool, } +/// Outcome of sending straight to a remembered device. +#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)] +pub struct ContactSendResult { + pub share: ShareResult, + /// False when the device was not running: the transfer is held here and the + /// device collects it the next time it opens VniDrop. + pub delivered: bool, +} + +/// A transfer this device is holding until its target comes back online. +/// +/// Cancelling the underlying transfer withdraws it. +#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)] +pub struct HeldOfferSummary { + pub offer_id: String, + pub endpoint_id: String, + pub transfer_id: u64, + pub transfer_name: String, + pub file_count: u64, + pub total_bytes: u64, + pub created_at: i64, +} + /// A transfer a paired device is offering. /// /// The ticket is deliberately absent: it is a capability, and it is handed over diff --git a/crates/vnidrop/src/contacts.rs b/crates/vnidrop/src/contacts.rs index 61af7b3..c91c073 100644 --- a/crates/vnidrop/src/contacts.rs +++ b/crates/vnidrop/src/contacts.rs @@ -91,6 +91,27 @@ pub(crate) async fn ensure_schema(pool: &SqlitePool) -> Result<()> { .execute(pool) .await?; + sqlx::query( + r#" + CREATE TABLE IF NOT EXISTS held_offers ( + offer_id TEXT PRIMARY KEY, + endpoint_id TEXT NOT NULL, + transfer_id INTEGER NOT NULL, + ticket TEXT NOT NULL, + transfer_name TEXT NOT NULL, + sender_display_name TEXT, + file_count INTEGER NOT NULL, + total_bytes INTEGER NOT NULL, + created_at INTEGER NOT NULL + ); + "#, + ) + .execute(pool) + .await?; + sqlx::query("CREATE INDEX IF NOT EXISTS idx_held_offers_endpoint ON held_offers(endpoint_id);") + .execute(pool) + .await?; + sqlx::query( r#" CREATE TABLE IF NOT EXISTS blocked_endpoints ( @@ -105,6 +126,23 @@ pub(crate) async fn ensure_schema(pool: &SqlitePool) -> Result<()> { Ok(()) } +/// An offer that could not be delivered because the target was not running. +/// +/// Held on this device, not a server: the share stays here and the receiver +/// collects the ticket when its app next comes to the foreground. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct HeldOffer { + pub(crate) offer_id: String, + pub(crate) endpoint_id: String, + pub(crate) transfer_id: u64, + pub(crate) ticket: String, + pub(crate) transfer_name: String, + pub(crate) sender_display_name: Option, + pub(crate) file_count: u64, + pub(crate) total_bytes: u64, + pub(crate) created_at: i64, +} + /// Contacts, grants, and blocks over the shared repository pool. #[derive(Debug, Clone)] pub(crate) struct ContactStore { @@ -442,6 +480,84 @@ impl ContactStore { Ok(rows.into_iter().map(|row| row.get(0)).collect()) } + // -- held offers ------------------------------------------------------ + + pub(crate) async fn insert_held_offer(&self, offer: &HeldOffer) -> Result<()> { + sqlx::query( + r#" + INSERT INTO held_offers + (offer_id, endpoint_id, transfer_id, ticket, transfer_name, + sender_display_name, file_count, total_bytes, created_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) + "#, + ) + .bind(&offer.offer_id) + .bind(&offer.endpoint_id) + .bind(offer.transfer_id as i64) + .bind(&offer.ticket) + .bind(&offer.transfer_name) + .bind(offer.sender_display_name.as_deref()) + .bind(offer.file_count as i64) + .bind(offer.total_bytes as i64) + .bind(offer.created_at) + .execute(&self.pool) + .await?; + Ok(()) + } + + /// Offers waiting for one device to come and collect them. + pub(crate) async fn held_offers_for(&self, endpoint_id: &str) -> Result> { + let rows = sqlx::query( + r#" + SELECT offer_id, endpoint_id, transfer_id, ticket, transfer_name, + sender_display_name, file_count, total_bytes, created_at + FROM held_offers + WHERE endpoint_id = ?1 + ORDER BY created_at ASC + "#, + ) + .bind(endpoint_id) + .fetch_all(&self.pool) + .await?; + Ok(rows.into_iter().map(row_to_held_offer).collect()) + } + + pub(crate) async fn list_held_offers(&self) -> Result> { + let rows = sqlx::query( + r#" + SELECT offer_id, endpoint_id, transfer_id, ticket, transfer_name, + sender_display_name, file_count, total_bytes, created_at + FROM held_offers + ORDER BY created_at ASC + "#, + ) + .fetch_all(&self.pool) + .await?; + Ok(rows.into_iter().map(row_to_held_offer).collect()) + } + + /// Consumed once handed over, so a device polling twice is not offered the + /// same transfer again. + pub(crate) async fn delete_held_offers(&self, offer_ids: &[String]) -> Result<()> { + let mut tx = self.pool.begin().await?; + for offer_id in offer_ids { + sqlx::query("DELETE FROM held_offers WHERE offer_id = ?1") + .bind(offer_id) + .execute(&mut *tx) + .await?; + } + tx.commit().await?; + Ok(()) + } + + pub(crate) async fn delete_held_offers_for_transfer(&self, transfer_id: u64) -> Result<()> { + sqlx::query("DELETE FROM held_offers WHERE transfer_id = ?1") + .bind(transfer_id as i64) + .execute(&self.pool) + .await?; + Ok(()) + } + // -- maintenance ------------------------------------------------------ #[cfg(test)] @@ -469,6 +585,20 @@ impl ContactStore { } } +fn row_to_held_offer(row: sqlx::sqlite::SqliteRow) -> HeldOffer { + HeldOffer { + offer_id: row.get(0), + endpoint_id: row.get(1), + transfer_id: row.get::(2) as u64, + ticket: row.get(3), + transfer_name: row.get(4), + sender_display_name: row.get(5), + file_count: row.get::(6) as u64, + total_bytes: row.get::(7) as u64, + created_at: row.get(8), + } +} + fn row_to_issued_grant(row: sqlx::sqlite::SqliteRow) -> Result { let grant_id = GrantId::decode(row.get::(0).as_str())?; let secret = parse_secret(row.get::(1).as_str()) diff --git a/crates/vnidrop/src/lib.rs b/crates/vnidrop/src/lib.rs index 5937e85..92f7779 100644 --- a/crates/vnidrop/src/lib.rs +++ b/crates/vnidrop/src/lib.rs @@ -20,11 +20,12 @@ mod util; pub use api::{ clear_inactive_transfer_cache, default_core_limits, default_core_network_config, - ContactSummary, CoreEvent, CoreEventSink, CoreLimits, CoreNetworkConfig, CoreRelayMode, - CoreStorageUsage, GrantLifetimeSetting, IncomingOffer, PendingPairing, PublishedOutput, - ReceiveOutputSink, ReceiveOutputSinkV2, ReceivedArtifact, ReceivedLocatorKind, ReceiverRequest, - RuntimeStatus, ShareMetadataInput, ShareResult, ShareSource, SourceKind, StoredTransfer, - TicketInspection, TransferAccessMode, TransferMetadata, + ContactSendResult, ContactSummary, CoreEvent, CoreEventSink, CoreLimits, CoreNetworkConfig, + CoreRelayMode, CoreStorageUsage, GrantLifetimeSetting, HeldOfferSummary, IncomingOffer, + PendingPairing, PublishedOutput, ReceiveOutputSink, ReceiveOutputSinkV2, ReceivedArtifact, + ReceivedLocatorKind, ReceiverRequest, RuntimeStatus, ShareMetadataInput, ShareResult, + ShareSource, SourceKind, StoredTransfer, TicketInspection, TransferAccessMode, + TransferMetadata, }; pub use error::VnidropError; pub use runtime::VnidropCore; diff --git a/crates/vnidrop/src/offer.rs b/crates/vnidrop/src/offer.rs index 44d902c..7c0ec4d 100644 --- a/crates/vnidrop/src/offer.rs +++ b/crates/vnidrop/src/offer.rs @@ -108,6 +108,19 @@ impl OfferService { } } +impl OfferService { + /// Hand a device the offers this one is holding for it. + /// + /// Needs no grant proof: iroh has already authenticated the remote endpoint + /// key, and the only thing returned is what this device already decided to + /// send to precisely that endpoint. A stranger polling gets an empty list. + async fn handle_poll(&self, remote_endpoint_id: &str) -> PolledOffers { + PolledOffers { + offers: self.pairing.collect_held_offers(remote_endpoint_id).await, + } + } +} + impl ProtocolHandler for OfferService { /// Accepts inbound connections from paired peers. /// @@ -144,6 +157,11 @@ impl ProtocolHandler for OfferService { .await; let _ = tx.send(response).await; } + OfferMessage::PollOffers(message) => { + let WithChannels { tx, .. } = message; + let response = self.handle_poll(&remote_endpoint_id).await; + let _ = tx.send(response).await; + } OfferMessage::SubmitOffer(message) => { let WithChannels { inner, tx, .. } = message; let response = self @@ -165,6 +183,10 @@ pub(crate) struct OfferClient { } impl OfferClient { + pub(crate) async fn poll_offers(&self) -> Result { + self.inner.rpc(PollOffers).await + } + pub(crate) async fn request_challenge(&self) -> Result { Ok(self.inner.rpc(RequestChallenge).await?.challenge) } @@ -276,6 +298,27 @@ pub(crate) enum OfferResponse { Refused { reason: String }, } +/// Ask a device whether it is holding anything for this one. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct PollOffers; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct PolledOffers { + pub(crate) offers: Vec, +} + +/// An offer collected by polling rather than pushed. Carries the ticket because +/// the sender already decided to send it to this endpoint; the local user still +/// confirms before anything is fetched. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct PolledOffer { + pub(crate) ticket: String, + pub(crate) transfer_name: String, + pub(crate) sender_display_name: Option, + pub(crate) file_count: u64, + pub(crate) total_bytes: u64, +} + #[rpc_requests(message = OfferMessage)] #[derive(Debug, Serialize, Deserialize)] enum OfferProtocol { @@ -287,4 +330,6 @@ enum OfferProtocol { RevokeGrant(RevokeGrant), #[rpc(tx=oneshot::Sender)] SubmitOffer(SubmitOffer), + #[rpc(tx=oneshot::Sender)] + PollOffers(PollOffers), } diff --git a/crates/vnidrop/src/offer_inbox.rs b/crates/vnidrop/src/offer_inbox.rs index 58071be..ced0948 100644 --- a/crates/vnidrop/src/offer_inbox.rs +++ b/crates/vnidrop/src/offer_inbox.rs @@ -4,9 +4,10 @@ //! not the transfer. Accepting hands the ticket to the platform layer, which //! runs the ordinary receive with its own destination rules. //! -//! Nothing here is persisted. An offer is a live connection to a running app, -//! so a restart correctly loses it rather than resurrecting a prompt whose -//! sender is long gone. +//! Nothing in this inbox is persisted: a prompt belongs to a live connection, +//! so a restart correctly loses it rather than resurrecting one whose sender is +//! long gone. Offers the *sender* could not deliver are a different thing and +//! do persist — see `held_offers` in [`crate::contacts`]. use std::{collections::HashMap, sync::Arc, time::Duration}; @@ -158,6 +159,60 @@ impl OfferInbox { } } + /// Add an offer collected by polling. + /// + /// Unlike [`Self::submit`] there is no remote waiting on the answer: the + /// sender handed the ticket over and moved on, so this returns immediately. + pub(crate) async fn enqueue( + &self, + from_endpoint_id: String, + transfer_name: String, + sender_display_name: Option, + file_count: u64, + total_bytes: u64, + ticket: String, + ) -> bool { + let offer_id = uuid::Uuid::new_v4().to_string(); + { + let mut pending = self.pending.lock().await; + if pending.len() >= self.max_pending { + return false; + } + if pending + .values() + .any(|offer| offer.from_endpoint_id == from_endpoint_id) + { + return false; + } + pending.insert( + offer_id.clone(), + PendingOffer { + offer_id: offer_id.clone(), + from_endpoint_id: from_endpoint_id.clone(), + sender_display_name: sender_display_name.clone(), + transfer_name: transfer_name.clone(), + file_count, + total_bytes, + received_at: now_ms(), + ticket, + }, + ); + } + self.event_hub.emit_endpoint( + "offer", + "offer-collected", + json!({ + "offer_id": offer_id, + "from_endpoint_id": from_endpoint_id, + "sender_display_name": sender_display_name, + "transfer_name": transfer_name, + "file_count": file_count, + "total_bytes": total_bytes, + }), + ); + true + } + pub(crate) async fn list(&self) -> Vec { self.pending.lock().await.values().cloned().collect() } diff --git a/crates/vnidrop/src/pairing.rs b/crates/vnidrop/src/pairing.rs index 71e2749..b793d3d 100644 --- a/crates/vnidrop/src/pairing.rs +++ b/crates/vnidrop/src/pairing.rs @@ -16,7 +16,7 @@ use crate::{ grant::{ Challenge, GrantLifetime, GrantProof, GrantRejection, GrantSecret, HeldGrant, IssuedGrant, }, - offer::{DeliverGrant, GrantDeliveryResponse, RevocationResponse, RevokeGrant}, + offer::{DeliverGrant, GrantDeliveryResponse, PolledOffer, RevocationResponse, RevokeGrant}, util::now_ms, }; @@ -290,6 +290,42 @@ impl PairingService { Ok(grant) } + /// Held offers addressed to `endpoint_id`, consumed as they are handed over. + /// + /// Deleting on delivery is what keeps a device that polls twice from being + /// offered the same transfer again. + pub(crate) async fn collect_held_offers(&self, endpoint_id: &str) -> Vec { + if self.contacts.is_blocked(endpoint_id).await.unwrap_or(false) { + return Vec::new(); + } + let Ok(held) = self.contacts.held_offers_for(endpoint_id).await else { + return Vec::new(); + }; + if held.is_empty() { + return Vec::new(); + } + let ids: Vec = held.iter().map(|offer| offer.offer_id.clone()).collect(); + if let Err(error) = self.contacts.delete_held_offers(&ids).await { + // Handing the same offer over twice is worse than not handing it + // over at all, so a failed consume aborts the delivery. + tracing::warn!(%error, "failed to consume held offers"); + return Vec::new(); + } + self.emit( + "held-offers-collected", + json!({ "peer_endpoint_id": endpoint_id, "count": held.len() }), + ); + held.into_iter() + .map(|offer| PolledOffer { + ticket: offer.ticket, + transfer_name: offer.transfer_name, + sender_display_name: offer.sender_display_name, + file_count: offer.file_count, + total_bytes: offer.total_bytes, + }) + .collect() + } + /// Validate a proof a peer presented, and push the idle deadline forward. /// /// The grant record is ours: we issued it, so we are the only party that diff --git a/crates/vnidrop/src/repository.rs b/crates/vnidrop/src/repository.rs index 00e0cd7..e42e48e 100644 --- a/crates/vnidrop/src/repository.rs +++ b/crates/vnidrop/src/repository.rs @@ -21,7 +21,7 @@ use crate::{ util::now_ms, }; -const SCHEMA_VERSION: i64 = 8; +const SCHEMA_VERSION: i64 = 9; #[derive(Debug, Clone)] pub(crate) struct Repository { diff --git a/crates/vnidrop/src/runtime/contacts.rs b/crates/vnidrop/src/runtime/contacts.rs index 3891101..f0d1128 100644 --- a/crates/vnidrop/src/runtime/contacts.rs +++ b/crates/vnidrop/src/runtime/contacts.rs @@ -10,12 +10,13 @@ use anyhow::{Context, Result}; use iroh::{EndpointAddr, EndpointId}; use serde_json::json; -use super::CoreInner; +use super::{CoreInner, POLL_MIN_INTERVAL_MS}; use crate::{ api::{ - ContactSummary, GrantLifetimeSetting, IncomingOffer, PendingPairing, ShareMetadataInput, - ShareResult, ShareSource, TransferAccessMode, + ContactSendResult, ContactSummary, GrantLifetimeSetting, HeldOfferSummary, IncomingOffer, + PendingPairing, ShareMetadataInput, ShareResult, ShareSource, TransferAccessMode, }, + contacts::HeldOffer, error::VnidropError, grant::{GrantId, HeldGrant}, offer::{ @@ -25,6 +26,14 @@ use crate::{ util::now_ms, }; +/// Whether a device may be polled again yet. +/// +/// Split out because the surrounding call needs two live nodes to exercise, +/// while the window itself is worth asserting on its own. +pub(crate) fn should_poll(last_polled_ms: Option, now_ms: i64) -> bool { + last_polled_ms.is_none_or(|last| now_ms - last >= POLL_MIN_INTERVAL_MS) +} + impl CoreInner { pub(super) async fn list_contacts(&self) -> Result> { let contacts = self @@ -152,7 +161,7 @@ impl CoreInner { endpoint_id: String, sources: Vec, mut metadata: ShareMetadataInput, - ) -> Result { + ) -> Result { let store = self.repository.contacts(); let grant = store .held_grant_for(&endpoint_id) @@ -171,17 +180,23 @@ impl CoreInner { let sender_name = metadata.sender_name.clone(); let share = self.share_files(sources, metadata).await?; - let outcome = self + // An unreachable device is the common case on mobile, not an error: the + // share stays here and the ticket waits for the peer to come and get it. + let outcome = match self .deliver_offer(&endpoint_id, &grant, &share, sender_name.as_deref()) .await - .inspect_err(|_| { - // Nothing to serve if the offer never landed. - let inner = self.clone(); - let transfer_id = share.transfer_id; - tokio::spawn(async move { - let _ = inner.cancel_idle_or_share(transfer_id).await; + { + Ok(outcome) => outcome, + Err(error) => { + self.hold_offer(&endpoint_id, &share, sender_name.as_deref()) + .await?; + tracing::debug!(%error, "offer held for later pickup"); + return Ok(ContactSendResult { + share, + delivered: false, }); - })?; + } + }; match outcome { OfferResponse::Accepted => { @@ -197,7 +212,10 @@ impl CoreInner { "offer-accepted", json!({ "peer_endpoint_id": endpoint_id }), ); - Ok(share) + Ok(ContactSendResult { + share, + delivered: true, + }) } OfferResponse::Declined { reason } | OfferResponse::Refused { reason } => { let _ = self.cancel_idle_or_share(share.transfer_id).await; @@ -221,6 +239,106 @@ impl CoreInner { } } + /// Keep an undeliverable offer on this device. + /// + /// The target is pre-authorised now rather than at pickup: it will dial + /// straight back after collecting the ticket, and the session outlives the + /// round trip. + async fn hold_offer( + self: &Arc, + endpoint_id: &str, + share: &ShareResult, + sender_name: Option<&str>, + ) -> Result<()> { + self.access_policy + .approve_endpoint(share.transfer_id, endpoint_id.to_string()) + .await; + self.repository + .contacts() + .insert_held_offer(&HeldOffer { + offer_id: uuid::Uuid::new_v4().to_string(), + endpoint_id: endpoint_id.to_string(), + transfer_id: share.transfer_id, + ticket: share.ticket.clone(), + transfer_name: share.transfer_name.clone(), + sender_display_name: sender_name.map(ToOwned::to_owned), + file_count: share.file_count, + total_bytes: share.total_size, + created_at: now_ms(), + }) + .await + .map_err(VnidropError::repository)?; + self.emit_transfer( + share.transfer_id, + "send", + "offer", + "offer-held", + json!({ "peer_endpoint_id": endpoint_id }), + ); + Ok(()) + } + + /// Ask remembered devices whether they are holding anything for this one. + /// + /// Deliberately only ever called from a foreground transition or an explicit + /// user action: polling reveals to every contact that the app was opened, + /// which is why it is neither automatic nor backgrounded. + pub(super) async fn poll_contacts_for_offers(self: &Arc) -> Result { + let store = self.repository.contacts(); + let contacts = store + .list_contacts() + .await + .map_err(VnidropError::repository)?; + let now = now_ms(); + let mut collected = 0u64; + + for contact in contacts { + if store + .is_blocked(&contact.endpoint_id) + .await + .unwrap_or(false) + { + continue; + } + { + // Rate limited per device so repeated app switching does not + // turn into a presence beacon. + let mut polled = self.last_polled.lock().await; + if !should_poll(polled.get(&contact.endpoint_id).copied(), now) { + continue; + } + polled.insert(contact.endpoint_id.clone(), now); + } + + let Ok(addr) = self.contact_addr(&contact.endpoint_id).await else { + continue; + }; + let client = OfferService::client(self.endpoint.clone(), addr); + let Ok(polled) = client.poll_offers().await else { + // Offline is the expected outcome, not a failure worth surfacing. + continue; + }; + for offer in polled.offers { + let added = self + .offers + .enqueue( + contact.endpoint_id.clone(), + offer.transfer_name, + offer.sender_display_name, + offer.file_count, + offer.total_bytes, + offer.ticket, + ) + .await; + if added { + collected += 1; + } + } + self.remember_addr(&contact.endpoint_id).await; + } + Ok(collected) + } + async fn deliver_offer( self: &Arc, endpoint_id: &str, @@ -258,6 +376,28 @@ impl CoreInner { .map_err(Into::into) } + /// Transfers waiting for their target to come back online. + pub(super) async fn list_held_offers(&self) -> Result> { + let held = self + .repository + .contacts() + .list_held_offers() + .await + .map_err(VnidropError::repository)?; + Ok(held + .into_iter() + .map(|offer| HeldOfferSummary { + offer_id: offer.offer_id, + endpoint_id: offer.endpoint_id, + transfer_id: offer.transfer_id, + transfer_name: offer.transfer_name, + file_count: offer.file_count, + total_bytes: offer.total_bytes, + created_at: offer.created_at, + }) + .collect()) + } + pub(super) async fn list_pending_offers(&self) -> Vec { self.offers .list() diff --git a/crates/vnidrop/src/runtime/facade.rs b/crates/vnidrop/src/runtime/facade.rs index 17d886c..58440f8 100644 --- a/crates/vnidrop/src/runtime/facade.rs +++ b/crates/vnidrop/src/runtime/facade.rs @@ -6,10 +6,11 @@ use serde_json::json; use super::CoreInner; use crate::{ api::{ - ContactSummary, CoreEvent, CoreEventSink, CoreLimits, CoreNetworkConfig, CoreStorageUsage, - GrantLifetimeSetting, IncomingOffer, PendingPairing, ReceiveOutputSink, - ReceiveOutputSinkV2, ReceivedArtifact, ReceiverRequest, RuntimeStatus, ShareMetadataInput, - ShareResult, ShareSource, StoredTransfer, TicketInspection, TransferAccessMode, + ContactSendResult, ContactSummary, CoreEvent, CoreEventSink, CoreLimits, CoreNetworkConfig, + CoreStorageUsage, GrantLifetimeSetting, HeldOfferSummary, IncomingOffer, PendingPairing, + ReceiveOutputSink, ReceiveOutputSinkV2, ReceivedArtifact, ReceiverRequest, RuntimeStatus, + ShareMetadataInput, ShareResult, ShareSource, StoredTransfer, TicketInspection, + TransferAccessMode, }, error::VnidropError, filesystem::platform_path, @@ -286,11 +287,27 @@ impl VnidropCore { endpoint_id: String, sources: Vec, metadata: ShareMetadataInput, - ) -> Result { + ) -> Result { self.block_on(self.inner.send_to_contact(endpoint_id, sources, metadata)) .map_err(VnidropError::transfer) } + /// Ask remembered devices whether they are holding transfers for this one. + /// + /// Call only from a foreground transition or an explicit user action: it + /// tells every contact that this device is awake. Returns how many offers + /// were collected. + pub fn poll_contacts_for_offers(&self) -> Result { + self.block_on(self.inner.poll_contacts_for_offers()) + .map_err(VnidropError::transfer) + } + + /// Transfers this device is holding for contacts that were not running. + pub fn list_held_offers(&self) -> Result, VnidropError> { + self.block_on(self.inner.list_held_offers()) + .map_err(VnidropError::repository) + } + /// Transfers paired devices are offering, awaiting this user's decision. pub fn list_pending_offers(&self) -> Vec { self.block_on(self.inner.list_pending_offers()) diff --git a/crates/vnidrop/src/runtime/lifecycle.rs b/crates/vnidrop/src/runtime/lifecycle.rs index cea2c9c..42a2e37 100644 --- a/crates/vnidrop/src/runtime/lifecycle.rs +++ b/crates/vnidrop/src/runtime/lifecycle.rs @@ -54,6 +54,13 @@ impl CoreInner { drop(active_shares); self.unregister_transfer_hashes(transfer_id).await; self.access_policy.remove_transfer(transfer_id).await; + // An offer waiting for pickup would hand out a ticket for content + // this device no longer serves. + let _ = self + .repository + .contacts() + .delete_held_offers_for_transfer(transfer_id) + .await; self.store.tags().delete(share_tag_name(&local_id)).await?; self.emit_transfer(transfer_id, "send", "lifecycle", "share-stopped", json!({})); return Ok(()); diff --git a/crates/vnidrop/src/runtime/mod.rs b/crates/vnidrop/src/runtime/mod.rs index 4611387..f18fa30 100644 --- a/crates/vnidrop/src/runtime/mod.rs +++ b/crates/vnidrop/src/runtime/mod.rs @@ -17,6 +17,8 @@ mod receive; mod share; mod storage; +#[cfg(test)] +pub(crate) use self::contacts::should_poll; pub use facade::VnidropCore; #[cfg(test)] pub(crate) use provider::{consume_request_updates, RequestStreamOutcome}; @@ -68,6 +70,10 @@ use crate::{ const RELAY_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); +/// Minimum gap between polls of the same device, so switching in and out of the +/// app does not announce presence to every contact repeatedly. +pub(super) const POLL_MIN_INTERVAL_MS: i64 = 5 * 60 * 1_000; + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum RelayStatus { Disabled, @@ -97,6 +103,8 @@ pub(super) struct CoreInner { pub(super) approval: ApprovalService, pub(super) pairing: PairingService, pub(super) offers: OfferInbox, + /// Endpoint → last poll time, for the rate limit above. + pub(super) last_polled: TokioMutex>, pub(super) limits: CoreLimits, pub(super) relay_mode: CoreRelayMode, pub(super) custom_relay_urls: Vec, @@ -362,6 +370,7 @@ impl CoreInner { approval, pairing, offers, + last_polled: TokioMutex::new(HashMap::new()), relay_mode, custom_relay_urls: relay_urls, transfer_slots: Semaphore::new(limits.max_concurrent_transfers as usize), diff --git a/crates/vnidrop/src/tests.rs b/crates/vnidrop/src/tests.rs index c14fbff..289570d 100644 --- a/crates/vnidrop/src/tests.rs +++ b/crates/vnidrop/src/tests.rs @@ -1,5 +1,7 @@ #[path = "tests/access_policy.rs"] mod access_policy_tests; +#[path = "tests/contact_polling.rs"] +mod contact_polling_tests; #[path = "tests/contacts.rs"] mod contacts_tests; #[path = "tests/error.rs"] diff --git a/crates/vnidrop/src/tests/contact_polling.rs b/crates/vnidrop/src/tests/contact_polling.rs new file mode 100644 index 0000000..344f68d --- /dev/null +++ b/crates/vnidrop/src/tests/contact_polling.rs @@ -0,0 +1,30 @@ +use crate::runtime::should_poll; + +const MINUTE_MS: i64 = 60 * 1_000; +const NOW: i64 = 1_700_000_000_000; + +#[test] +fn a_device_never_polled_is_polled() { + assert!(should_poll(None, NOW)); +} + +#[test] +fn a_device_polled_recently_is_skipped() { + // Switching in and out of the app must not re-announce presence. + assert!(!should_poll(Some(NOW), NOW)); + assert!(!should_poll(Some(NOW - MINUTE_MS), NOW)); + assert!(!should_poll(Some(NOW - 4 * MINUTE_MS), NOW)); +} + +#[test] +fn a_device_polled_before_the_window_is_polled_again() { + assert!(should_poll(Some(NOW - 5 * MINUTE_MS), NOW)); + assert!(should_poll(Some(NOW - 60 * MINUTE_MS), NOW)); +} + +/// A clock that jumped backwards must not lock polling out forever. +#[test] +fn a_future_timestamp_is_treated_as_recent_rather_than_permanent() { + assert!(!should_poll(Some(NOW + MINUTE_MS), NOW)); + assert!(should_poll(Some(NOW + MINUTE_MS), NOW + 6 * MINUTE_MS)); +} diff --git a/crates/vnidrop/src/tests/repository.rs b/crates/vnidrop/src/tests/repository.rs index 485b769..d02cfb6 100644 --- a/crates/vnidrop/src/tests/repository.rs +++ b/crates/vnidrop/src/tests/repository.rs @@ -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(), 8); + assert_eq!(repository.schema_version().await.unwrap(), 9); repository .insert_transfer(transfer( 7, @@ -645,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(), 8); + assert_eq!(repository.schema_version().await.unwrap(), 9); let stored = repository.list_transfers().await.unwrap().remove(0); assert_eq!(stored.transfer_id, 7); assert_eq!(stored.local_id, "legacy-7-send"); diff --git a/crates/vnidrop/tests/offer.rs b/crates/vnidrop/tests/offer.rs index 09c42ec..24b8ff9 100644 --- a/crates/vnidrop/tests/offer.rs +++ b/crates/vnidrop/tests/offer.rs @@ -10,8 +10,8 @@ use std::{ use support::{RecordingSink, TestNode}; use vnidrop::{ - IncomingOffer, ShareMetadataInput, ShareResult, ShareSource, SourceKind, TransferAccessMode, - VnidropCore, VnidropError, + ContactSendResult, IncomingOffer, ShareMetadataInput, ShareSource, SourceKind, + TransferAccessMode, VnidropCore, VnidropError, }; fn endpoint_id(node: &TestNode) -> String { @@ -69,7 +69,7 @@ fn send_in_background( to: String, path: &Path, transfer_id: u64, -) -> std::thread::JoinHandle> { +) -> std::thread::JoinHandle> { let sources = sources(path); std::thread::spawn(move || core.send_to_contact(to, sources, metadata(transfer_id))) } @@ -138,7 +138,7 @@ fn an_accepted_offer_transfers_without_prompting_the_sender() { // recorded as already approved. let requests = sender .core - .list_receiver_requests(share.transfer_id) + .list_receiver_requests(share.share.transfer_id) .unwrap(); assert_eq!(requests.len(), 1); assert!( @@ -269,7 +269,7 @@ fn an_offer_share_is_never_public() { .list_transfers() .unwrap() .into_iter() - .find(|transfer| transfer.transfer_id == share.transfer_id) + .find(|transfer| transfer.transfer_id == share.share.transfer_id) .expect("share recorded"); assert_eq!(stored.access_mode, TransferAccessMode::ApprovalRequired); } @@ -369,3 +369,195 @@ fn an_ordinary_ticket_receive_still_prompts_the_sender() { .unwrap(); handle.join().unwrap().expect("receive completes"); } + +// MARK: - Held offers and the foreground pull + +/// Restartable node, for simulating a device that was not running. +struct RestartableNode { + dir: tempfile::TempDir, + core: Option, +} + +impl RestartableNode { + fn new() -> Self { + let dir = tempfile::tempdir().unwrap(); + let core = support::CoreGuard::start(dir.path(), Arc::new(RecordingSink::default())); + Self { + dir, + core: Some(core), + } + } + + fn core(&self) -> &VnidropCore { + self.core.as_ref().expect("node is running") + } + + fn stop(&mut self) { + if let Some(core) = self.core.take() { + core.shutdown(); + } + } + + fn start(&mut self) { + self.core = Some(support::CoreGuard::start( + self.dir.path(), + Arc::new(RecordingSink::default()), + )); + } +} + +/// Pair so `sender` may reach the restartable node. +fn pair_with_restartable(sender: &TestNode, receiver: &RestartableNode) { + let receiver_id = receiver.core().status().endpoint_id; + receiver + .core() + .allow_device_to_reach_me(endpoint_id(sender), Some("Receiver".to_string())) + .expect("grant delivered"); + + let started = Instant::now(); + while !sender + .core + .list_pending_pairings() + .iter() + .any(|pending| pending.endpoint_id == receiver_id) + { + assert!( + started.elapsed() < Duration::from_secs(10), + "pairing offer never surfaced" + ); + std::thread::sleep(Duration::from_millis(25)); + } + sender + .core + .respond_to_pairing(receiver_id, true) + .expect("consent recorded"); +} + +/// The whole point of §11: a closed app is not an error, it is a delay. +#[test] +fn an_offer_to_a_device_that_is_not_running_is_held_and_collected_later() { + let source_dir = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join("shared.txt"); + std::fs::write(&source_path, b"held content").unwrap(); + let output_dir = tempfile::tempdir().unwrap(); + + let sender = TestNode::new(); + let mut receiver = RestartableNode::new(); + pair_with_restartable(&sender, &receiver); + let receiver_id = receiver.core().status().endpoint_id; + + receiver.stop(); + + let outcome = sender + .core + .send_to_contact(receiver_id, sources(&source_path), metadata(5_001)) + .expect("an unreachable device is not a failure"); + assert!( + !outcome.delivered, + "nothing was delivered, the offer is waiting" + ); + let held = sender.core.list_held_offers().unwrap(); + assert_eq!(held.len(), 1); + assert_eq!(held[0].transfer_id, outcome.share.transfer_id); + + receiver.start(); + let collected = receiver + .core() + .poll_contacts_for_offers() + .expect("poll succeeds"); + + assert_eq!(collected, 1); + let offer = receiver.core().list_pending_offers().remove(0); + assert_eq!(offer.transfer_name, "shared.txt"); + + let ticket = receiver + .core() + .respond_to_offer(offer.offer_id, true) + .expect("accepting yields the ticket"); + receiver + .core() + .receive( + ticket, + output_dir.path().to_string_lossy().to_string(), + Some("Receiver".to_string()), + ) + .expect("receive completes"); + + assert_eq!( + std::fs::read(output_dir.path().join("shared.txt")).unwrap(), + b"held content" + ); + assert!( + sender.core.list_held_offers().unwrap().is_empty(), + "a collected offer is no longer held" + ); +} + +/// Collected offers are consumed, so a second pull does not re-deliver them. +#[test] +fn polling_twice_does_not_collect_the_same_offer_again() { + let source_dir = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join("shared.txt"); + std::fs::write(&source_path, b"content").unwrap(); + + let sender = TestNode::new(); + let mut receiver = RestartableNode::new(); + pair_with_restartable(&sender, &receiver); + let receiver_id = receiver.core().status().endpoint_id; + receiver.stop(); + sender + .core + .send_to_contact(receiver_id, sources(&source_path), metadata(5_002)) + .unwrap(); + + // A fresh core each time, so the per-device poll rate limit does not mask + // the consume-on-delivery behaviour being asserted here. + receiver.start(); + assert_eq!(receiver.core().poll_contacts_for_offers().unwrap(), 1); + receiver.stop(); + receiver.start(); + assert_eq!( + receiver.core().poll_contacts_for_offers().unwrap(), + 0, + "the offer was already handed over" + ); +} + +/// Cancelling the transfer withdraws the ticket that was waiting for pickup. +#[test] +fn cancelling_a_transfer_withdraws_its_held_offer() { + let source_dir = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join("shared.txt"); + std::fs::write(&source_path, b"content").unwrap(); + + let sender = TestNode::new(); + let mut receiver = RestartableNode::new(); + pair_with_restartable(&sender, &receiver); + let receiver_id = receiver.core().status().endpoint_id; + receiver.stop(); + let outcome = sender + .core + .send_to_contact(receiver_id, sources(&source_path), metadata(5_004)) + .unwrap(); + assert_eq!(sender.core.list_held_offers().unwrap().len(), 1); + + sender + .core + .cancel_transfer(outcome.share.transfer_id) + .unwrap(); + + assert!(sender.core.list_held_offers().unwrap().is_empty()); + receiver.start(); + assert_eq!(receiver.core().poll_contacts_for_offers().unwrap(), 0); +} + +/// A device with no relationship learns nothing by polling. +#[test] +fn polling_a_device_that_holds_nothing_for_you_returns_nothing() { + let sender = TestNode::new(); + let receiver = TestNode::new(); + pair(&receiver, &sender); + + assert_eq!(receiver.core.poll_contacts_for_offers().unwrap(), 0); + assert!(receiver.core.list_pending_offers().is_empty()); +}