Harden Rust transfer core

crates/vnidrop/src/access_policy.rs

@@ -48,6 +48,9 @@ impl AccessPolicy {
         transfer_id: u64,
         endpoint_id: Option<&str>,
     ) -> AccessDecision {
+        // This is intentionally only the provider-side gate for milestone one.
+        // A later handshake can add receiver-request/sender-approval events on
+        // top without weakening the default public sharing behavior.
         match self
             .modes
             .read()

crates/vnidrop/src/error.rs

@@ -2,13 +2,63 @@ use std::io;
 
 #[derive(Debug, thiserror::Error, uniffi::Error)]
 pub enum VnidropError {
-    #[error("{reason}")]
+    #[error("initialization failed: {reason}")]
-    Generic { reason: String },
+    Initialization { reason: String },
+    #[error("ticket error: {reason}")]
+    Ticket { reason: String },
+    #[error("filesystem error: {reason}")]
+    Filesystem { reason: String },
+    #[error("transfer error: {reason}")]
+    Transfer { reason: String },
+    #[error("permission error: {reason}")]
+    Permission { reason: String },
+    #[error("repository error: {reason}")]
+    Repository { reason: String },
+    #[error("internal error: {reason}")]
+    Internal { reason: String },
+}
+
+impl VnidropError {
+    pub(crate) fn initialization(error: impl Into<anyhow::Error>) -> Self {
+        Self::Initialization {
+            reason: error.into().to_string(),
+        }
+    }
+
+    pub(crate) fn ticket(error: impl Into<anyhow::Error>) -> Self {
+        Self::Ticket {
+            reason: error.into().to_string(),
+        }
+    }
+
+    pub(crate) fn filesystem(error: impl Into<anyhow::Error>) -> Self {
+        Self::Filesystem {
+            reason: error.into().to_string(),
+        }
+    }
+
+    pub(crate) fn transfer(error: impl Into<anyhow::Error>) -> Self {
+        Self::Transfer {
+            reason: error.into().to_string(),
+        }
+    }
+
+    pub(crate) fn permission(error: impl Into<anyhow::Error>) -> Self {
+        Self::Permission {
+            reason: error.into().to_string(),
+        }
+    }
+
+    pub(crate) fn repository(error: impl Into<anyhow::Error>) -> Self {
+        Self::Repository {
+            reason: error.into().to_string(),
+        }
+    }
 }
 
 impl From<anyhow::Error> for VnidropError {
     fn from(error: anyhow::Error) -> Self {
-        Self::Generic {
+        Self::Internal {
             reason: error.to_string(),
         }
     }
@@ -16,7 +66,7 @@ impl From<anyhow::Error> for VnidropError {
 
 impl From<io::Error> for VnidropError {
     fn from(error: io::Error) -> Self {
-        Self::Generic {
+        Self::Filesystem {
             reason: error.to_string(),
         }
     }

crates/vnidrop/src/event_hub.rs

@@ -0,0 +1,130 @@
+use std::sync::{Arc, Mutex};
+
+use serde_json::Value;
+use tokio::{
+    sync::{mpsc, oneshot, Mutex as TokioMutex},
+    task::JoinHandle,
+};
+
+use crate::{
+    api::{CoreEvent, CoreEventSink},
+    repository::Repository,
+    util::now_ms,
+};
+
+enum EventCommand {
+    Persist(CoreEvent),
+    Flush(oneshot::Sender<()>),
+    Shutdown(oneshot::Sender<()>),
+}
+
+pub(crate) struct EventHub {
+    sink: Arc<dyn CoreEventSink>,
+    tx: mpsc::UnboundedSender<EventCommand>,
+    join: TokioMutex<Option<JoinHandle<()>>>,
+    sequence: Mutex<u64>,
+}
+
+impl EventHub {
+    pub(crate) fn start(repository: Repository, sink: Arc<dyn CoreEventSink>) -> Self {
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let join = tokio::spawn(async move {
+            while let Some(command) = rx.recv().await {
+                match command {
+                    EventCommand::Persist(event) => {
+                        if let Err(error) = repository.insert_event(&event).await {
+                            tracing::warn!(%error, event_id = %event.id, "failed to persist core event");
+                        }
+                    }
+                    EventCommand::Flush(done) => {
+                        let _ = done.send(());
+                    }
+                    EventCommand::Shutdown(done) => {
+                        let _ = done.send(());
+                        break;
+                    }
+                }
+            }
+        });
+
+        Self {
+            sink,
+            tx,
+            join: TokioMutex::new(Some(join)),
+            sequence: Mutex::new(1),
+        }
+    }
+
+    pub(crate) fn emit_endpoint(&self, phase: &str, kind: &str, data: Value) {
+        self.emit("endpoint", None, None, phase, kind, data);
+    }
+
+    pub(crate) fn emit_transfer(
+        &self,
+        transfer_id: u64,
+        direction: &str,
+        phase: &str,
+        kind: &str,
+        data: Value,
+    ) {
+        self.emit(
+            "transfer",
+            Some(transfer_id),
+            Some(direction.to_string()),
+            phase,
+            kind,
+            data,
+        );
+    }
+
+    pub(crate) async fn flush(&self) {
+        let (tx, rx) = oneshot::channel();
+        if self.tx.send(EventCommand::Flush(tx)).is_ok() {
+            let _ = rx.await;
+        }
+    }
+
+    pub(crate) async fn shutdown(&self) {
+        let (tx, rx) = oneshot::channel();
+        if self.tx.send(EventCommand::Shutdown(tx)).is_ok() {
+            let _ = rx.await;
+        }
+        if let Some(join) = self.join.lock().await.take() {
+            let _ = join.await;
+        }
+    }
+
+    fn emit(
+        &self,
+        scope: &str,
+        transfer_id: Option<u64>,
+        direction: Option<String>,
+        phase: &str,
+        kind: &str,
+        data: Value,
+    ) {
+        let timestamp = now_ms();
+        let mut sequence = self.sequence.lock().expect("event sequence lock poisoned");
+        let id = format!("{timestamp}-{}", *sequence);
+        *sequence += 1;
+        drop(sequence);
+
+        // Compose observes this event synchronously, while SQLite persistence is
+        // serialized through the queue.  That keeps the UI responsive without
+        // losing the ability to flush persisted history during shutdown/tests.
+        let event = CoreEvent {
+            id,
+            timestamp,
+            scope: scope.to_string(),
+            transfer_id,
+            direction,
+            phase: phase.to_string(),
+            kind: kind.to_string(),
+            data_json: data.to_string(),
+        };
+        if self.tx.send(EventCommand::Persist(event.clone())).is_err() {
+            tracing::warn!(event_id = %event.id, "event persistence queue is closed");
+        }
+        self.sink.on_event(event);
+    }
+}

crates/vnidrop/src/filesystem.rs

@@ -122,10 +122,16 @@ pub(crate) fn collect_import_files(sources: Vec<ShareSource>) -> Result<Vec<Impo
 }
 
 fn source_path(source: &ShareSource) -> Result<PathBuf> {
+    if source.value.trim().is_empty() {
+        anyhow::bail!("source path must not be empty");
+    }
     platform_path(&source.value)
 }
 
 pub(crate) fn platform_path(value: &str) -> Result<PathBuf> {
+    if value.trim().is_empty() {
+        anyhow::bail!("path must not be empty");
+    }
     if let Some(without_scheme) = value.strip_prefix("file://") {
         return Ok(PathBuf::from(percent_decode_file_url_path(without_scheme)?));
     }
@@ -230,7 +236,11 @@ pub(crate) async fn wait_for_writer(
 }
 
 pub(crate) fn validated_relative_string(name: &str) -> Result<String> {
-    path_to_string(Path::new(name), true)
+    let value = path_to_string(Path::new(name), true)?;
+    if value.trim().is_empty() {
+        anyhow::bail!("relative path must not be empty");
+    }
+    Ok(value)
 }
 
 pub(crate) fn path_to_string(path: impl AsRef<Path>, must_be_relative: bool) -> Result<String> {

crates/vnidrop/src/lib.rs

@@ -1,6 +1,7 @@
 mod access_policy;
 mod api;
 mod error;
+mod event_hub;
 mod filesystem;
 mod logging;
 mod repository;
@@ -19,4 +20,5 @@ pub use runtime::VnidropCore;
 uniffi::setup_scaffolding!();
 
 #[cfg(test)]
-mod tests;
+#[path = "tests.rs"]
+mod core_tests;

crates/vnidrop/src/repository.rs

@@ -9,11 +9,24 @@ use sqlx::{
 use crate::api::{CoreEvent, StoredTransfer};
 use crate::util::now_ms;
 
+const SCHEMA_VERSION: i64 = 1;
+
 #[derive(Debug, Clone)]
 pub(crate) struct Repository {
     pool: SqlitePool,
 }
 
+pub(crate) struct TransferUpsert<'a> {
+    pub(crate) transfer_id: u64,
+    pub(crate) direction: &'a str,
+    pub(crate) status: &'a str,
+    pub(crate) transfer_name: Option<&'a str>,
+    pub(crate) content_hash: Option<&'a str>,
+    pub(crate) ticket: Option<&'a str>,
+    pub(crate) file_count: u64,
+    pub(crate) total_size: u64,
+}
+
 impl Repository {
     pub(crate) async fn open(app_data_dir: &Path) -> Result<Self> {
         let db_path = app_data_dir.join("vnidrop.sqlite3");
@@ -30,6 +43,8 @@ impl Repository {
     }
 
     async fn ensure_schema(&self) -> Result<()> {
+        // The app owns this SQLite file.  Keep migrations explicit so future
+        // desktop/mobile releases can move user history forward in place.
         sqlx::query(
             r#"
             CREATE TABLE IF NOT EXISTS transfers (
@@ -71,20 +86,22 @@ impl Repository {
         )
         .execute(&self.pool)
         .await?;
+
+        sqlx::query(&format!("PRAGMA user_version = {SCHEMA_VERSION}"))
+            .execute(&self.pool)
+            .await?;
         Ok(())
     }
 
-    pub(crate) async fn upsert_transfer(
+    #[cfg(test)]
-        &self,
+    pub(crate) async fn schema_version(&self) -> Result<i64> {
-        transfer_id: u64,
+        let row = sqlx::query("PRAGMA user_version")
-        direction: &str,
+            .fetch_one(&self.pool)
-        status: &str,
+            .await?;
-        transfer_name: Option<&str>,
+        Ok(row.get(0))
-        content_hash: Option<&str>,
+    }
-        ticket: Option<&str>,
+
-        file_count: u64,
+    pub(crate) async fn upsert_transfer(&self, transfer: TransferUpsert<'_>) -> Result<()> {
-        total_size: u64,
-    ) -> Result<()> {
         let now = now_ms();
         sqlx::query(
             r#"
@@ -104,14 +121,14 @@ impl Repository {
                 updated_at = excluded.updated_at;
             "#,
         )
-        .bind(transfer_id as i64)
+        .bind(transfer.transfer_id as i64)
-        .bind(direction)
+        .bind(transfer.direction)
-        .bind(status)
+        .bind(transfer.status)
-        .bind(transfer_name)
+        .bind(transfer.transfer_name)
-        .bind(content_hash)
+        .bind(transfer.content_hash)
-        .bind(ticket)
+        .bind(transfer.ticket)
-        .bind(file_count as i64)
+        .bind(transfer.file_count as i64)
-        .bind(total_size as i64)
+        .bind(transfer.total_size as i64)
         .bind(now)
         .execute(&self.pool)
         .await?;

crates/vnidrop/src/runtime.rs

@@ -3,7 +3,10 @@ use std::{
     fs::File,
     io,
     path::{Path, PathBuf},
-    sync::{Arc, Mutex},
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc,
+    },
 };
 
 use anyhow::{Context, Result};
@@ -21,7 +24,10 @@ use iroh_blobs::{
 };
 use n0_future::BufferedStreamExt;
 use serde_json::json;
-use tokio::sync::{mpsc, oneshot, Mutex as TokioMutex};
+use tokio::{
+    sync::{mpsc, oneshot, Mutex as TokioMutex},
+    task::JoinHandle,
+};
 
 use crate::{
     access_policy::{AccessDecision, AccessPolicy},
@@ -30,18 +36,26 @@ use crate::{
         StoredTransfer, TicketInspection, TransferAccessMode, TransferMetadata,
     },
     error::VnidropError,
+    event_hub::EventHub,
     filesystem::{
         collect_import_files, default_collection_name, platform_path,
         read_stream_from_blocking_reader, safe_output_path, wait_for_writer,
         write_stream_to_blocking_writer, TransferImport,
     },
     logging::init_logging,
-    repository::Repository,
+    repository::{Repository, TransferUpsert},
     secret::load_or_create_secret,
     ticket::{parse_transfer_ticket, ParsedTransferTicket, VnidropTicket},
-    util::{non_empty, now_ms, unique_transfer_id},
+    util::{non_empty, unique_transfer_id},
 };
 
+const STATUS_SHARING: &str = "sharing";
+const STATUS_RECEIVING: &str = "receiving";
+const STATUS_DONE: &str = "done";
+const STATUS_CANCELLED: &str = "cancelled";
+const STATUS_STOPPED: &str = "stopped";
+const STATUS_FAILED: &str = "failed";
+
 #[derive(uniffi::Object)]
 pub struct VnidropCore {
     runtime: tokio::runtime::Runtime,
@@ -49,17 +63,20 @@ pub struct VnidropCore {
 }
 
 struct CoreInner {
+    // Kotlin owns app lifecycle and platform file picking; this Rust object
+    // owns the Iroh endpoint, blob store, transfer history, and byte streaming.
     endpoint: Endpoint,
     router: Router,
     store: FsStore,
     repository: Repository,
+    event_hub: EventHub,
     access_policy: Arc<AccessPolicy>,
-    event_sink: Arc<dyn CoreEventSink>,
     active_transfers: TokioMutex<HashMap<u64, oneshot::Sender<()>>>,
     active_shares: TokioMutex<HashMap<u64, TempTag>>,
     hash_to_transfer: TokioMutex<HashMap<String, u64>>,
     connection_endpoints: TokioMutex<HashMap<u64, String>>,
-    sequence: Mutex<u64>,
+    provider_task: TokioMutex<Option<JoinHandle<()>>>,
+    shutdown_started: AtomicBool,
 }
 
 #[uniffi::export]
@@ -74,7 +91,9 @@ impl VnidropCore {
             .thread_name("vnidrop")
             .build()?;
         let app_data_dir = PathBuf::from(app_data_dir);
-        let inner = runtime.block_on(CoreInner::start(app_data_dir, event_sink))?;
+        let inner = runtime
+            .block_on(CoreInner::start(app_data_dir, event_sink))
+            .map_err(VnidropError::initialization)?;
         Ok(Arc::new(Self { runtime, inner }))
     }
 
@@ -89,7 +108,7 @@ impl VnidropCore {
     ) -> Result<ShareResult, VnidropError> {
         self.runtime
             .block_on(self.inner.share_files(sources, metadata))
-            .map_err(Into::into)
+            .map_err(VnidropError::transfer)
     }
 
     pub fn receive(
@@ -98,16 +117,29 @@ impl VnidropCore {
         output_dir: String,
         receiver_name: Option<String>,
     ) -> Result<(), VnidropError> {
-        let output_dir = platform_path(&output_dir)?;
+        if let Err(error) =
+            parse_transfer_ticket(&ticket).context("failed to parse transfer ticket")
+        {
+            self.runtime.block_on(async {
+                self.inner.emit_endpoint(
+                    "error",
+                    "invalid-ticket",
+                    json!({ "reason": error.to_string() }),
+                );
+                self.inner.event_hub.flush().await;
+            });
+            return Err(VnidropError::ticket(error));
+        }
+        let output_dir = platform_path(&output_dir).map_err(VnidropError::filesystem)?;
         self.runtime
             .block_on(self.inner.receive(ticket, output_dir, receiver_name))
-            .map_err(Into::into)
+            .map_err(VnidropError::transfer)
     }
 
     pub fn cancel_transfer(&self, transfer_id: u64) -> Result<(), VnidropError> {
         self.runtime
             .block_on(self.inner.cancel_transfer(transfer_id))
-            .map_err(Into::into)
+            .map_err(VnidropError::transfer)
     }
 
     pub fn set_transfer_access_mode(
@@ -117,7 +149,7 @@ impl VnidropCore {
     ) -> Result<(), VnidropError> {
         self.runtime
             .block_on(self.inner.set_transfer_access_mode(transfer_id, mode))
-            .map_err(Into::into)
+            .map_err(VnidropError::permission)
     }
 
     pub fn approve_endpoint_for_transfer(
@@ -130,23 +162,25 @@ impl VnidropCore {
                 self.inner
                     .approve_endpoint_for_transfer(transfer_id, endpoint_id),
             )
-            .map_err(Into::into)
+            .map_err(VnidropError::permission)
     }
 
     pub fn list_transfers(&self) -> Result<Vec<StoredTransfer>, VnidropError> {
         self.runtime
             .block_on(self.inner.repository.list_transfers())
-            .map_err(Into::into)
+            .map_err(VnidropError::repository)
     }
 
     pub fn list_events(&self, transfer_id: Option<u64>) -> Result<Vec<CoreEvent>, VnidropError> {
         self.runtime
-            .block_on(self.inner.repository.list_events(transfer_id))
+            .block_on(self.inner.list_events(transfer_id))
-            .map_err(Into::into)
+            .map_err(VnidropError::repository)
     }
 
     pub fn inspect_ticket(&self, ticket: String) -> Result<TicketInspection, VnidropError> {
-        let parsed = parse_transfer_ticket(&ticket).context("failed to parse transfer ticket")?;
+        let parsed = parse_transfer_ticket(&ticket)
+            .context("failed to parse transfer ticket")
+            .map_err(VnidropError::ticket)?;
         Ok(TicketInspection {
             kind: if parsed.metadata.is_some() {
                 "vnidrop".to_string()
@@ -177,24 +211,28 @@ impl CoreInner {
             .await?;
         endpoint.online().await;
 
+        // Provider events are where the sender sees remote readers.  The core
+        // uses them for send progress and for the current approval gate.
         let (events, event_rx) = EventSender::channel(128, EventMask::ALL_READONLY);
         let blobs = BlobsProtocol::new(&store, Some(events));
         let router = Router::builder(endpoint.clone())
             .accept(iroh_blobs::ALPN, blobs)
             .spawn();
+        let event_hub = EventHub::start(repository.clone(), event_sink);
 
         let inner = Arc::new(Self {
             endpoint,
             router,
             store,
             repository,
+            event_hub,
             access_policy: AccessPolicy::new(),
-            event_sink,
             active_transfers: TokioMutex::new(HashMap::new()),
             active_shares: TokioMutex::new(HashMap::new()),
             hash_to_transfer: TokioMutex::new(HashMap::new()),
             connection_endpoints: TokioMutex::new(HashMap::new()),
-            sequence: Mutex::new(1),
+            provider_task: TokioMutex::new(None),
+            shutdown_started: AtomicBool::new(false),
         });
 
         inner.emit_endpoint(
@@ -206,7 +244,7 @@ impl CoreInner {
                 "store_root": store_root.to_string_lossy(),
             }),
         );
-        inner.spawn_provider_event_task(event_rx);
+        inner.spawn_provider_event_task(event_rx).await;
         Ok(inner)
     }
 
@@ -223,6 +261,29 @@ impl CoreInner {
         self: &Arc<Self>,
         sources: Vec<ShareSource>,
         metadata: ShareMetadataInput,
+    ) -> Result<ShareResult> {
+        let transfer_id = metadata.transfer_id;
+        let result = self.share_files_inner(sources, metadata).await;
+        if let Err(error) = &result {
+            self.emit_transfer(
+                transfer_id,
+                "send",
+                "error",
+                "failed",
+                json!({ "reason": error.to_string() }),
+            );
+            let _ = self
+                .repository
+                .update_transfer_status(transfer_id, STATUS_FAILED)
+                .await;
+        }
+        result
+    }
+
+    async fn share_files_inner(
+        self: &Arc<Self>,
+        sources: Vec<ShareSource>,
+        metadata: ShareMetadataInput,
     ) -> Result<ShareResult> {
         if sources.is_empty() {
             anyhow::bail!("at least one source is required");
@@ -266,16 +327,16 @@ impl CoreInner {
             .await
             .insert(metadata.transfer_id, import.tag);
         self.repository
-            .upsert_transfer(
+            .upsert_transfer(TransferUpsert {
-                metadata.transfer_id,
+                transfer_id: metadata.transfer_id,
-                "send",
+                direction: "send",
-                "sharing",
+                status: STATUS_SHARING,
-                Some(&transfer_name),
+                transfer_name: Some(&transfer_name),
-                Some(&import.root_hash.to_string()),
+                content_hash: Some(&import.root_hash.to_string()),
-                Some(&ticket),
+                ticket: Some(&ticket),
-                import.file_count,
+                file_count: import.file_count,
-                import.total_size,
+                total_size: import.total_size,
-            )
+            })
             .await?;
 
         self.emit_transfer(
@@ -308,12 +369,25 @@ impl CoreInner {
         output_dir: PathBuf,
         receiver_name: Option<String>,
     ) -> Result<()> {
-        let parsed = parse_transfer_ticket(&ticket).context("failed to parse transfer ticket")?;
+        let parsed = match parse_transfer_ticket(&ticket).context("failed to parse transfer ticket")
+        {
+            Ok(parsed) => parsed,
+            Err(error) => {
+                self.emit_endpoint(
+                    "error",
+                    "invalid-ticket",
+                    json!({ "reason": error.to_string() }),
+                );
+                return Err(error);
+            }
+        };
         let transfer_id = parsed
             .metadata
             .as_ref()
             .map(|metadata| metadata.transfer_id)
             .unwrap_or_else(unique_transfer_id);
+        // Cancellation is cooperative: it stops our receive future and marks
+        // local state while lower-level Iroh work unwinds naturally.
         let (shutdown_tx, mut shutdown_rx) = oneshot::channel();
         self.active_transfers
             .lock()
@@ -326,6 +400,19 @@ impl CoreInner {
         };
 
         self.active_transfers.lock().await.remove(&transfer_id);
+        if let Err(error) = &result {
+            self.emit_transfer(
+                transfer_id,
+                "receive",
+                "error",
+                "failed",
+                json!({ "reason": error.to_string() }),
+            );
+            let _ = self
+                .repository
+                .update_transfer_status(transfer_id, STATUS_FAILED)
+                .await;
+        }
         result
     }
 
@@ -336,7 +423,8 @@ impl CoreInner {
         output_dir: PathBuf,
         receiver_name: Option<String>,
     ) -> Result<()> {
-        let metadata_json = serde_json::to_value(&parsed.metadata).unwrap_or_else(|_| json!(null));
+        let metadata_json =
+            serde_json::to_value(&parsed.metadata).unwrap_or(serde_json::Value::Null);
         self.emit_transfer(
             transfer_id,
             "receive",
@@ -348,30 +436,30 @@ impl CoreInner {
             }),
         );
         self.repository
-            .upsert_transfer(
+            .upsert_transfer(TransferUpsert {
                 transfer_id,
-                "receive",
+                direction: "receive",
-                "receiving",
+                status: STATUS_RECEIVING,
-                parsed
+                transfer_name: parsed
                     .metadata
                     .as_ref()
                     .map(|metadata| metadata.transfer_name.as_str()),
-                parsed
+                content_hash: parsed
                     .metadata
                     .as_ref()
                     .map(|metadata| metadata.content_hash.as_str()),
-                None,
+                ticket: None,
-                parsed
+                file_count: parsed
                     .metadata
                     .as_ref()
                     .map(|metadata| metadata.file_count)
                     .unwrap_or_default(),
-                parsed
+                total_size: parsed
                     .metadata
                     .as_ref()
                     .map(|metadata| metadata.total_size)
                     .unwrap_or_default(),
-            )
+            })
             .await?;
         tokio::fs::create_dir_all(&output_dir).await?;
 
@@ -419,7 +507,7 @@ impl CoreInner {
         self.export_collection(transfer_id, total_files, output_dir, collection)
             .await?;
         self.repository
-            .update_transfer_status(transfer_id, "done")
+            .update_transfer_status(transfer_id, STATUS_DONE)
             .await?;
         self.emit_transfer(transfer_id, "receive", "lifecycle", "done", json!({}));
         Ok(())
@@ -436,7 +524,7 @@ impl CoreInner {
                 json!({}),
             );
             self.repository
-                .update_transfer_status(transfer_id, "cancelled")
+                .update_transfer_status(transfer_id, STATUS_CANCELLED)
                 .await?;
             return Ok(());
         }
@@ -453,7 +541,7 @@ impl CoreInner {
                 .retain(|_, id| *id != transfer_id);
             self.access_policy.remove_transfer(transfer_id).await;
             self.repository
-                .update_transfer_status(transfer_id, "stopped")
+                .update_transfer_status(transfer_id, STATUS_STOPPED)
                 .await?;
             self.emit_transfer(transfer_id, "send", "lifecycle", "share-stopped", json!({}));
             return Ok(());
@@ -496,7 +584,13 @@ impl CoreInner {
     }
 
     async fn shutdown(&self) {
+        if self.shutdown_started.swap(true, Ordering::SeqCst) {
+            return;
+        }
         self.emit_endpoint("shutdown", "service-shutdown", json!({}));
+        // Flush before stopping the router so the app can show the shutdown
+        // event even if the process exits soon after Compose disposes the core.
+        self.event_hub.flush().await;
         if let Err(error) = self.router.shutdown().await {
             self.emit_endpoint(
                 "shutdown",
@@ -504,6 +598,11 @@ impl CoreInner {
                 json!({ "error": error.to_string() }),
             );
         }
+        if let Some(task) = self.provider_task.lock().await.take() {
+            task.abort();
+            let _ = task.await;
+        }
+        self.event_hub.shutdown().await;
     }
 
     async fn import_sources(
@@ -693,13 +792,14 @@ impl CoreInner {
         Ok(())
     }
 
-    fn spawn_provider_event_task(self: &Arc<Self>, mut rx: mpsc::Receiver<ProviderMessage>) {
+    async fn spawn_provider_event_task(self: &Arc<Self>, mut rx: mpsc::Receiver<ProviderMessage>) {
         let core = self.clone();
-        tokio::spawn(async move {
+        let task = tokio::spawn(async move {
             while let Some(message) = rx.recv().await {
                 core.handle_provider_message(message).await;
             }
         });
+        *self.provider_task.lock().await = Some(task);
     }
 
     async fn handle_provider_message(self: &Arc<Self>, message: ProviderMessage) {
@@ -917,6 +1017,9 @@ impl CoreInner {
         request_id: u64,
         mut rx: irpc::channel::mpsc::Receiver<RequestUpdate>,
     ) {
+        // Request update tasks are tied to individual provider streams.  Router
+        // shutdown closes those streams; only the long-lived provider receiver
+        // is tracked directly for explicit shutdown.
         let core = self.clone();
         tokio::spawn(async move {
             while let Ok(Some(update)) = rx.recv().await {
@@ -965,7 +1068,7 @@ impl CoreInner {
     }
 
     fn emit_endpoint(&self, phase: &str, kind: &str, data: serde_json::Value) {
-        self.emit("endpoint", None, None, phase, kind, data);
+        self.event_hub.emit_endpoint(phase, kind, data);
     }
 
     fn emit_transfer(
@@ -976,47 +1079,12 @@ impl CoreInner {
         kind: &str,
         data: serde_json::Value,
     ) {
-        self.emit(
+        self.event_hub
-            "transfer",
+            .emit_transfer(transfer_id, direction, phase, kind, data);
-            Some(transfer_id),
-            Some(direction.to_string()),
-            phase,
-            kind,
-            data,
-        );
     }
 
-    fn emit(
+    async fn list_events(&self, transfer_id: Option<u64>) -> Result<Vec<CoreEvent>> {
-        &self,
+        self.event_hub.flush().await;
-        scope: &str,
+        self.repository.list_events(transfer_id).await
-        transfer_id: Option<u64>,
-        direction: Option<String>,
-        phase: &str,
-        kind: &str,
-        data: serde_json::Value,
-    ) {
-        let timestamp = now_ms();
-        let mut sequence = self.sequence.lock().expect("event sequence lock poisoned");
-        let id = format!("{timestamp}-{}", *sequence);
-        *sequence += 1;
-        drop(sequence);
-        let event = CoreEvent {
-            id,
-            timestamp,
-            scope: scope.to_string(),
-            transfer_id,
-            direction,
-            phase: phase.to_string(),
-            kind: kind.to_string(),
-            data_json: data.to_string(),
-        };
-        let repository = self.repository.clone();
-        let event_for_repository = event.clone();
-        tokio::spawn(async move {
-            if let Err(error) = repository.insert_event(&event_for_repository).await {
-                tracing::warn!(%error, "failed to persist core event");
-            }
-        });
-        self.event_sink.on_event(event);
     }
 }

crates/vnidrop/src/tests.rs

@@ -4,15 +4,18 @@ mod tests {
     use std::os::fd::AsRawFd;
     use std::{io::Read, path::Path, sync::Arc};
 
+    use data_encoding::BASE64URL_NOPAD;
     use iroh::SecretKey;
     use iroh_blobs::{ticket::BlobTicket, BlobFormat, Hash};
+    use serde_json::json;
 
     use crate::{
         access_policy::{AccessDecision, AccessPolicy},
         api::{CoreEvent, CoreEventSink, ShareSource, SourceKind, TransferMetadata},
+        error::VnidropError,
         filesystem::{
-            collect_import_files, path_to_string, percent_decode_file_url_path,
+            collect_import_files, default_collection_name, path_to_string,
-            validated_relative_string,
+            percent_decode_file_url_path, validated_relative_string,
         },
         repository::Repository,
         runtime::VnidropCore,
@@ -77,6 +80,58 @@ mod tests {
         assert!(parse_transfer_ticket("not-a-ticket").is_err());
     }
 
+    #[test]
+    fn ticket_rejects_unsupported_versions_and_mismatched_hashes() {
+        let secret = SecretKey::generate();
+        let addr = iroh::EndpointAddr::new(secret.public());
+        let blob_ticket = BlobTicket::new(addr, Hash::new([5; 32]), BlobFormat::HashSeq);
+        let payload = json!({
+            "version": 2,
+            "blob_ticket": blob_ticket.to_string(),
+            "metadata": {
+                "version": 1,
+                "transfer_id": 7,
+                "transfer_name": "bad version",
+                "sender_name": null,
+                "created_at": 1,
+                "content_hash": blob_ticket.hash().to_string(),
+                "file_count": 1,
+                "total_size": 10
+            }
+        });
+        let encoded = format!(
+            "vnd1:{}",
+            BASE64URL_NOPAD.encode(payload.to_string().as_bytes())
+        );
+        assert!(parse_transfer_ticket(&encoded)
+            .unwrap_err()
+            .to_string()
+            .contains("unsupported VniDrop ticket version"));
+
+        let payload = json!({
+            "version": 1,
+            "blob_ticket": blob_ticket.to_string(),
+            "metadata": {
+                "version": 1,
+                "transfer_id": 7,
+                "transfer_name": "bad hash",
+                "sender_name": null,
+                "created_at": 1,
+                "content_hash": Hash::new([6; 32]).to_string(),
+                "file_count": 1,
+                "total_size": 10
+            }
+        });
+        let encoded = format!(
+            "vnd1:{}",
+            BASE64URL_NOPAD.encode(payload.to_string().as_bytes())
+        );
+        assert!(parse_transfer_ticket(&encoded)
+            .unwrap_err()
+            .to_string()
+            .contains("metadata hash does not match"));
+    }
+
     #[tokio::test]
     async fn secret_persists() {
         let temp = tempfile::tempdir().unwrap();
@@ -90,6 +145,7 @@ mod tests {
         assert!(path_to_string(Path::new("../escape"), true).is_err());
         assert!(path_to_string(Path::new("/absolute"), true).is_err());
         assert!(validated_relative_string("bad\\name").is_err());
+        assert!(validated_relative_string("").is_err());
     }
 
     #[test]
@@ -121,6 +177,61 @@ mod tests {
         assert_eq!(content, "fd-backed import");
     }
 
+    #[cfg(unix)]
+    #[test]
+    fn file_descriptor_source_rejects_invalid_values() {
+        assert!(collect_import_files(vec![ShareSource {
+            kind: SourceKind::FileDescriptor,
+            value: "not-an-fd".to_string(),
+            display_name: Some("from-fd.txt".to_string()),
+            is_directory: false,
+        }])
+        .is_err());
+
+        assert!(collect_import_files(vec![ShareSource {
+            kind: SourceKind::FileDescriptor,
+            value: "-1".to_string(),
+            display_name: Some("from-fd.txt".to_string()),
+            is_directory: false,
+        }])
+        .is_err());
+    }
+
+    #[test]
+    fn android_content_uri_must_be_opened_by_platform_code() {
+        let error = collect_import_files(vec![ShareSource {
+            kind: SourceKind::AndroidContentUri,
+            value: "content://media/item".to_string(),
+            display_name: Some("from-uri.txt".to_string()),
+            is_directory: false,
+        }])
+        .unwrap_err()
+        .to_string();
+        assert!(error.contains("ParcelFileDescriptor"));
+    }
+
+    #[test]
+    fn directory_sources_preserve_safe_relative_names() {
+        let temp = tempfile::tempdir().unwrap();
+        let root = temp.path().join("picked");
+        std::fs::create_dir_all(root.join("nested")).unwrap();
+        std::fs::write(root.join("nested").join("a.txt"), b"a").unwrap();
+        std::fs::write(root.join("b.txt"), b"b").unwrap();
+
+        let mut files = collect_import_files(vec![ShareSource {
+            kind: SourceKind::Path,
+            value: root.to_string_lossy().to_string(),
+            display_name: Some("Album".to_string()),
+            is_directory: true,
+        }])
+        .unwrap();
+        files.sort_by(|a, b| a.collection_name.cmp(&b.collection_name));
+
+        assert_eq!(default_collection_name(&files), "Album");
+        assert_eq!(files[0].collection_name, "Album/b.txt");
+        assert_eq!(files[1].collection_name, "Album/nested/a.txt");
+    }
+
     #[test]
     fn can_initialize_core() {
         let temp = tempfile::tempdir().unwrap();
@@ -138,17 +249,18 @@ mod tests {
     async fn repository_persists_transfers_and_events() {
         let temp = tempfile::tempdir().unwrap();
         let repository = Repository::open(temp.path()).await.unwrap();
+        assert_eq!(repository.schema_version().await.unwrap(), 1);
         repository
-            .upsert_transfer(
+            .upsert_transfer(crate::repository::TransferUpsert {
-                7,
+                transfer_id: 7,
-                "send",
+                direction: "send",
-                "sharing",
+                status: "sharing",
-                Some("demo"),
+                transfer_name: Some("demo"),
-                Some("hash"),
+                content_hash: Some("hash"),
-                Some("ticket"),
+                ticket: Some("ticket"),
-                1,
+                file_count: 1,
-                12,
+                total_size: 12,
-            )
+            })
             .await
             .unwrap();
         repository
@@ -172,6 +284,12 @@ mod tests {
         let events = repository.list_events(Some(7)).await.unwrap();
         assert_eq!(events.len(), 1);
         assert_eq!(events[0].kind, "created");
+
+        let reopened = Repository::open(temp.path()).await.unwrap();
+        let transfers = reopened.list_transfers().await.unwrap();
+        assert_eq!(transfers.len(), 1);
+        let events = reopened.list_events(Some(7)).await.unwrap();
+        assert_eq!(events[0].id, "event-1");
     }
 
     #[tokio::test]
@@ -193,5 +311,36 @@ mod tests {
             policy.decide(99, Some("node-a")).await,
             AccessDecision::Allow
         );
+        assert_eq!(
+            policy.decide(99, None).await,
+            AccessDecision::Deny {
+                reason: "missing-endpoint-id"
+            }
+        );
+    }
+
+    #[test]
+    fn invalid_receive_ticket_is_typed_and_persisted_as_event() {
+        let temp = tempfile::tempdir().unwrap();
+        let core = VnidropCore::initialize(
+            temp.path().to_string_lossy().to_string(),
+            Arc::new(TestSink),
+        )
+        .unwrap();
+
+        let error = core
+            .receive(
+                "not-a-ticket".to_string(),
+                temp.path().to_string_lossy().to_string(),
+                None,
+            )
+            .unwrap_err();
+        assert!(matches!(error, VnidropError::Ticket { .. }));
+
+        let events = core.list_events(None).unwrap();
+        assert!(events
+            .iter()
+            .any(|event| event.phase == "error" && event.kind == "invalid-ticket"));
+        core.shutdown();
     }
 }

crates/vnidrop/src/ticket.rs

@@ -8,6 +8,7 @@ use serde::{Deserialize, Serialize};
 use crate::api::TransferMetadata;
 
 const VNIDROP_TICKET_PREFIX: &str = "vnd1:";
+const VNIDROP_TICKET_VERSION: u8 = 1;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub(crate) struct VnidropTicket {
@@ -19,7 +20,7 @@ pub(crate) struct VnidropTicket {
 impl VnidropTicket {
     pub(crate) fn new(blob_ticket: BlobTicket, metadata: TransferMetadata) -> Self {
         Self {
-            version: 1,
+            version: VNIDROP_TICKET_VERSION,
             blob_ticket: blob_ticket.to_string(),
             metadata,
         }
@@ -55,8 +56,26 @@ pub(crate) fn parse_transfer_ticket(value: &str) -> Result<ParsedTransferTicket>
     let normalized = normalize_ticket_input(value);
     if normalized.starts_with(VNIDROP_TICKET_PREFIX) {
         let ticket = VnidropTicket::decode(&normalized)?;
+        if ticket.version != VNIDROP_TICKET_VERSION {
+            anyhow::bail!("unsupported VniDrop ticket version {}", ticket.version);
+        }
+        if ticket.metadata.version != VNIDROP_TICKET_VERSION {
+            anyhow::bail!(
+                "unsupported VniDrop metadata version {}",
+                ticket.metadata.version
+            );
+        }
+        if ticket.metadata.transfer_id == 0 {
+            anyhow::bail!("VniDrop ticket metadata is missing a valid transfer id");
+        }
+        if ticket.metadata.transfer_name.trim().is_empty() {
+            anyhow::bail!("VniDrop ticket metadata is missing a transfer name");
+        }
         let blob_ticket = BlobTicket::from_str(&ticket.blob_ticket)
             .context("invalid BlobTicket inside VniDrop ticket")?;
+        if ticket.metadata.content_hash != blob_ticket.hash().to_string() {
+            anyhow::bail!("VniDrop ticket metadata hash does not match BlobTicket hash");
+        }
         return Ok(ParsedTransferTicket {
             blob_ticket,
             metadata: Some(ticket.metadata),
@@ -71,5 +90,8 @@ pub(crate) fn parse_transfer_ticket(value: &str) -> Result<ParsedTransferTicket>
 }
 
 fn normalize_ticket_input(value: &str) -> String {
+    // Tickets are commonly copied from text views or chat apps that insert line
+    // breaks.  Strip whitespace only; other corrupt characters should still be
+    // rejected by the base64 or BlobTicket decoders.
     value.chars().filter(|char| !char.is_whitespace()).collect()
 }

crates/vnidrop/tests/local_transfer.rs

@@ -1,6 +1,9 @@
 use std::sync::{Arc, Mutex};
 
-use vnidrop::{CoreEvent, CoreEventSink, ShareMetadataInput, ShareSource, SourceKind, VnidropCore};
+use vnidrop::{
+    CoreEvent, CoreEventSink, ShareMetadataInput, ShareSource, SourceKind, TransferAccessMode,
+    VnidropCore,
+};
 
 #[derive(Default)]
 struct RecordingSink {
@@ -13,6 +16,12 @@ impl CoreEventSink for RecordingSink {
     }
 }
 
+impl RecordingSink {
+    fn events(&self) -> Vec<CoreEvent> {
+        self.events.lock().unwrap().clone()
+    }
+}
+
 #[test]
 fn two_local_cores_transfer_file() {
     let sender_dir = tempfile::tempdir().unwrap();
@@ -21,9 +30,11 @@ fn two_local_cores_transfer_file() {
     let source_path = sender_dir.path().join("hello.txt");
     std::fs::write(&source_path, b"hello from vnidrop").unwrap();
 
+    let sender_sink = Arc::new(RecordingSink::default());
+    let receiver_sink = Arc::new(RecordingSink::default());
     let sender = VnidropCore::initialize(
         sender_dir.path().join("core").to_string_lossy().to_string(),
-        Arc::new(RecordingSink::default()),
+        sender_sink.clone(),
     )
     .unwrap();
     let receiver = VnidropCore::initialize(
@@ -32,7 +43,7 @@ fn two_local_cores_transfer_file() {
             .join("core")
             .to_string_lossy()
             .to_string(),
-        Arc::new(RecordingSink::default()),
+        receiver_sink.clone(),
     )
     .unwrap();
 
@@ -68,3 +79,185 @@ fn two_local_cores_transfer_file() {
     sender.shutdown();
     receiver.shutdown();
 }
+
+#[test]
+fn two_local_cores_transfer_directory() {
+    let sender_dir = tempfile::tempdir().unwrap();
+    let receiver_dir = tempfile::tempdir().unwrap();
+    let output_dir = tempfile::tempdir().unwrap();
+    let source_root = sender_dir.path().join("photos");
+    std::fs::create_dir_all(source_root.join("nested")).unwrap();
+    std::fs::write(source_root.join("cover.txt"), b"cover").unwrap();
+    std::fs::write(source_root.join("nested").join("inside.txt"), b"inside").unwrap();
+
+    let sender = VnidropCore::initialize(
+        sender_dir.path().join("core").to_string_lossy().to_string(),
+        Arc::new(RecordingSink::default()),
+    )
+    .unwrap();
+    let receiver = VnidropCore::initialize(
+        receiver_dir
+            .path()
+            .join("core")
+            .to_string_lossy()
+            .to_string(),
+        Arc::new(RecordingSink::default()),
+    )
+    .unwrap();
+
+    let share = sender
+        .share_files(
+            vec![ShareSource {
+                kind: SourceKind::Path,
+                value: source_root.to_string_lossy().to_string(),
+                display_name: Some("photos".to_string()),
+                is_directory: true,
+            }],
+            ShareMetadataInput {
+                transfer_id: 8,
+                transfer_name: Some("photos".to_string()),
+                sender_name: Some("sender".to_string()),
+            },
+        )
+        .unwrap();
+
+    receiver
+        .receive(
+            share.ticket,
+            output_dir.path().to_string_lossy().to_string(),
+            Some("receiver".to_string()),
+        )
+        .unwrap();
+
+    assert_eq!(
+        std::fs::read(output_dir.path().join("photos").join("cover.txt")).unwrap(),
+        b"cover"
+    );
+    assert_eq!(
+        std::fs::read(
+            output_dir
+                .path()
+                .join("photos")
+                .join("nested")
+                .join("inside.txt")
+        )
+        .unwrap(),
+        b"inside"
+    );
+
+    sender.shutdown();
+    receiver.shutdown();
+}
+
+#[test]
+fn approval_required_denies_then_allows_receiver() {
+    let sender_dir = tempfile::tempdir().unwrap();
+    let receiver_dir = tempfile::tempdir().unwrap();
+    let denied_output = tempfile::tempdir().unwrap();
+    let allowed_output = tempfile::tempdir().unwrap();
+    let source_path = sender_dir.path().join("private.txt");
+    std::fs::write(&source_path, b"approved content").unwrap();
+
+    let sender_sink = Arc::new(RecordingSink::default());
+    let sender = VnidropCore::initialize(
+        sender_dir.path().join("core").to_string_lossy().to_string(),
+        sender_sink.clone(),
+    )
+    .unwrap();
+    let receiver = VnidropCore::initialize(
+        receiver_dir
+            .path()
+            .join("core")
+            .to_string_lossy()
+            .to_string(),
+        Arc::new(RecordingSink::default()),
+    )
+    .unwrap();
+
+    let share = sender
+        .share_files(
+            vec![ShareSource {
+                kind: SourceKind::Path,
+                value: source_path.to_string_lossy().to_string(),
+                display_name: Some("private.txt".to_string()),
+                is_directory: false,
+            }],
+            ShareMetadataInput {
+                transfer_id: 9,
+                transfer_name: Some("private".to_string()),
+                sender_name: None,
+            },
+        )
+        .unwrap();
+    sender
+        .set_transfer_access_mode(share.transfer_id, TransferAccessMode::ApprovalRequired)
+        .unwrap();
+
+    assert!(receiver
+        .receive(
+            share.ticket.clone(),
+            denied_output.path().to_string_lossy().to_string(),
+            Some("receiver".to_string()),
+        )
+        .is_err());
+    assert!(sender_sink
+        .events()
+        .iter()
+        .any(|event| event.phase == "access" && event.kind == "request-denied"));
+
+    sender
+        .approve_endpoint_for_transfer(share.transfer_id, receiver.status().endpoint_id)
+        .unwrap();
+    receiver
+        .receive(
+            share.ticket,
+            allowed_output.path().to_string_lossy().to_string(),
+            Some("receiver".to_string()),
+        )
+        .unwrap();
+    assert_eq!(
+        std::fs::read(allowed_output.path().join("private.txt")).unwrap(),
+        b"approved content"
+    );
+
+    sender.shutdown();
+    receiver.shutdown();
+}
+
+#[test]
+fn cancelling_share_updates_status_and_events() {
+    let sender_dir = tempfile::tempdir().unwrap();
+    let source_path = sender_dir.path().join("cancel.txt");
+    std::fs::write(&source_path, b"cancel me").unwrap();
+    let sink = Arc::new(RecordingSink::default());
+    let sender = VnidropCore::initialize(
+        sender_dir.path().join("core").to_string_lossy().to_string(),
+        sink.clone(),
+    )
+    .unwrap();
+
+    let share = sender
+        .share_files(
+            vec![ShareSource {
+                kind: SourceKind::Path,
+                value: source_path.to_string_lossy().to_string(),
+                display_name: Some("cancel.txt".to_string()),
+                is_directory: false,
+            }],
+            ShareMetadataInput {
+                transfer_id: 10,
+                transfer_name: Some("cancel".to_string()),
+                sender_name: None,
+            },
+        )
+        .unwrap();
+    sender.cancel_transfer(share.transfer_id).unwrap();
+
+    let transfers = sender.list_transfers().unwrap();
+    assert_eq!(transfers[0].status, "stopped");
+    assert!(sink
+        .events()
+        .iter()
+        .any(|event| event.kind == "share-stopped"));
+    sender.shutdown();
+}