diff --git a/crates/vnidrop/CORE_FLOW.md b/crates/vnidrop/CORE_FLOW.md index f63a556..96cf5bd 100644 --- a/crates/vnidrop/CORE_FLOW.md +++ b/crates/vnidrop/CORE_FLOW.md @@ -48,3 +48,36 @@ bytes through Kotlin memory. descriptor; Rust duplicates the descriptor before streaming. - iOS starts the security-scoped URL lease in Kotlin and keeps it alive while Rust streams from the accessible file URL/path. + +## Durability And Filesystem Policy + +- SQLite records have a local UUID in addition to the protocol transfer ID. + Schema-v2 records are migrated in place and keep their tickets and history. +- Imports and receives are recorded before work begins. A process restart marks + interrupted work failed and expires approval requests that no longer have an + in-memory responder. +- Persisted shares are restored only when their root collection is complete and + readable. Missing or corrupt roots fail closed and emit a recovery event. +- Receive destinations use a no-overwrite policy. Rust writes a uniquely named + temporary file in the destination directory, syncs it, and atomically + publishes it with a no-clobber hard link. Failure or cancellation removes the + temporary file. Stale VniDrop temporary files are cleaned on later writes. +- Foreign output sinks receive exactly one terminal callback after a successful + `start_file`: `finish_file` or `abort_file`. + +## Blob Retention Policy + +Stopping a share immediately removes its provider mapping and approval state, +so neither VniDrop nor legacy blob tickets can read it. Physical blob chunks are +not force-deleted at stop time because content-addressed chunks may be shared by +another active collection. They remain eligible for the blob store's garbage +collection. Restart reconciliation never restores a stopped share. + +## Resource Limits + +`CoreLimits` controls source count, collection files and bytes, path and ticket +sizes, metadata, retained events, pending approvals, concurrent transfers, and +the event persistence queue. `initialize` uses conservative defaults; +`initialize_with_limits` supports stricter deployments and tests. Cheap limits +are checked before durable or network work, while remote collection limits are +checked before downloading file content. diff --git a/crates/vnidrop/src/access_policy.rs b/crates/vnidrop/src/access_policy.rs index 6a7cc56..1099ae0 100644 --- a/crates/vnidrop/src/access_policy.rs +++ b/crates/vnidrop/src/access_policy.rs @@ -90,6 +90,20 @@ impl AccessPolicy { } } +pub(crate) fn mode_from_storage(value: &str) -> TransferAccessMode { + match value { + "public" => TransferAccessMode::Public, + _ => TransferAccessMode::ApprovalRequired, + } +} + +pub(crate) fn mode_to_storage(mode: &TransferAccessMode) -> &'static str { + match mode { + TransferAccessMode::Public => "public", + TransferAccessMode::ApprovalRequired => "approval_required", + } +} + #[derive(Debug, Clone)] struct ApprovalSession { expires_at: Option, diff --git a/crates/vnidrop/src/api.rs b/crates/vnidrop/src/api.rs index dc01589..fabd451 100644 --- a/crates/vnidrop/src/api.rs +++ b/crates/vnidrop/src/api.rs @@ -1,8 +1,96 @@ +use anyhow::Context; use iroh_blobs::Hash; use serde::{Deserialize, Serialize}; use crate::util::{non_empty, now_ms}; +#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)] +pub struct CoreLimits { + pub max_sources: u64, + pub max_collection_files: u64, + pub max_total_bytes: u64, + pub max_path_bytes: u64, + pub max_ticket_bytes: u64, + pub max_metadata_bytes: u64, + pub max_events: u64, + pub max_pending_approvals: u64, + pub max_concurrent_transfers: u64, + pub event_queue_capacity: u64, +} + +impl Default for CoreLimits { + fn default() -> Self { + Self { + max_sources: 128, + max_collection_files: 10_000, + max_total_bytes: 1024 * 1024 * 1024 * 1024, + max_path_bytes: 4_096, + max_ticket_bytes: 1024 * 1024, + max_metadata_bytes: 16 * 1024, + max_events: 500, + max_pending_approvals: 1_024, + max_concurrent_transfers: 8, + event_queue_capacity: 1_024, + } + } +} + +impl CoreLimits { + pub(crate) fn validate(&self) -> anyhow::Result<()> { + let positive = [ + ("max_sources", self.max_sources), + ("max_collection_files", self.max_collection_files), + ("max_total_bytes", self.max_total_bytes), + ("max_path_bytes", self.max_path_bytes), + ("max_ticket_bytes", self.max_ticket_bytes), + ("max_metadata_bytes", self.max_metadata_bytes), + ("max_events", self.max_events), + ("max_pending_approvals", self.max_pending_approvals), + ("max_concurrent_transfers", self.max_concurrent_transfers), + ("event_queue_capacity", self.event_queue_capacity), + ]; + for (name, value) in positive { + if value == 0 { + anyhow::bail!("core limit {name} must be greater than zero"); + } + } + for (name, value) in [ + ("max_pending_approvals", self.max_pending_approvals), + ("max_concurrent_transfers", self.max_concurrent_transfers), + ("event_queue_capacity", self.event_queue_capacity), + ] { + usize::try_from(value) + .with_context(|| format!("core limit {name} exceeds platform capacity"))?; + } + if self.max_total_bytes > i64::MAX as u64 || self.max_events > i64::MAX as u64 { + anyhow::bail!("SQLite-backed limits must fit in a signed 64-bit integer"); + } + Ok(()) + } + + pub(crate) fn validate_metadata_text( + &self, + field: &str, + value: Option<&str>, + ) -> anyhow::Result<()> { + if let Some(value) = value { + if value.len() as u64 > self.max_metadata_bytes { + anyhow::bail!( + "{field} is {} bytes, limit is {}", + value.len(), + self.max_metadata_bytes + ); + } + } + Ok(()) + } +} + +#[uniffi::export] +pub fn default_core_limits() -> CoreLimits { + CoreLimits::default() +} + #[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)] pub struct CoreEvent { pub id: String, @@ -29,6 +117,11 @@ pub trait ReceiveOutputSink: Send + Sync { bytes: Vec, ) -> Result<(), crate::error::VnidropError>; fn finish_file(&self, relative_path: String) -> Result<(), crate::error::VnidropError>; + fn abort_file( + &self, + relative_path: String, + reason: String, + ) -> Result<(), crate::error::VnidropError>; } #[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)] @@ -70,7 +163,9 @@ pub enum TransferAccessMode { #[derive(Debug, Clone, Serialize, Deserialize, uniffi::Record)] pub struct StoredTransfer { + pub local_id: String, pub transfer_id: u64, + pub peer_id: Option, pub direction: String, pub status: String, pub transfer_name: Option, diff --git a/crates/vnidrop/src/approval.rs b/crates/vnidrop/src/approval.rs index d63efac..31382a7 100644 --- a/crates/vnidrop/src/approval.rs +++ b/crates/vnidrop/src/approval.rs @@ -10,6 +10,7 @@ use crate::{ event_hub::EventHub, handshake::{HandshakeResponse, RequestTransfer}, repository::{ReceiverRequestInsert, Repository}, + transfer_state::ReceiverRequestStatus, util::now_ms, }; @@ -29,6 +30,8 @@ pub(crate) struct ApprovalService { event_hub: Arc, access_policy: Arc, pending: Arc>>>, + max_pending: usize, + max_metadata_bytes: u64, } impl ApprovalService { @@ -36,12 +39,16 @@ impl ApprovalService { repository: Repository, event_hub: Arc, access_policy: Arc, + max_pending: usize, + max_metadata_bytes: u64, ) -> Self { Self { repository, event_hub, access_policy, pending: Arc::new(Mutex::new(HashMap::new())), + max_pending, + max_metadata_bytes, } } @@ -52,7 +59,11 @@ impl ApprovalService { reason: Option, ) -> anyhow::Result<()> { let sender = self.pending.lock().await.remove(&request_id); - let status = if accepted { "accepted" } else { "refused" }; + let status = if accepted { + ReceiverRequestStatus::Accepted + } else { + ReceiverRequestStatus::Refused + }; self.repository .update_receiver_request_status(&request_id, status, reason.as_deref()) .await?; @@ -73,6 +84,25 @@ impl ApprovalService { remote_endpoint_id: String, request: RequestTransfer, ) -> HandshakeResponse { + let metadata_values = [ + request.transfer_hash.as_str(), + request.transfer_name.as_str(), + request.receiver_name.as_deref().unwrap_or_default(), + request.receiver_device_name.as_deref().unwrap_or_default(), + request.app_version.as_str(), + ]; + if metadata_values + .iter() + .any(|value| value.len() as u64 > self.max_metadata_bytes) + { + return self + .deny( + request.transfer_id, + remote_endpoint_id, + "metadata-too-large", + ) + .await; + } self.event_hub.emit_transfer( request.transfer_id, "send", @@ -112,7 +142,19 @@ impl ApprovalService { ) -> HandshakeResponse { let request_id = Uuid::new_v4().to_string(); let (tx, rx) = oneshot::channel(); - self.pending.lock().await.insert(request_id.clone(), tx); + let mut pending = self.pending.lock().await; + if pending.len() >= self.max_pending { + drop(pending); + return self + .deny( + request.transfer_id, + remote_endpoint_id, + "too-many-pending-approvals", + ) + .await; + } + pending.insert(request_id.clone(), tx); + drop(pending); let insert_result = self .repository @@ -189,7 +231,7 @@ impl ApprovalService { .repository .update_receiver_request_status( &request_id, - "expired", + ReceiverRequestStatus::Expired, Some("approval timed out"), ) .await; diff --git a/crates/vnidrop/src/event_hub.rs b/crates/vnidrop/src/event_hub.rs index 22ec182..a728762 100644 --- a/crates/vnidrop/src/event_hub.rs +++ b/crates/vnidrop/src/event_hub.rs @@ -9,9 +9,100 @@ use tokio::{ use crate::{ api::{CoreEvent, CoreEventSink}, repository::Repository, + transfer_state::TransferDirection, util::now_ms, }; +#[derive(Debug, Clone, Copy)] +enum EventScope { + Endpoint, + Transfer, +} + +impl EventScope { + const fn as_str(self) -> &'static str { + match self { + Self::Endpoint => "endpoint", + Self::Transfer => "transfer", + } + } +} + +#[derive(Debug, Clone, Copy)] +enum EventPhase { + Startup, + Recovery, + Shutdown, + Provider, + Error, + Import, + Ticket, + Lifecycle, + Network, + Download, + Export, + Access, + Handshake, + Approval, + Transfer, +} + +impl EventPhase { + fn parse(value: &str) -> Option { + match value { + "startup" => Some(Self::Startup), + "recovery" => Some(Self::Recovery), + "shutdown" => Some(Self::Shutdown), + "provider" => Some(Self::Provider), + "error" => Some(Self::Error), + "import" => Some(Self::Import), + "ticket" => Some(Self::Ticket), + "lifecycle" => Some(Self::Lifecycle), + "network" => Some(Self::Network), + "download" => Some(Self::Download), + "export" => Some(Self::Export), + "access" => Some(Self::Access), + "handshake" => Some(Self::Handshake), + "approval" => Some(Self::Approval), + "transfer" => Some(Self::Transfer), + _ => None, + } + } + + const fn as_str(self) -> &'static str { + match self { + Self::Startup => "startup", + Self::Recovery => "recovery", + Self::Shutdown => "shutdown", + Self::Provider => "provider", + Self::Error => "error", + Self::Import => "import", + Self::Ticket => "ticket", + Self::Lifecycle => "lifecycle", + Self::Network => "network", + Self::Download => "download", + Self::Export => "export", + Self::Access => "access", + Self::Handshake => "handshake", + Self::Approval => "approval", + Self::Transfer => "transfer", + } + } +} + +struct EventKind(String); + +impl EventKind { + fn parse(value: &str) -> Option { + let valid = !value.is_empty() + && value.len() <= 64 + && value + .bytes() + .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-'); + valid.then(|| Self(value.to_string())) + } +} + enum EventCommand { Persist(CoreEvent), Flush(oneshot::Sender<()>), @@ -20,19 +111,24 @@ enum EventCommand { pub(crate) struct EventHub { sink: Arc, - tx: mpsc::UnboundedSender, + tx: mpsc::Sender, join: TokioMutex>>, sequence: Mutex, } impl EventHub { - pub(crate) fn start(repository: Repository, sink: Arc) -> Self { - let (tx, mut rx) = mpsc::unbounded_channel(); + pub(crate) fn start( + repository: Repository, + sink: Arc, + queue_capacity: usize, + max_history: u64, + ) -> Self { + let (tx, mut rx) = mpsc::channel(queue_capacity); 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 { + if let Err(error) = repository.insert_event(&event, max_history).await { tracing::warn!(%error, event_id = %event.id, "failed to persist core event"); } } @@ -56,7 +152,15 @@ impl EventHub { } pub(crate) fn emit_endpoint(&self, phase: &str, kind: &str, data: Value) { - self.emit("endpoint", None, None, phase, kind, data); + let Some(phase) = EventPhase::parse(phase) else { + tracing::warn!(phase, "dropped event with unknown phase"); + return; + }; + let Some(kind) = EventKind::parse(kind) else { + tracing::warn!(kind, "dropped event with invalid kind"); + return; + }; + self.emit(EventScope::Endpoint, None, None, phase, kind, data); } pub(crate) fn emit_transfer( @@ -67,10 +171,22 @@ impl EventHub { kind: &str, data: Value, ) { + let Ok(direction) = TransferDirection::try_from(direction) else { + tracing::warn!(direction, "dropped event with unknown direction"); + return; + }; + let Some(phase) = EventPhase::parse(phase) else { + tracing::warn!(phase, "dropped event with unknown phase"); + return; + }; + let Some(kind) = EventKind::parse(kind) else { + tracing::warn!(kind, "dropped event with invalid kind"); + return; + }; self.emit( - "transfer", + EventScope::Transfer, Some(transfer_id), - Some(direction.to_string()), + Some(direction), phase, kind, data, @@ -79,14 +195,14 @@ impl EventHub { pub(crate) async fn flush(&self) { let (tx, rx) = oneshot::channel(); - if self.tx.send(EventCommand::Flush(tx)).is_ok() { + if self.tx.send(EventCommand::Flush(tx)).await.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() { + if self.tx.send(EventCommand::Shutdown(tx)).await.is_ok() { let _ = rx.await; } if let Some(join) = self.join.lock().await.take() { @@ -96,15 +212,20 @@ impl EventHub { fn emit( &self, - scope: &str, + scope: EventScope, transfer_id: Option, - direction: Option, - phase: &str, - kind: &str, + direction: Option, + phase: EventPhase, + kind: EventKind, data: Value, ) { let timestamp = now_ms(); - let mut sequence = self.sequence.lock().expect("event sequence lock poisoned"); + // A panic while formatting a previous event cannot invalidate an + // integer counter, so recover the guard instead of cascading a panic. + let mut sequence = self + .sequence + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); let id = format!("{timestamp}-{}", *sequence); *sequence += 1; drop(sequence); @@ -115,15 +236,15 @@ impl EventHub { let event = CoreEvent { id, timestamp, - scope: scope.to_string(), + scope: scope.as_str().to_string(), transfer_id, - direction, - phase: phase.to_string(), - kind: kind.to_string(), + direction: direction.map(|direction| direction.as_str().to_string()), + phase: phase.as_str().to_string(), + kind: kind.0, 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"); + if let Err(error) = self.tx.try_send(EventCommand::Persist(event.clone())) { + tracing::warn!(event_id = %event.id, %error, "event persistence queue dropped event"); } self.sink.on_event(event); } diff --git a/crates/vnidrop/src/filesystem.rs b/crates/vnidrop/src/filesystem.rs index 0180e14..44d0056 100644 --- a/crates/vnidrop/src/filesystem.rs +++ b/crates/vnidrop/src/filesystem.rs @@ -1,21 +1,27 @@ #[cfg(unix)] use std::os::fd::{FromRawFd, OwnedFd}; use std::{ - fs::File, + fs::{File, OpenOptions}, io::{self, Read, Write}, path::{Component, Path, PathBuf}, + time::{Duration, SystemTime}, }; +#[cfg(unix)] +use std::os::unix::fs::OpenOptionsExt; + use anyhow::{Context, Result}; use bytes::Bytes; use iroh_blobs::{api::TempTag, Hash}; +use uuid::Uuid; use crate::{ - api::{ShareSource, SourceKind}, + api::{CoreLimits, ShareSource, SourceKind}, util::non_empty, }; const STREAM_BUFFER_LEN: usize = 1024 * 1024; +const STALE_PART_AGE: Duration = Duration::from_secs(24 * 60 * 60); #[derive(Debug)] pub(crate) struct TransferImport { @@ -39,6 +45,109 @@ pub(crate) enum ImportSource { FileDescriptor(OwnedFd), } +pub(crate) struct AtomicOutputFile { + target: PathBuf, + temporary: PathBuf, + committed: bool, +} + +impl AtomicOutputFile { + pub(crate) fn create(output_dir: &Path, relative_path: &str) -> Result<(Self, File)> { + let target = safe_output_path(output_dir, relative_path)?; + let parent = target + .parent() + .context("output file must have a parent directory")?; + std::fs::create_dir_all(parent)?; + + let canonical_root = std::fs::canonicalize(output_dir)?; + let canonical_parent = std::fs::canonicalize(parent)?; + if !canonical_parent.starts_with(&canonical_root) { + anyhow::bail!("output path escapes the selected directory"); + } + cleanup_stale_temporary_files(parent, STALE_PART_AGE)?; + if std::fs::symlink_metadata(&target).is_ok() { + anyhow::bail!("destination already exists: {}", target.display()); + } + + let final_name = target + .file_name() + .and_then(|name| name.to_str()) + .context("output filename is not valid UTF-8")?; + let temporary = parent.join(format!(".{final_name}.vnidrop-{}.part", Uuid::new_v4())); + let mut options = OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + options.custom_flags(libc::O_NOFOLLOW); + let file = options + .open(&temporary) + .with_context(|| format!("failed to create {}", temporary.display()))?; + + // Recheck after opening the temporary file so a swapped ancestor is + // detected before bytes are published to the final destination. + let canonical_parent_after_open = std::fs::canonicalize(parent)?; + if canonical_parent_after_open != canonical_parent { + let _ = std::fs::remove_file(&temporary); + anyhow::bail!("output directory changed while opening destination"); + } + + Ok(( + Self { + target, + temporary, + committed: false, + }, + file, + )) + } + + pub(crate) fn commit(mut self) -> Result<()> { + // Creating a hard link is an atomic no-clobber publication on the same + // filesystem. It fails if another writer created the destination. + std::fs::hard_link(&self.temporary, &self.target) + .with_context(|| format!("failed to commit {}", self.target.display()))?; + self.committed = true; + if let Err(error) = std::fs::remove_file(&self.temporary) { + tracing::warn!(%error, path = %self.temporary.display(), "failed to remove committed temporary file"); + } + Ok(()) + } +} + +pub(crate) fn cleanup_stale_temporary_files( + directory: &Path, + minimum_age: Duration, +) -> Result { + let now = SystemTime::now(); + let mut removed = 0; + for entry in std::fs::read_dir(directory)? { + let entry = entry?; + let name = entry.file_name(); + let name = name.to_string_lossy(); + if !(name.starts_with('.') && name.contains(".vnidrop-") && name.ends_with(".part")) { + continue; + } + let metadata = entry.metadata()?; + let age = metadata + .modified() + .ok() + .and_then(|modified| now.duration_since(modified).ok()) + .unwrap_or_default(); + if age >= minimum_age && metadata.is_file() { + std::fs::remove_file(entry.path())?; + removed += 1; + } + } + Ok(removed) +} + +impl Drop for AtomicOutputFile { + fn drop(&mut self) { + if !self.committed { + let _ = std::fs::remove_file(&self.temporary); + } + } +} + impl ImportSource { pub(crate) fn open(self) -> Result { match self { @@ -51,7 +160,22 @@ impl ImportSource { } } +#[cfg(test)] pub(crate) fn collect_import_files(sources: Vec) -> Result> { + collect_import_files_with_limits(sources, &CoreLimits::default()) +} + +pub(crate) fn collect_import_files_with_limits( + sources: Vec, + limits: &CoreLimits, +) -> Result> { + if sources.len() as u64 > limits.max_sources { + anyhow::bail!( + "source count {} exceeds limit {}", + sources.len(), + limits.max_sources + ); + } let mut files = Vec::new(); for source in sources { match source.kind { @@ -118,6 +242,34 @@ pub(crate) fn collect_import_files(sources: Vec) -> Result limits.max_collection_files { + anyhow::bail!( + "collection file count {} exceeds limit {}", + files.len(), + limits.max_collection_files + ); + } + let mut known_total = 0u64; + for file in &files { + if file.collection_name.len() as u64 > limits.max_path_bytes { + anyhow::bail!( + "collection path exceeds {} bytes: {}", + limits.max_path_bytes, + file.collection_name + ); + } + if let ImportSource::Path(path) = &file.source { + known_total = known_total + .checked_add(std::fs::metadata(path)?.len()) + .context("collection size overflow")?; + if known_total > limits.max_total_bytes { + anyhow::bail!( + "collection size {known_total} exceeds limit {}", + limits.max_total_bytes + ); + } + } + } Ok(files) } @@ -212,7 +364,7 @@ where pub(crate) fn write_stream_to_blocking_writer( mut writer: W, rx: async_channel::Receiver>>, -) -> io::Result<()> +) -> io::Result where W: Write, { @@ -222,12 +374,13 @@ where None => break, } } - writer.flush() + writer.flush()?; + Ok(writer) } -pub(crate) async fn wait_for_writer( - task: std::thread::JoinHandle>, -) -> Result> { +pub(crate) async fn wait_for_writer( + task: std::thread::JoinHandle>, +) -> Result> { tokio::task::spawn_blocking(move || { task.join() .map_err(|_| anyhow::anyhow!("export writer thread panicked")) diff --git a/crates/vnidrop/src/lib.rs b/crates/vnidrop/src/lib.rs index 932ac74..d6bce50 100644 --- a/crates/vnidrop/src/lib.rs +++ b/crates/vnidrop/src/lib.rs @@ -10,12 +10,13 @@ mod repository; mod runtime; mod secret; mod ticket; +mod transfer_state; mod util; pub use api::{ - CoreEvent, CoreEventSink, ReceiveOutputSink, ReceiverRequest, RuntimeStatus, - ShareMetadataInput, ShareResult, ShareSource, SourceKind, StoredTransfer, TicketInspection, - TransferAccessMode, TransferMetadata, + default_core_limits, CoreEvent, CoreEventSink, CoreLimits, ReceiveOutputSink, ReceiverRequest, + RuntimeStatus, ShareMetadataInput, ShareResult, ShareSource, SourceKind, StoredTransfer, + TicketInspection, TransferAccessMode, TransferMetadata, }; pub use error::VnidropError; pub use runtime::VnidropCore; diff --git a/crates/vnidrop/src/repository.rs b/crates/vnidrop/src/repository.rs index 1528b99..8cd05df 100644 --- a/crates/vnidrop/src/repository.rs +++ b/crates/vnidrop/src/repository.rs @@ -1,30 +1,58 @@ use std::{path::Path, str::FromStr}; -use anyhow::Result; +#[cfg(test)] +use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, +}; + +use anyhow::{Context, Result}; use sqlx::{ sqlite::{SqliteConnectOptions, SqlitePoolOptions}, Row, SqlitePool, }; +use uuid::Uuid; -use crate::api::{CoreEvent, ReceiverRequest, StoredTransfer}; -use crate::util::now_ms; +use crate::{ + api::{CoreEvent, ReceiverRequest, StoredTransfer}, + transfer_state::{ReceiverRequestStatus, TransferDirection, TransferStatus}, + util::now_ms, +}; -const SCHEMA_VERSION: i64 = 1; +const SCHEMA_VERSION: i64 = 3; #[derive(Debug, Clone)] pub(crate) struct Repository { pool: SqlitePool, + #[cfg(test)] + fail_next_write: Arc, } pub(crate) struct TransferUpsert<'a> { pub(crate) transfer_id: u64, - pub(crate) direction: &'a str, - pub(crate) status: &'a str, + pub(crate) peer_id: Option<&'a str>, + pub(crate) direction: TransferDirection, + pub(crate) status: TransferStatus, 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, + pub(crate) access_mode: &'a str, +} + +#[derive(Debug, Clone)] +pub(crate) struct PersistedShare { + pub(crate) transfer_id: u64, + pub(crate) content_hash: String, + pub(crate) access_mode: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct RecoveredTransfer { + pub(crate) transfer_id: u64, + pub(crate) direction: TransferDirection, + pub(crate) previous_status: TransferStatus, } pub(crate) struct ReceiverRequestInsert<'a> { @@ -47,7 +75,11 @@ impl Repository { .max_connections(4) .connect_with(options) .await?; - let repository = Self { pool }; + let repository = Self { + pool, + #[cfg(test)] + fail_next_write: Arc::new(AtomicBool::new(false)), + }; repository.ensure_schema().await?; Ok(repository) } @@ -59,6 +91,9 @@ impl Repository { r#" CREATE TABLE IF NOT EXISTS transfers ( transfer_id INTEGER PRIMARY KEY, + local_id TEXT NOT NULL, + protocol_transfer_id INTEGER NOT NULL, + peer_id TEXT, direction TEXT NOT NULL, status TEXT NOT NULL, transfer_name TEXT, @@ -66,6 +101,7 @@ impl Repository { ticket TEXT, file_count INTEGER NOT NULL DEFAULT 0, total_size INTEGER NOT NULL DEFAULT 0, + access_mode TEXT NOT NULL DEFAULT 'approval_required', created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); @@ -74,6 +110,60 @@ impl Repository { .execute(&self.pool) .await?; + let columns = sqlx::query("PRAGMA table_info(transfers)") + .fetch_all(&self.pool) + .await?; + let has_access_mode = columns + .iter() + .any(|row| row.get::(1) == "access_mode"); + if !has_access_mode { + sqlx::query( + "ALTER TABLE transfers ADD COLUMN access_mode TEXT NOT NULL DEFAULT 'approval_required'", + ) + .execute(&self.pool) + .await?; + } + + let has_local_id = columns + .iter() + .any(|row| row.get::(1) == "local_id"); + if !has_local_id { + sqlx::query("ALTER TABLE transfers ADD COLUMN local_id TEXT") + .execute(&self.pool) + .await?; + } + let has_protocol_transfer_id = columns + .iter() + .any(|row| row.get::(1) == "protocol_transfer_id"); + if !has_protocol_transfer_id { + sqlx::query("ALTER TABLE transfers ADD COLUMN protocol_transfer_id INTEGER") + .execute(&self.pool) + .await?; + } + let has_peer_id = columns + .iter() + .any(|row| row.get::(1) == "peer_id"); + if !has_peer_id { + sqlx::query("ALTER TABLE transfers ADD COLUMN peer_id TEXT") + .execute(&self.pool) + .await?; + } + sqlx::query( + r#" + UPDATE transfers + SET local_id = COALESCE(local_id, 'legacy-' || transfer_id || '-' || direction), + protocol_transfer_id = COALESCE(protocol_transfer_id, transfer_id) + WHERE local_id IS NULL OR protocol_transfer_id IS NULL + "#, + ) + .execute(&self.pool) + .await?; + sqlx::query( + "CREATE UNIQUE INDEX IF NOT EXISTS idx_transfers_local_id ON transfers(local_id)", + ) + .execute(&self.pool) + .await?; + sqlx::query( r#" CREATE TABLE IF NOT EXISTS transfer_events ( @@ -137,55 +227,275 @@ impl Repository { Ok(row.get(0)) } - pub(crate) async fn upsert_transfer(&self, transfer: TransferUpsert<'_>) -> Result<()> { + pub(crate) async fn insert_transfer(&self, transfer: TransferUpsert<'_>) -> Result<()> { + self.maybe_fail_write()?; let now = now_ms(); sqlx::query( r#" INSERT INTO transfers ( - transfer_id, direction, status, transfer_name, content_hash, ticket, - file_count, total_size, created_at, updated_at + transfer_id, local_id, protocol_transfer_id, peer_id, direction, status, + transfer_name, content_hash, ticket, file_count, total_size, access_mode, + created_at, updated_at ) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?9) - ON CONFLICT(transfer_id) DO UPDATE SET - direction = excluded.direction, - status = excluded.status, - transfer_name = excluded.transfer_name, - content_hash = excluded.content_hash, - ticket = excluded.ticket, - file_count = excluded.file_count, - total_size = excluded.total_size, - updated_at = excluded.updated_at; + VALUES (?1, ?2, ?1, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?12); "#, ) - .bind(transfer.transfer_id as i64) - .bind(transfer.direction) - .bind(transfer.status) + .bind(to_db_id(transfer.transfer_id)?) + .bind(Uuid::new_v4().to_string()) + .bind(transfer.peer_id) + .bind(transfer.direction.as_str()) + .bind(transfer.status.as_str()) .bind(transfer.transfer_name) .bind(transfer.content_hash) .bind(transfer.ticket) - .bind(transfer.file_count as i64) - .bind(transfer.total_size as i64) + .bind(to_db_id(transfer.file_count)?) + .bind(to_db_id(transfer.total_size)?) + .bind(transfer.access_mode) .bind(now) .execute(&self.pool) .await?; Ok(()) } - pub(crate) async fn update_transfer_status( - &self, - transfer_id: u64, - status: &str, - ) -> Result<()> { - sqlx::query("UPDATE transfers SET status = ?1, updated_at = ?2 WHERE transfer_id = ?3") - .bind(status) - .bind(now_ms()) - .bind(transfer_id as i64) - .execute(&self.pool) - .await?; + pub(crate) async fn start_receive(&self, transfer: TransferUpsert<'_>) -> Result<()> { + self.maybe_fail_write()?; + if transfer.direction != TransferDirection::Receive + || transfer.status != TransferStatus::Receiving + { + anyhow::bail!("receive must start in the receiving state"); + } + let now = now_ms(); + let result = sqlx::query( + r#" + INSERT INTO transfers ( + transfer_id, local_id, protocol_transfer_id, peer_id, direction, status, + transfer_name, content_hash, ticket, file_count, total_size, access_mode, + created_at, updated_at + ) + VALUES (?1, ?2, ?1, ?3, 'receive', 'receiving', ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?10) + ON CONFLICT(transfer_id) DO UPDATE SET + status = 'receiving', + transfer_name = excluded.transfer_name, + content_hash = excluded.content_hash, + ticket = excluded.ticket, + file_count = excluded.file_count, + total_size = excluded.total_size, + access_mode = excluded.access_mode, + peer_id = excluded.peer_id, + updated_at = excluded.updated_at + WHERE transfers.direction = 'receive' + AND transfers.status IN ('done', 'failed', 'cancelled') + "#, + ) + .bind(to_db_id(transfer.transfer_id)?) + .bind(Uuid::new_v4().to_string()) + .bind(transfer.peer_id) + .bind(transfer.transfer_name) + .bind(transfer.content_hash) + .bind(transfer.ticket) + .bind(to_db_id(transfer.file_count)?) + .bind(to_db_id(transfer.total_size)?) + .bind(transfer.access_mode) + .bind(now) + .execute(&self.pool) + .await?; + require_one_changed(result.rows_affected(), "start receive")?; Ok(()) } - pub(crate) async fn insert_event(&self, event: &CoreEvent) -> Result<()> { + pub(crate) async fn complete_share_import(&self, transfer: TransferUpsert<'_>) -> Result<()> { + self.maybe_fail_write()?; + if transfer.direction != TransferDirection::Send + || transfer.status != TransferStatus::Sharing + { + anyhow::bail!("share import must complete in the sharing state"); + } + let result = sqlx::query( + r#" + UPDATE transfers + SET status = ?1, + transfer_name = ?2, + content_hash = ?3, + ticket = ?4, + file_count = ?5, + total_size = ?6, + access_mode = ?7, + updated_at = ?8 + WHERE transfer_id = ?9 + AND direction = 'send' + AND status = 'importing' + "#, + ) + .bind(transfer.status.as_str()) + .bind(transfer.transfer_name) + .bind(transfer.content_hash) + .bind(transfer.ticket) + .bind(to_db_id(transfer.file_count)?) + .bind(to_db_id(transfer.total_size)?) + .bind(transfer.access_mode) + .bind(now_ms()) + .bind(to_db_id(transfer.transfer_id)?) + .execute(&self.pool) + .await?; + require_one_changed(result.rows_affected(), "complete share import")?; + Ok(()) + } + + pub(crate) async fn transition_transfer_status( + &self, + transfer_id: u64, + expected: TransferStatus, + next: TransferStatus, + ) -> Result<()> { + self.maybe_fail_write()?; + if !expected.can_transition_to(next) { + anyhow::bail!( + "illegal transfer status transition: {} -> {}", + expected.as_str(), + next.as_str() + ); + } + let result = sqlx::query( + r#" + UPDATE transfers + SET status = ?1, updated_at = ?2 + WHERE transfer_id = ?3 AND status = ?4 + "#, + ) + .bind(next.as_str()) + .bind(now_ms()) + .bind(to_db_id(transfer_id)?) + .bind(expected.as_str()) + .execute(&self.pool) + .await?; + if result.rows_affected() == 0 { + let current = sqlx::query("SELECT status FROM transfers WHERE transfer_id = ?1") + .bind(to_db_id(transfer_id)?) + .fetch_optional(&self.pool) + .await?; + if current + .as_ref() + .map(|row| row.get::(0)) + .as_deref() + == Some(next.as_str()) + { + return Ok(()); + } + } + require_one_changed(result.rows_affected(), "transition transfer status")?; + Ok(()) + } + + pub(crate) async fn update_active_share_access_mode( + &self, + transfer_id: u64, + access_mode: &str, + ) -> Result<()> { + self.maybe_fail_write()?; + let result = sqlx::query( + r#" + UPDATE transfers + SET access_mode = ?1, updated_at = ?2 + WHERE transfer_id = ?3 + AND direction = 'send' + AND status = 'sharing' + "#, + ) + .bind(access_mode) + .bind(now_ms()) + .bind(to_db_id(transfer_id)?) + .execute(&self.pool) + .await?; + require_one_changed(result.rows_affected(), "update active share access mode")?; + Ok(()) + } + + pub(crate) async fn recover_interrupted_transfers(&self) -> Result> { + self.maybe_fail_write()?; + let mut transaction = self.pool.begin().await?; + let rows = sqlx::query( + r#" + SELECT transfer_id, direction, status + FROM transfers + WHERE status IN ('importing', 'receiving') + ORDER BY created_at ASC + "#, + ) + .fetch_all(&mut *transaction) + .await?; + let recovered = rows + .into_iter() + .map(|row| { + Ok(RecoveredTransfer { + transfer_id: row.get::("transfer_id") as u64, + direction: TransferDirection::try_from( + row.get::("direction").as_str(), + )?, + previous_status: TransferStatus::try_from( + row.get::("status").as_str(), + )?, + }) + }) + .collect::>>()?; + + if !recovered.is_empty() { + sqlx::query( + r#" + UPDATE transfers + SET status = 'failed', updated_at = ?1 + WHERE status IN ('importing', 'receiving') + "#, + ) + .bind(now_ms()) + .execute(&mut *transaction) + .await?; + } + transaction.commit().await?; + Ok(recovered) + } + + #[cfg(test)] + pub(crate) fn fail_next_write(&self) { + self.fail_next_write.store(true, Ordering::SeqCst); + } + + #[cfg(test)] + fn maybe_fail_write(&self) -> Result<()> { + if self.fail_next_write.swap(false, Ordering::SeqCst) { + anyhow::bail!("injected repository write failure"); + } + Ok(()) + } + + #[cfg(not(test))] + fn maybe_fail_write(&self) -> Result<()> { + Ok(()) + } + + pub(crate) async fn list_active_shares(&self) -> Result> { + let rows = sqlx::query( + r#" + SELECT transfer_id, content_hash, access_mode + FROM transfers + WHERE direction = 'send' + AND status = 'sharing' + AND content_hash IS NOT NULL + "#, + ) + .fetch_all(&self.pool) + .await?; + Ok(rows + .into_iter() + .map(|row| PersistedShare { + transfer_id: row.get::(0) as u64, + content_hash: row.get::(1), + access_mode: row.get::(2), + }) + .collect()) + } + + pub(crate) async fn insert_event(&self, event: &CoreEvent, max_history: u64) -> Result<()> { + let mut transaction = self.pool.begin().await?; sqlx::query( r#" INSERT OR REPLACE INTO transfer_events ( @@ -197,13 +507,25 @@ impl Repository { .bind(&event.id) .bind(event.timestamp) .bind(&event.scope) - .bind(event.transfer_id.map(|value| value as i64)) + .bind(event.transfer_id.map(to_db_id).transpose()?) .bind(&event.direction) .bind(&event.phase) .bind(&event.kind) .bind(&event.data_json) - .execute(&self.pool) + .execute(&mut *transaction) .await?; + sqlx::query( + r#" + DELETE FROM transfer_events + WHERE id NOT IN ( + SELECT id FROM transfer_events ORDER BY timestamp DESC, id DESC LIMIT ?1 + ) + "#, + ) + .bind(to_db_id(max_history)?) + .execute(&mut *transaction) + .await?; + transaction.commit().await?; Ok(()) } @@ -222,7 +544,7 @@ impl Repository { "#, ) .bind(request.id) - .bind(request.transfer_id as i64) + .bind(to_db_id(request.transfer_id)?) .bind(request.remote_endpoint_id) .bind(request.transfer_name) .bind(request.receiver_name) @@ -237,7 +559,7 @@ impl Repository { pub(crate) async fn update_receiver_request_status( &self, id: &str, - status: &str, + status: ReceiverRequestStatus, reason: Option<&str>, ) -> Result<()> { let result = sqlx::query( @@ -248,7 +570,7 @@ impl Repository { AND status = 'requested' "#, ) - .bind(status) + .bind(status.as_str()) .bind(reason) .bind(now_ms()) .bind(id) @@ -260,6 +582,21 @@ impl Repository { Ok(()) } + pub(crate) async fn expire_pending_receiver_requests(&self, reason: &str) -> Result { + let result = sqlx::query( + r#" + UPDATE receiver_requests + SET status = 'expired', reason = ?1, responded_at = ?2 + WHERE status = 'requested' + "#, + ) + .bind(reason) + .bind(now_ms()) + .execute(&self.pool) + .await?; + Ok(result.rows_affected()) + } + pub(crate) async fn list_receiver_requests( &self, transfer_id: u64, @@ -274,7 +611,7 @@ impl Repository { ORDER BY requested_at DESC "#, ) - .bind(transfer_id as i64) + .bind(to_db_id(transfer_id)?) .fetch_all(&self.pool) .await?; Ok(rows.into_iter().map(row_to_receiver_request).collect()) @@ -292,7 +629,7 @@ impl Repository { ) "#, ) - .bind(transfer_id as i64) + .bind(to_db_id(transfer_id)?) .bind(content_hash) .fetch_one(&self.pool) .await?; @@ -303,6 +640,7 @@ impl Repository { let rows = sqlx::query( r#" SELECT transfer_id, direction, status, transfer_name, content_hash, ticket, + local_id, protocol_transfer_id, peer_id, file_count, total_size, created_at, updated_at FROM transfers ORDER BY updated_at DESC @@ -310,10 +648,14 @@ impl Repository { ) .fetch_all(&self.pool) .await?; - Ok(rows.into_iter().map(row_to_transfer).collect()) + rows.into_iter().map(row_to_transfer).collect() } - pub(crate) async fn list_events(&self, transfer_id: Option) -> Result> { + pub(crate) async fn list_events( + &self, + transfer_id: Option, + limit: u64, + ) -> Result> { let rows = if let Some(transfer_id) = transfer_id { sqlx::query( r#" @@ -321,9 +663,11 @@ impl Repository { FROM transfer_events WHERE transfer_id = ?1 ORDER BY timestamp ASC + LIMIT ?2 "#, ) - .bind(transfer_id as i64) + .bind(to_db_id(transfer_id)?) + .bind(to_db_id(limit)?) .fetch_all(&self.pool) .await? } else { @@ -332,9 +676,10 @@ impl Repository { SELECT id, timestamp, scope, transfer_id, direction, phase, kind, data_json FROM transfer_events ORDER BY timestamp DESC - LIMIT 500 + LIMIT ?1 "#, ) + .bind(to_db_id(limit)?) .fetch_all(&self.pool) .await? }; @@ -342,11 +687,30 @@ impl Repository { } } -fn row_to_transfer(row: sqlx::sqlite::SqliteRow) -> StoredTransfer { - StoredTransfer { +fn require_one_changed(rows_affected: u64, operation: &str) -> Result<()> { + if rows_affected != 1 { + anyhow::bail!("{operation} expected one matching transfer, changed {rows_affected}"); + } + Ok(()) +} + +fn to_db_id(value: u64) -> Result { + i64::try_from(value).context("transfer id exceeds SQLite signed integer range") +} + +fn row_to_transfer(row: sqlx::sqlite::SqliteRow) -> Result { + let direction = row.get::("direction"); + let status = row.get::("status"); + Ok(StoredTransfer { + local_id: row.get("local_id"), transfer_id: row.get::("transfer_id") as u64, - direction: row.get("direction"), - status: row.get("status"), + peer_id: row.get("peer_id"), + direction: TransferDirection::try_from(direction.as_str())? + .as_str() + .to_string(), + status: TransferStatus::try_from(status.as_str())? + .as_str() + .to_string(), transfer_name: row.get("transfer_name"), content_hash: row.get("content_hash"), ticket: row.get("ticket"), @@ -354,7 +718,7 @@ fn row_to_transfer(row: sqlx::sqlite::SqliteRow) -> StoredTransfer { total_size: row.get::("total_size") as u64, created_at: row.get("created_at"), updated_at: row.get("updated_at"), - } + }) } fn row_to_event(row: sqlx::sqlite::SqliteRow) -> CoreEvent { @@ -373,6 +737,7 @@ fn row_to_event(row: sqlx::sqlite::SqliteRow) -> CoreEvent { } fn row_to_receiver_request(row: sqlx::sqlite::SqliteRow) -> ReceiverRequest { + let status = row.get::("status"); ReceiverRequest { id: row.get("id"), transfer_id: row.get::("transfer_id") as u64, @@ -381,7 +746,9 @@ fn row_to_receiver_request(row: sqlx::sqlite::SqliteRow) -> ReceiverRequest { receiver_name: row.get("receiver_name"), receiver_device_name: row.get("receiver_device_name"), app_version: row.get("app_version"), - status: row.get("status"), + status: ReceiverRequestStatus::try_from(status.as_str()) + .map(|status| status.as_str().to_string()) + .unwrap_or_else(|_| "unknown".to_string()), reason: row.get("reason"), requested_at: row.get("requested_at"), responded_at: row.get("responded_at"), diff --git a/crates/vnidrop/src/runtime.rs b/crates/vnidrop/src/runtime.rs index fa48753..0e74216 100644 --- a/crates/vnidrop/src/runtime.rs +++ b/crates/vnidrop/src/runtime.rs @@ -1,8 +1,8 @@ use std::{ collections::HashMap, - fs::File, io, path::{Path, PathBuf}, + str::FromStr, sync::{ atomic::{AtomicBool, Ordering}, Arc, @@ -25,14 +25,14 @@ use iroh_blobs::{ use n0_future::BufferedStreamExt; use serde_json::json; use tokio::{ - sync::{mpsc, oneshot, Mutex as TokioMutex}, + sync::{mpsc, oneshot, Mutex as TokioMutex, Semaphore}, task::JoinHandle, }; use crate::{ - access_policy::{AccessDecision, AccessPolicy}, + access_policy::{mode_from_storage, mode_to_storage, AccessDecision, AccessPolicy}, api::{ - CoreEvent, CoreEventSink, ReceiveOutputSink, ReceiverRequest, RuntimeStatus, + CoreEvent, CoreEventSink, CoreLimits, ReceiveOutputSink, ReceiverRequest, RuntimeStatus, ShareMetadataInput, ShareResult, ShareSource, StoredTransfer, TicketInspection, TransferAccessMode, TransferMetadata, }, @@ -40,25 +40,19 @@ use crate::{ error::VnidropError, event_hub::EventHub, filesystem::{ - collect_import_files, default_collection_name, platform_path, - read_stream_from_blocking_reader, safe_output_path, validated_relative_string, - wait_for_writer, write_stream_to_blocking_writer, TransferImport, + collect_import_files_with_limits, default_collection_name, platform_path, + read_stream_from_blocking_reader, validated_relative_string, wait_for_writer, + write_stream_to_blocking_writer, AtomicOutputFile, TransferImport, }, handshake::{HandshakeResponse, HandshakeService}, logging::init_logging, repository::{Repository, TransferUpsert}, secret::load_or_create_secret, - ticket::{parse_transfer_ticket, ParsedTransferTicket, VnidropTicket}, + ticket::{parse_transfer_ticket_with_limits, ParsedTransferTicket, VnidropTicket}, + transfer_state::{TransferDirection, TransferStatus}, 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, @@ -74,20 +68,82 @@ struct CoreInner { repository: Repository, event_hub: Arc, approval: ApprovalService, + limits: CoreLimits, + transfer_slots: Semaphore, access_policy: Arc, - active_transfers: TokioMutex>>, - active_shares: TokioMutex>, + active_transfers: TokioMutex>, + // Newly imported shares retain a TempTag for the lifetime of this process. + // Restored shares have no in-memory tag, but remain tracked so they can be + // counted and explicitly revoked after a restart. + active_shares: TokioMutex>>, hash_to_transfer: TokioMutex>, connection_endpoints: TokioMutex>, provider_task: TokioMutex>>, shutdown_started: AtomicBool, } +struct ActiveTransfer { + direction: TransferDirection, + cancel: oneshot::Sender<()>, +} + enum ReceiveTarget { Directory(PathBuf), OutputSink(Arc), } +struct OutputSinkFile<'a> { + sink: &'a dyn ReceiveOutputSink, + relative_path: String, + terminal: bool, +} + +impl<'a> OutputSinkFile<'a> { + fn start(sink: &'a dyn ReceiveOutputSink, relative_path: String) -> Result { + sink.start_file(relative_path.clone()) + .map_err(|error| anyhow::anyhow!(error.to_string()))?; + Ok(Self { + sink, + relative_path, + terminal: false, + }) + } + + fn write(&self, bytes: Vec) -> Result<()> { + self.sink + .write_chunk(self.relative_path.clone(), bytes) + .map_err(|error| anyhow::anyhow!(error.to_string())) + } + + fn finish(mut self) -> Result<()> { + // finish_file is terminal even when the foreign implementation reports + // an error; implementations must release their open resource before + // returning so Rust never invokes two terminal callbacks. + self.terminal = true; + self.sink + .finish_file(self.relative_path.clone()) + .map_err(|error| anyhow::anyhow!(error.to_string())) + } +} + +impl Drop for OutputSinkFile<'_> { + fn drop(&mut self) { + if !self.terminal { + self.terminal = true; + if let Err(error) = self.sink.abort_file( + self.relative_path.clone(), + "transfer interrupted before file completion".to_string(), + ) { + tracing::warn!( + %error, + relative_path = %self.relative_path, + "failed to abort receive output sink file" + ); + } + } + } +} + #[uniffi::export] impl VnidropCore { #[uniffi::constructor] @@ -95,13 +151,23 @@ impl VnidropCore { app_data_dir: String, event_sink: Arc, ) -> Result, VnidropError> { + Self::initialize_with_limits(app_data_dir, event_sink, CoreLimits::default()) + } + + #[uniffi::constructor] + pub fn initialize_with_limits( + app_data_dir: String, + event_sink: Arc, + limits: CoreLimits, + ) -> Result, VnidropError> { + limits.validate().map_err(VnidropError::initialization)?; let runtime = tokio::runtime::Builder::new_multi_thread() .enable_all() .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)) + .block_on(CoreInner::start(app_data_dir, event_sink, limits)) .map_err(VnidropError::initialization)?; Ok(Arc::new(Self { runtime, inner })) } @@ -126,8 +192,8 @@ impl VnidropCore { output_dir: String, receiver_name: Option, ) -> Result<(), VnidropError> { - if let Err(error) = - parse_transfer_ticket(&ticket).context("failed to parse transfer ticket") + if let Err(error) = parse_transfer_ticket_with_limits(&ticket, &self.inner.limits) + .context("failed to parse transfer ticket") { self.runtime.block_on(async { self.inner.emit_endpoint( @@ -151,8 +217,8 @@ impl VnidropCore { output_sink: Arc, receiver_name: Option, ) -> Result<(), VnidropError> { - if let Err(error) = - parse_transfer_ticket(&ticket).context("failed to parse transfer ticket") + if let Err(error) = parse_transfer_ticket_with_limits(&ticket, &self.inner.limits) + .context("failed to parse transfer ticket") { self.runtime.block_on(async { self.inner.emit_endpoint( @@ -234,7 +300,7 @@ impl VnidropCore { } pub fn inspect_ticket(&self, ticket: String) -> Result { - let parsed = parse_transfer_ticket(&ticket) + let parsed = parse_transfer_ticket_with_limits(&ticket, &self.inner.limits) .context("failed to parse transfer ticket") .map_err(VnidropError::ticket)?; Ok(TicketInspection { @@ -254,7 +320,11 @@ impl VnidropCore { } impl CoreInner { - async fn start(app_data_dir: PathBuf, event_sink: Arc) -> Result> { + async fn start( + app_data_dir: PathBuf, + event_sink: Arc, + limits: CoreLimits, + ) -> Result> { tokio::fs::create_dir_all(&app_data_dir).await?; init_logging(&app_data_dir)?; let secret_key = load_or_create_secret(&app_data_dir).await?; @@ -271,10 +341,76 @@ impl CoreInner { // 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 event_hub = Arc::new(EventHub::start(repository.clone(), event_sink)); + let recovered_transfers = repository.recover_interrupted_transfers().await?; + let event_hub = Arc::new(EventHub::start( + repository.clone(), + event_sink, + limits.event_queue_capacity as usize, + limits.max_events, + )); + for recovered in recovered_transfers { + event_hub.emit_transfer( + recovered.transfer_id, + recovered.direction.as_str(), + "recovery", + "interrupted-transfer-failed", + json!({ "previous_status": recovered.previous_status.as_str() }), + ); + } + let expired_requests = repository + .expire_pending_receiver_requests("application restarted before approval") + .await?; + if expired_requests > 0 { + event_hub.emit_endpoint( + "recovery", + "pending-approvals-expired", + json!({ "count": expired_requests }), + ); + } let access_policy = AccessPolicy::new(); - let approval = - ApprovalService::new(repository.clone(), event_hub.clone(), access_policy.clone()); + // Restore share ownership and access mode before the router can serve + // any request. Unknown persisted modes fail closed in mode_from_storage. + let mut restored_hashes = HashMap::new(); + let mut restored_active_shares = HashMap::new(); + for share in repository.list_active_shares().await? { + let transfer_id = share.transfer_id; + let valid_root = match Hash::from_str(&share.content_hash) { + Ok(hash) => { + store.blobs().has(hash).await.unwrap_or(false) + && Collection::load(hash, store.as_ref()).await.is_ok() + } + Err(_) => false, + }; + if !valid_root { + repository + .transition_transfer_status( + transfer_id, + TransferStatus::Sharing, + TransferStatus::Failed, + ) + .await?; + event_hub.emit_transfer( + transfer_id, + TransferDirection::Send.as_str(), + "recovery", + "share-root-missing-or-corrupt", + json!({ "content_hash": share.content_hash }), + ); + continue; + } + restored_hashes.insert(share.content_hash, transfer_id); + restored_active_shares.insert(transfer_id, None); + access_policy + .set_mode(transfer_id, mode_from_storage(&share.access_mode)) + .await; + } + let approval = ApprovalService::new( + repository.clone(), + event_hub.clone(), + access_policy.clone(), + limits.max_pending_approvals as usize, + limits.max_metadata_bytes, + ); let handshake = HandshakeService::new(approval.clone()); let router = Router::builder(endpoint.clone()) .accept(iroh_blobs::ALPN, blobs) @@ -288,10 +424,12 @@ impl CoreInner { repository, event_hub, approval, + transfer_slots: Semaphore::new(limits.max_concurrent_transfers as usize), + limits, access_policy, active_transfers: TokioMutex::new(HashMap::new()), - active_shares: TokioMutex::new(HashMap::new()), - hash_to_transfer: TokioMutex::new(HashMap::new()), + active_shares: TokioMutex::new(restored_active_shares), + hash_to_transfer: TokioMutex::new(restored_hashes), connection_endpoints: TokioMutex::new(HashMap::new()), provider_task: TokioMutex::new(None), shutdown_started: AtomicBool::new(false), @@ -324,20 +462,73 @@ impl CoreInner { sources: Vec, metadata: ShareMetadataInput, ) -> Result { + let _permit = self + .transfer_slots + .acquire() + .await + .context("transfer limiter is closed")?; 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() }), + if sources.is_empty() { + anyhow::bail!("at least one source is required"); + } + if sources.len() as u64 > self.limits.max_sources { + anyhow::bail!( + "source count {} exceeds limit {}", + sources.len(), + self.limits.max_sources ); - let _ = self - .repository - .update_transfer_status(transfer_id, STATUS_FAILED) - .await; + } + self.limits + .validate_metadata_text("transfer name", metadata.transfer_name.as_deref())?; + self.limits + .validate_metadata_text("sender name", metadata.sender_name.as_deref())?; + self.repository + .insert_transfer(TransferUpsert { + transfer_id, + peer_id: None, + direction: TransferDirection::Send, + status: TransferStatus::Importing, + transfer_name: metadata.transfer_name.as_deref(), + content_hash: None, + ticket: None, + file_count: 0, + total_size: 0, + access_mode: mode_to_storage(&TransferAccessMode::ApprovalRequired), + }) + .await?; + let (cancel, mut cancelled) = oneshot::channel(); + self.active_transfers.lock().await.insert( + transfer_id, + ActiveTransfer { + direction: TransferDirection::Send, + cancel, + }, + ); + let (result, was_cancelled) = tokio::select! { + result = self.share_files_inner(sources, metadata) => (result, false), + _ = &mut cancelled => (Err(anyhow::anyhow!("transfer cancelled")), true), + }; + self.active_transfers.lock().await.remove(&transfer_id); + if let Err(error) = &result { + if was_cancelled { + self.emit_transfer(transfer_id, "send", "lifecycle", "cancelled", json!({})); + } else { + self.emit_transfer( + transfer_id, + "send", + "error", + "failed", + json!({ "reason": error.to_string() }), + ); + let _ = self + .repository + .transition_transfer_status( + transfer_id, + TransferStatus::Importing, + TransferStatus::Failed, + ) + .await; + } } result } @@ -347,10 +538,6 @@ impl CoreInner { sources: Vec, metadata: ShareMetadataInput, ) -> Result { - if sources.is_empty() { - anyhow::bail!("at least one source is required"); - } - self.emit_transfer( metadata.transfer_id, "send", @@ -376,30 +563,36 @@ impl CoreInner { let ticket = VnidropTicket::new(blob_ticket.clone(), ticket_metadata) .encode() .context("failed to encode VniDrop transfer ticket")?; + let content_hash = import.root_hash.to_string(); + // Persist the completed share before exposing it through the provider. + // The remaining in-memory registrations are infallible and can be + // reconstructed from SQLite if the process exits immediately after. + self.repository + .complete_share_import(TransferUpsert { + transfer_id: metadata.transfer_id, + peer_id: None, + direction: TransferDirection::Send, + status: TransferStatus::Sharing, + transfer_name: Some(&transfer_name), + content_hash: Some(&content_hash), + ticket: Some(&ticket), + file_count: import.file_count, + total_size: import.total_size, + access_mode: mode_to_storage(&TransferAccessMode::ApprovalRequired), + }) + .await?; self.hash_to_transfer .lock() .await - .insert(import.root_hash.to_string(), metadata.transfer_id); + .insert(content_hash, metadata.transfer_id); self.access_policy .set_mode(metadata.transfer_id, TransferAccessMode::ApprovalRequired) .await; self.active_shares .lock() .await - .insert(metadata.transfer_id, import.tag); - self.repository - .upsert_transfer(TransferUpsert { - transfer_id: metadata.transfer_id, - direction: "send", - status: STATUS_SHARING, - transfer_name: Some(&transfer_name), - content_hash: Some(&import.root_hash.to_string()), - ticket: Some(&ticket), - file_count: import.file_count, - total_size: import.total_size, - }) - .await?; + .insert(metadata.transfer_id, Some(import.tag)); self.emit_transfer( metadata.transfer_id, @@ -455,7 +648,13 @@ impl CoreInner { target: ReceiveTarget, receiver_name: Option, ) -> Result<()> { - let parsed = match parse_transfer_ticket(&ticket).context("failed to parse transfer ticket") + let _permit = self + .transfer_slots + .acquire() + .await + .context("transfer limiter is closed")?; + let parsed = match parse_transfer_ticket_with_limits(&ticket, &self.limits) + .context("failed to parse transfer ticket") { Ok(parsed) => parsed, Err(error) => { @@ -472,32 +671,45 @@ impl CoreInner { .as_ref() .map(|metadata| metadata.transfer_id) .unwrap_or_else(unique_transfer_id); + self.persist_receive_start(transfer_id, &parsed, receiver_name.as_deref()) + .await?; // 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() - .await - .insert(transfer_id, shutdown_tx); + self.active_transfers.lock().await.insert( + transfer_id, + ActiveTransfer { + direction: TransferDirection::Receive, + cancel: shutdown_tx, + }, + ); - let result = tokio::select! { - result = self.receive_inner(transfer_id, parsed, target, receiver_name) => result, - _ = &mut shutdown_rx => Err(anyhow::anyhow!("transfer cancelled")), + let (result, cancelled) = tokio::select! { + result = self.receive_inner(transfer_id, parsed, target, receiver_name) => (result, false), + _ = &mut shutdown_rx => (Err(anyhow::anyhow!("transfer cancelled")), true), }; 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; + if cancelled { + self.emit_transfer(transfer_id, "receive", "lifecycle", "cancelled", json!({})); + } else { + self.emit_transfer( + transfer_id, + "receive", + "error", + "failed", + json!({ "reason": error.to_string() }), + ); + let _ = self + .repository + .transition_transfer_status( + transfer_id, + TransferStatus::Receiving, + TransferStatus::Failed, + ) + .await; + } } result } @@ -509,44 +721,6 @@ impl CoreInner { target: ReceiveTarget, receiver_name: Option, ) -> Result<()> { - let metadata_json = - serde_json::to_value(&parsed.metadata).unwrap_or(serde_json::Value::Null); - self.emit_transfer( - transfer_id, - "receive", - "lifecycle", - "started", - json!({ - "metadata": metadata_json, - "receiver_name": receiver_name, - }), - ); - self.repository - .upsert_transfer(TransferUpsert { - transfer_id, - direction: "receive", - status: STATUS_RECEIVING, - transfer_name: parsed - .metadata - .as_ref() - .map(|metadata| metadata.transfer_name.as_str()), - content_hash: parsed - .metadata - .as_ref() - .map(|metadata| metadata.content_hash.as_str()), - ticket: None, - file_count: parsed - .metadata - .as_ref() - .map(|metadata| metadata.file_count) - .unwrap_or_default(), - total_size: parsed - .metadata - .as_ref() - .map(|metadata| metadata.total_size) - .unwrap_or_default(), - }) - .await?; if let ReceiveTarget::Directory(output_dir) = &target { tokio::fs::create_dir_all(output_dir).await?; } @@ -572,8 +746,23 @@ impl CoreInner { get_hash_seq_and_sizes(&connection, &hash_and_format.hash, 1024 * 1024 * 32, None) .await .context("failed to get file sizes")?; - let total_size = sizes.iter().copied().sum::(); + let total_size = sizes + .iter() + .try_fold(0u64, |total, size| total.checked_add(*size)) + .context("remote collection size overflow")?; let total_files = sizes.len().saturating_sub(1) as u64; + if total_files > self.limits.max_collection_files { + anyhow::bail!( + "remote collection has {total_files} files, limit is {}", + self.limits.max_collection_files + ); + } + if total_size > self.limits.max_total_bytes { + anyhow::bail!( + "remote collection size {total_size} exceeds limit {}", + self.limits.max_total_bytes + ); + } self.emit_transfer( transfer_id, "receive", @@ -604,42 +793,111 @@ impl CoreInner { self.export_collection(transfer_id, total_files, target, collection) .await?; self.repository - .update_transfer_status(transfer_id, STATUS_DONE) + .transition_transfer_status( + transfer_id, + TransferStatus::Receiving, + TransferStatus::Done, + ) .await?; self.emit_transfer(transfer_id, "receive", "lifecycle", "done", json!({})); Ok(()) } + async fn persist_receive_start( + &self, + transfer_id: u64, + parsed: &ParsedTransferTicket, + receiver_name: Option<&str>, + ) -> Result<()> { + let peer_id = parsed.blob_ticket.addr().id.to_string(); + self.repository + .start_receive(TransferUpsert { + transfer_id, + peer_id: Some(&peer_id), + direction: TransferDirection::Receive, + status: TransferStatus::Receiving, + transfer_name: parsed + .metadata + .as_ref() + .map(|metadata| metadata.transfer_name.as_str()), + content_hash: parsed + .metadata + .as_ref() + .map(|metadata| metadata.content_hash.as_str()), + ticket: None, + file_count: parsed + .metadata + .as_ref() + .map(|metadata| metadata.file_count) + .unwrap_or_default(), + total_size: parsed + .metadata + .as_ref() + .map(|metadata| metadata.total_size) + .unwrap_or_default(), + access_mode: mode_to_storage(&TransferAccessMode::ApprovalRequired), + }) + .await?; + let metadata_json = + serde_json::to_value(&parsed.metadata).unwrap_or(serde_json::Value::Null); + self.emit_transfer( + transfer_id, + "receive", + "lifecycle", + "started", + json!({ + "metadata": metadata_json, + "receiver_name": receiver_name, + }), + ); + Ok(()) + } + async fn cancel_transfer(&self, transfer_id: u64) -> Result<()> { - if let Some(tx) = self.active_transfers.lock().await.remove(&transfer_id) { - let _ = tx.send(()); + let mut active_transfers = self.active_transfers.lock().await; + if let Some(direction) = active_transfers + .get(&transfer_id) + .map(|active| active.direction) + { + let expected = match direction { + TransferDirection::Send => TransferStatus::Importing, + TransferDirection::Receive => TransferStatus::Receiving, + }; + self.repository + .transition_transfer_status(transfer_id, expected, TransferStatus::Cancelled) + .await?; + let active = active_transfers + .remove(&transfer_id) + .ok_or_else(|| anyhow::anyhow!("active transfer registration disappeared"))?; + drop(active_transfers); + let _ = active.cancel.send(()); self.emit_transfer( transfer_id, - "app", + direction.as_str(), "lifecycle", "cancel-requested", json!({}), ); - self.repository - .update_transfer_status(transfer_id, STATUS_CANCELLED) - .await?; return Ok(()); } - if self - .active_shares - .lock() - .await - .remove(&transfer_id) - .is_some() - { + drop(active_transfers); + + let mut active_shares = self.active_shares.lock().await; + if active_shares.contains_key(&transfer_id) { + self.repository + .transition_transfer_status( + transfer_id, + TransferStatus::Sharing, + TransferStatus::Stopped, + ) + .await?; + active_shares.remove(&transfer_id); + drop(active_shares); self.hash_to_transfer .lock() .await .retain(|_, id| *id != transfer_id); self.access_policy.remove_transfer(transfer_id).await; - self.repository - .update_transfer_status(transfer_id, STATUS_STOPPED) - .await?; self.emit_transfer(transfer_id, "send", "lifecycle", "share-stopped", json!({})); return Ok(()); } @@ -651,6 +909,9 @@ impl CoreInner { transfer_id: u64, mode: TransferAccessMode, ) -> Result<()> { + self.repository + .update_active_share_access_mode(transfer_id, mode_to_storage(&mode)) + .await?; self.access_policy.set_mode(transfer_id, mode.clone()).await; self.emit_transfer( transfer_id, @@ -750,7 +1011,7 @@ impl CoreInner { transfer_id: u64, sources: Vec, ) -> Result { - let files = collect_import_files(sources)?; + let files = collect_import_files_with_limits(sources, &self.limits)?; let default_name = default_collection_name(&files); let parallelism = num_cpus::get().min(8); let mut names_and_tags = n0_future::stream::iter(files) @@ -777,7 +1038,16 @@ impl CoreInner { .collect::>>()?; names_and_tags.sort_by(|(a, _, _), (b, _, _)| a.cmp(b)); - let total_size = names_and_tags.iter().map(|(_, _, size)| *size).sum::(); + let total_size = names_and_tags + .iter() + .try_fold(0u64, |total, (_, _, size)| total.checked_add(*size)) + .context("collection size overflow")?; + if total_size > self.limits.max_total_bytes { + anyhow::bail!( + "collection size {total_size} exceeds transfer limit {}", + self.limits.max_total_bytes + ); + } let (collection, tags) = names_and_tags .into_iter() .map(|(name, tag, _)| ((name, tag.hash()), tag)) @@ -816,7 +1086,15 @@ impl CoreInner { anyhow::bail!("import stream ended without a tag"); }; match item { - AddProgressItem::Size(item_size) => size = item_size, + AddProgressItem::Size(item_size) => { + if item_size > self.limits.max_total_bytes { + anyhow::bail!( + "file size {item_size} exceeds transfer limit {}", + self.limits.max_total_bytes + ); + } + size = item_size; + } AddProgressItem::CopyProgress(offset) => { self.emit_transfer( transfer_id, @@ -895,13 +1173,14 @@ impl CoreInner { relative_path: &str, hash: Hash, ) -> Result<()> { - let target = safe_output_path(output_dir, relative_path)?; - if let Some(parent) = target.parent() { - tokio::fs::create_dir_all(parent).await?; + if relative_path.len() as u64 > self.limits.max_path_bytes { + anyhow::bail!( + "output path exceeds {} bytes: {relative_path}", + self.limits.max_path_bytes + ); } + let (pending_file, writer) = AtomicOutputFile::create(output_dir, relative_path)?; let (tx, rx) = async_channel::bounded::>>(2); - let writer = File::create(&target) - .with_context(|| format!("failed to create {}", target.display()))?; let writer_task = std::thread::spawn(move || write_stream_to_blocking_writer(writer, rx)); let mut stream = self.store.export_ranges(hash, 0..u64::MAX).stream(); let mut file_size = 0; @@ -943,7 +1222,9 @@ impl CoreInner { tx.send(Ok(None)) .await .map_err(|_| anyhow::anyhow!("export writer closed for {relative_path}"))?; - wait_for_writer(writer_task).await??; + let writer = wait_for_writer(writer_task).await??; + tokio::task::spawn_blocking(move || writer.sync_all()).await??; + pending_file.commit()?; Ok(()) } @@ -956,10 +1237,14 @@ impl CoreInner { relative_path: &str, hash: Hash, ) -> Result<()> { + if relative_path.len() as u64 > self.limits.max_path_bytes { + anyhow::bail!( + "output path exceeds {} bytes: {relative_path}", + self.limits.max_path_bytes + ); + } let relative_path = validated_relative_string(relative_path)?; - output_sink - .start_file(relative_path.clone()) - .map_err(|error| anyhow::anyhow!(error.to_string()))?; + let output_file = OutputSinkFile::start(output_sink, relative_path.clone())?; let mut stream = self.store.export_ranges(hash, 0..u64::MAX).stream(); let mut file_size = 0; let mut exported = 0; @@ -974,9 +1259,7 @@ impl CoreInner { ); } exported += leaf.data.len() as u64; - output_sink - .write_chunk(relative_path.clone(), leaf.data.to_vec()) - .map_err(|error| anyhow::anyhow!(error.to_string()))?; + output_file.write(leaf.data.to_vec())?; self.emit_transfer( transfer_id, "receive", @@ -997,9 +1280,7 @@ impl CoreInner { } } - output_sink - .finish_file(relative_path) - .map_err(|error| anyhow::anyhow!(error.to_string()))?; + output_file.finish()?; Ok(()) } @@ -1296,6 +1577,8 @@ impl CoreInner { async fn list_events(&self, transfer_id: Option) -> Result> { self.event_hub.flush().await; - self.repository.list_events(transfer_id).await + self.repository + .list_events(transfer_id, self.limits.max_events) + .await } } diff --git a/crates/vnidrop/src/secret.rs b/crates/vnidrop/src/secret.rs index 546b051..3839c60 100644 --- a/crates/vnidrop/src/secret.rs +++ b/crates/vnidrop/src/secret.rs @@ -1,5 +1,8 @@ use std::{io, path::Path, str::FromStr}; +#[cfg(unix)] +use std::os::unix::fs::PermissionsExt; + use anyhow::{Context, Result}; use data_encoding::HEXLOWER; use iroh::SecretKey; @@ -18,13 +21,21 @@ pub(crate) async fn load_or_create_secret(app_data_dir: &Path) -> Result { let secret = SecretKey::generate(); tokio::fs::write(&path, HEXLOWER.encode(&secret.to_bytes())).await?; + restrict_permissions(&path).await?; Ok(secret) } Err(error) => Err(error.into()), } } + +async fn restrict_permissions(path: &Path) -> Result<()> { + #[cfg(unix)] + tokio::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600)).await?; + Ok(()) +} diff --git a/crates/vnidrop/src/ticket.rs b/crates/vnidrop/src/ticket.rs index 665f744..76f9198 100644 --- a/crates/vnidrop/src/ticket.rs +++ b/crates/vnidrop/src/ticket.rs @@ -5,7 +5,7 @@ use data_encoding::BASE64URL_NOPAD; use iroh_blobs::ticket::BlobTicket; use serde::{Deserialize, Serialize}; -use crate::api::TransferMetadata; +use crate::api::{CoreLimits, TransferMetadata}; const VNIDROP_TICKET_PREFIX: &str = "vnd1:"; const VNIDROP_TICKET_VERSION: u8 = 1; @@ -52,7 +52,22 @@ pub(crate) struct ParsedTransferTicket { pub(crate) metadata: Option, } +#[cfg(test)] pub(crate) fn parse_transfer_ticket(value: &str) -> Result { + parse_transfer_ticket_with_limits(value, &CoreLimits::default()) +} + +pub(crate) fn parse_transfer_ticket_with_limits( + value: &str, + limits: &CoreLimits, +) -> Result { + if value.len() as u64 > limits.max_ticket_bytes { + anyhow::bail!( + "ticket is {} bytes, limit is {}", + value.len(), + limits.max_ticket_bytes + ); + } let normalized = normalize_ticket_input(value); if normalized.starts_with(VNIDROP_TICKET_PREFIX) { let ticket = VnidropTicket::decode(&normalized)?; @@ -71,6 +86,11 @@ pub(crate) fn parse_transfer_ticket(value: &str) -> Result if ticket.metadata.transfer_name.trim().is_empty() { anyhow::bail!("VniDrop ticket metadata is missing a transfer name"); } + limits.validate_metadata_text( + "transfer name", + Some(ticket.metadata.transfer_name.as_str()), + )?; + limits.validate_metadata_text("sender name", ticket.metadata.sender_name.as_deref())?; 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() { diff --git a/crates/vnidrop/src/transfer_state.rs b/crates/vnidrop/src/transfer_state.rs new file mode 100644 index 0000000..e25f211 --- /dev/null +++ b/crates/vnidrop/src/transfer_state.rs @@ -0,0 +1,114 @@ +use anyhow::{bail, Result}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum TransferDirection { + Send, + Receive, +} + +impl TransferDirection { + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::Send => "send", + Self::Receive => "receive", + } + } +} + +impl TryFrom<&str> for TransferDirection { + type Error = anyhow::Error; + + fn try_from(value: &str) -> Result { + match value { + "send" => Ok(Self::Send), + "receive" => Ok(Self::Receive), + _ => bail!("unknown transfer direction: {value}"), + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum TransferStatus { + Importing, + Sharing, + Receiving, + Done, + Failed, + Cancelled, + Stopped, +} + +impl TransferStatus { + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::Importing => "importing", + Self::Sharing => "sharing", + Self::Receiving => "receiving", + Self::Done => "done", + Self::Failed => "failed", + Self::Cancelled => "cancelled", + Self::Stopped => "stopped", + } + } + + pub(crate) const fn can_transition_to(self, next: Self) -> bool { + matches!( + (self, next), + ( + Self::Importing, + Self::Sharing | Self::Failed | Self::Cancelled + ) | (Self::Sharing, Self::Stopped | Self::Failed) + | (Self::Receiving, Self::Done | Self::Failed | Self::Cancelled) + ) + } +} + +impl TryFrom<&str> for TransferStatus { + type Error = anyhow::Error; + + fn try_from(value: &str) -> Result { + match value { + "importing" => Ok(Self::Importing), + "sharing" => Ok(Self::Sharing), + "receiving" => Ok(Self::Receiving), + "done" => Ok(Self::Done), + "failed" => Ok(Self::Failed), + "cancelled" => Ok(Self::Cancelled), + "stopped" => Ok(Self::Stopped), + _ => bail!("unknown transfer status: {value}"), + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum ReceiverRequestStatus { + Requested, + Accepted, + Refused, + Expired, +} + +impl ReceiverRequestStatus { + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::Requested => "requested", + Self::Accepted => "accepted", + Self::Refused => "refused", + Self::Expired => "expired", + } + } +} + +impl TryFrom<&str> for ReceiverRequestStatus { + type Error = anyhow::Error; + + fn try_from(value: &str) -> Result { + match value { + "requested" => Ok(Self::Requested), + "accepted" => Ok(Self::Accepted), + "refused" => Ok(Self::Refused), + "expired" => Ok(Self::Expired), + _ => bail!("unknown receiver request status: {value}"), + } + } +}