feat(core): hold undeliverable offers and collect them on demand

An unreachable device is a delay, not a failure: the share stays here and
the ticket waits in held_offers (schema 8 -> 9) until that device comes and
collects it. No server and no push, per the design.

Polling needs no grant proof. iroh has already authenticated the remote
endpoint key, and a device is only handed offers addressed to precisely that
endpoint, so a stranger polling learns nothing. Offers are consumed on
delivery, so polling twice does not re-deliver, and cancelling the transfer
withdraws the waiting ticket.

Polling is rate limited per device: it tells every contact the app was
opened, so it must never become a presence beacon.

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

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<String>,
+    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<Vec<HeldOffer>> {
+        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<Vec<HeldOffer>> {
+        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::<i64, _>(2) as u64,
+        ticket: row.get(3),
+        transfer_name: row.get(4),
+        sender_display_name: row.get(5),
+        file_count: row.get::<i64, _>(6) as u64,
+        total_bytes: row.get::<i64, _>(7) as u64,
+        created_at: row.get(8),
+    }
+}
+
 fn row_to_issued_grant(row: sqlx::sqlite::SqliteRow) -> Result<IssuedGrant> {
     let grant_id = GrantId::decode(row.get::<String, _>(0).as_str())?;
     let secret = parse_secret(row.get::<String, _>(1).as_str())

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;

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<PolledOffers, irpc::Error> {
+        self.inner.rpc(PollOffers).await
+    }
+
     pub(crate) async fn request_challenge(&self) -> Result<Challenge, irpc::Error> {
         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<PolledOffer>,
+}
+
+/// 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<String>,
+    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<OfferResponse>)]
     SubmitOffer(SubmitOffer),
+    #[rpc(tx=oneshot::Sender<PolledOffers>)]
+    PollOffers(PollOffers),
 }

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<String>,
+        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<PendingOffer> {
         self.pending.lock().await.values().cloned().collect()
     }

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<PolledOffer> {
+        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<String> = 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

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 {

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<i64>, 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<Vec<ContactSummary>> {
         let contacts = self
@@ -152,7 +161,7 @@ impl CoreInner {
         endpoint_id: String,
         sources: Vec<ShareSource>,
         mut metadata: ShareMetadataInput,
-    ) -> Result<ShareResult> {
+    ) -> Result<ContactSendResult> {
         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<Self>,
+        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<Self>) -> Result<u64> {
+        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<Self>,
         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<Vec<HeldOfferSummary>> {
+        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<IncomingOffer> {
         self.offers
             .list()

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<ShareSource>,
         metadata: ShareMetadataInput,
-    ) -> Result<ShareResult, VnidropError> {
+    ) -> Result<ContactSendResult, VnidropError> {
         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<u64, VnidropError> {
+        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<Vec<HeldOfferSummary>, 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<IncomingOffer> {
         self.block_on(self.inner.list_pending_offers())

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(());

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<HashMap<String, i64>>,
     pub(super) limits: CoreLimits,
     pub(super) relay_mode: CoreRelayMode,
     pub(super) custom_relay_urls: Vec<RelayUrl>,
@@ -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),

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"]

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));
+}

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");

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<Result<ShareResult, VnidropError>> {
+) -> std::thread::JoinHandle<Result<ContactSendResult, VnidropError>> {
     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<support::CoreGuard>,
+}
+
+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());
+}