diff --git a/.github/workflows/rust-core.yml b/.github/workflows/rust-core.yml
new file mode 100644
index 0000000..192b88b
--- /dev/null
+++ b/.github/workflows/rust-core.yml
@@ -0,0 +1,37 @@
+name: Rust core
+
+on:
+ pull_request:
+ paths:
+ - "Cargo.toml"
+ - "Cargo.lock"
+ - "crates/vnidrop/**"
+ - ".github/workflows/rust-core.yml"
+ push:
+ paths:
+ - "Cargo.toml"
+ - "Cargo.lock"
+ - "crates/vnidrop/**"
+ - ".github/workflows/rust-core.yml"
+
+permissions:
+ contents: read
+
+jobs:
+ quality:
+ runs-on: ubuntu-latest
+ timeout-minutes: 30
+ steps:
+ - uses: actions/checkout@v4
+ - name: Install Rust quality components
+ run: rustup component add clippy rustfmt
+ - name: Check formatting
+ run: cargo fmt --all -- --check
+ - name: Run strict Clippy
+ run: cargo clippy --workspace --all-targets -- -D warnings
+ - name: Run unit and integration tests
+ run: cargo test --workspace --all-targets
+ - name: Check documentation
+ env:
+ RUSTDOCFLAGS: -D warnings
+ run: cargo doc --workspace --no-deps
diff --git a/Cargo.lock b/Cargo.lock
index 158af04..1261b9b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -5056,6 +5056,7 @@ version = "0.1.0"
dependencies = [
"anyhow",
"async-channel",
+ "blake3",
"bytes",
"data-encoding",
"futures",
diff --git a/androidApp/build.gradle.kts b/androidApp/build.gradle.kts
index 3866d0e..0930536 100644
--- a/androidApp/build.gradle.kts
+++ b/androidApp/build.gradle.kts
@@ -58,6 +58,6 @@ android {
tasks.configureEach {
if (name == "mergeDebugJniLibFolders" || name == "mergeDebugNativeLibs") {
- dependsOn(":shared:cargoBuildAndroidArm64Debug")
+ dependsOn(":shared:copyAndroidAndroidArm64Debug")
}
}
diff --git a/androidApp/src/main/AndroidManifest.xml b/androidApp/src/main/AndroidManifest.xml
index 542f6f8..2f9818c 100644
--- a/androidApp/src/main/AndroidManifest.xml
+++ b/androidApp/src/main/AndroidManifest.xml
@@ -3,6 +3,9 @@
+
+
+
+
+
+
diff --git a/androidApp/src/main/kotlin/com/vnidrop/app/MainActivity.kt b/androidApp/src/main/kotlin/com/vnidrop/app/MainActivity.kt
index c1bef0a..b8513e0 100644
--- a/androidApp/src/main/kotlin/com/vnidrop/app/MainActivity.kt
+++ b/androidApp/src/main/kotlin/com/vnidrop/app/MainActivity.kt
@@ -4,25 +4,13 @@ import android.os.Bundle
import androidx.activity.ComponentActivity
import androidx.activity.compose.setContent
import androidx.activity.enableEdgeToEdge
-import androidx.compose.runtime.Composable
-import androidx.compose.ui.tooling.preview.Preview
-import com.vnidrop.app.core.attachAndroidFilePickerContext
class MainActivity : ComponentActivity() {
override fun onCreate(savedInstanceState: Bundle?) {
enableEdgeToEdge()
super.onCreate(savedInstanceState)
- attachAndroidFilePickerContext(this)
- attachAndroidPlatformContext(this)
-
setContent {
- App()
+ App(rememberAndroidAppDependencies(this))
}
}
}
-
-@Preview
-@Composable
-fun AppAndroidPreview() {
- App()
-}
diff --git a/androidApp/src/main/res/xml/file_paths.xml b/androidApp/src/main/res/xml/file_paths.xml
new file mode 100644
index 0000000..db732ae
--- /dev/null
+++ b/androidApp/src/main/res/xml/file_paths.xml
@@ -0,0 +1,4 @@
+
+
+
+
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/Cargo.toml b/crates/vnidrop/Cargo.toml
index 9925bab..7c1e77c 100644
--- a/crates/vnidrop/Cargo.toml
+++ b/crates/vnidrop/Cargo.toml
@@ -11,6 +11,7 @@ crate-type = ["cdylib", "staticlib", "rlib"]
anyhow = "1.0.102"
async-channel = "2.5.0"
bytes = "1.11.1"
+blake3 = "1.8.3"
data-encoding = "2.11.0"
futures = "0.3"
futures-lite = "2.6.1"
diff --git a/crates/vnidrop/src/access_policy.rs b/crates/vnidrop/src/access_policy.rs
index 6a7cc56..77ae471 100644
--- a/crates/vnidrop/src/access_policy.rs
+++ b/crates/vnidrop/src/access_policy.rs
@@ -26,6 +26,13 @@ impl AccessPolicy {
self.modes.write().await.insert(transfer_id, mode);
}
+ pub(crate) async fn allows_without_approval(&self, transfer_id: u64) -> bool {
+ matches!(
+ self.modes.read().await.get(&transfer_id),
+ Some(TransferAccessMode::Public)
+ )
+ }
+
pub(crate) async fn remove_transfer(&self, transfer_id: u64) {
self.modes.write().await.remove(&transfer_id);
self.approved_sessions
@@ -90,6 +97,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 9f12cb9..f7ad1b7 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,
@@ -20,6 +108,22 @@ pub trait CoreEventSink: Send + Sync {
fn on_event(&self, event: CoreEvent);
}
+#[uniffi::export(with_foreign)]
+pub trait ReceiveOutputSink: Send + Sync {
+ fn start_file(&self, relative_path: String) -> Result<(), crate::error::VnidropError>;
+ fn write_chunk(
+ &self,
+ relative_path: String,
+ 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)]
pub struct RuntimeStatus {
pub endpoint_id: String,
@@ -49,9 +153,10 @@ pub struct ShareMetadataInput {
pub transfer_id: u64,
pub transfer_name: Option,
pub sender_name: Option,
+ pub access_mode: TransferAccessMode,
}
-#[derive(Debug, Clone, Serialize, Deserialize, uniffi::Enum)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, uniffi::Enum)]
pub enum TransferAccessMode {
Public,
ApprovalRequired,
@@ -59,7 +164,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,
@@ -67,6 +174,7 @@ pub struct StoredTransfer {
pub ticket: Option,
pub file_count: u64,
pub total_size: u64,
+ pub access_mode: TransferAccessMode,
pub created_at: i64,
pub updated_at: i64,
}
@@ -136,4 +244,5 @@ pub struct ReceiverRequest {
pub reason: Option,
pub requested_at: i64,
pub responded_at: Option,
+ pub completed_at: Option,
}
diff --git a/crates/vnidrop/src/approval.rs b/crates/vnidrop/src/approval.rs
index d63efac..b8ab720 100644
--- a/crates/vnidrop/src/approval.rs
+++ b/crates/vnidrop/src/approval.rs
@@ -8,8 +8,9 @@ use uuid::Uuid;
use crate::{
access_policy::AccessPolicy,
event_hub::EventHub,
- handshake::{HandshakeResponse, RequestTransfer},
+ handshake::{DeliveryReceipt, DeliveryReceiptResponse, HandshakeResponse, RequestTransfer},
repository::{ReceiverRequestInsert, Repository},
+ transfer_state::ReceiverRequestStatus,
util::now_ms,
};
@@ -29,19 +30,60 @@ pub(crate) struct ApprovalService {
event_hub: Arc,
access_policy: Arc,
pending: Arc>>>,
+ max_pending: usize,
+ max_metadata_bytes: u64,
}
impl ApprovalService {
+ pub(crate) async fn complete_delivery(
+ &self,
+ remote_endpoint_id: String,
+ receipt: DeliveryReceipt,
+ ) -> DeliveryReceiptResponse {
+ let token_hash = receipt_token_hash(&receipt.token);
+ match self
+ .repository
+ .complete_receiver_delivery(
+ &receipt.request_id,
+ receipt.transfer_id,
+ &remote_endpoint_id,
+ &token_hash,
+ )
+ .await
+ {
+ Ok(()) => {
+ self.event_hub.emit_transfer(
+ receipt.transfer_id,
+ "send",
+ "delivery",
+ "receiver-completed",
+ json!({ "request_id": receipt.request_id, "remote_endpoint_id": remote_endpoint_id }),
+ );
+ DeliveryReceiptResponse::Recorded
+ }
+ Err(error) => {
+ tracing::warn!(%error, "rejected receiver delivery receipt");
+ DeliveryReceiptResponse::Rejected {
+ reason: "invalid-receipt".to_string(),
+ }
+ }
+ }
+ }
+
pub(crate) fn new(
repository: Repository,
event_hub: Arc,
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 +94,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 +119,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",
@@ -90,8 +155,17 @@ impl ApprovalService {
.await
{
Ok(true) => {
- self.wait_for_sender_decision(remote_endpoint_id, request)
+ if self
+ .access_policy
+ .allows_without_approval(request.transfer_id)
.await
+ {
+ self.allow_without_sender_decision(remote_endpoint_id, request)
+ .await
+ } else {
+ self.wait_for_sender_decision(remote_endpoint_id, request)
+ .await
+ }
}
Ok(false) => {
self.deny(request.transfer_id, remote_endpoint_id, "unknown-transfer")
@@ -105,6 +179,64 @@ impl ApprovalService {
}
}
+ async fn allow_without_sender_decision(
+ &self,
+ remote_endpoint_id: String,
+ request: RequestTransfer,
+ ) -> HandshakeResponse {
+ let request_id = Uuid::new_v4().to_string();
+ if self
+ .repository
+ .insert_receiver_request(ReceiverRequestInsert {
+ id: &request_id,
+ transfer_id: request.transfer_id,
+ remote_endpoint_id: &remote_endpoint_id,
+ transfer_name: &request.transfer_name,
+ receiver_name: request.receiver_name.as_deref(),
+ receiver_device_name: request.receiver_device_name.as_deref(),
+ app_version: &request.app_version,
+ })
+ .await
+ .is_err()
+ || self
+ .repository
+ .update_receiver_request_status(&request_id, ReceiverRequestStatus::Accepted, None)
+ .await
+ .is_err()
+ {
+ return self
+ .deny(request.transfer_id, remote_endpoint_id, "repository-error")
+ .await;
+ }
+ let token = Uuid::new_v4().to_string();
+ if self
+ .repository
+ .set_receiver_receipt_token(&request_id, &receipt_token_hash(&token))
+ .await
+ .is_err()
+ {
+ return self
+ .deny(request.transfer_id, remote_endpoint_id, "repository-error")
+ .await;
+ }
+ let expires_at = now_ms() + APPROVAL_TTL_MS;
+ self.event_hub.emit_transfer(
+ request.transfer_id,
+ "send",
+ "access",
+ "receiver-auto-approved",
+ json!({
+ "remote_endpoint_id": remote_endpoint_id,
+ "expires_at": expires_at,
+ }),
+ );
+ HandshakeResponse::Approved {
+ request_id,
+ token,
+ expires_at,
+ }
+ }
+
async fn wait_for_sender_decision(
&self,
remote_endpoint_id: String,
@@ -112,7 +244,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
@@ -171,7 +315,21 @@ impl ApprovalService {
"expires_at": expires_at,
}),
);
- HandshakeResponse::Approved { token, expires_at }
+ if let Err(error) = self
+ .repository
+ .set_receiver_receipt_token(&decision.request_id, &receipt_token_hash(&token))
+ .await
+ {
+ tracing::error!(%error, "failed to attach delivery receipt token");
+ return HandshakeResponse::Denied {
+ reason: "repository-error".to_string(),
+ };
+ }
+ HandshakeResponse::Approved {
+ request_id: decision.request_id,
+ token,
+ expires_at,
+ }
}
Ok(Ok(decision)) => {
self.deny(
@@ -189,7 +347,7 @@ impl ApprovalService {
.repository
.update_receiver_request_status(
&request_id,
- "expired",
+ ReceiverRequestStatus::Expired,
Some("approval timed out"),
)
.await;
@@ -219,3 +377,7 @@ impl ApprovalService {
HandshakeResponse::Denied { reason }
}
}
+
+fn receipt_token_hash(token: &str) -> String {
+ blake3::hash(token.as_bytes()).to_hex().to_string()
+}
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