From f0b06ad1cffd3c7c7d395e0862d612fa886a5a79 Mon Sep 17 00:00:00 2001 From: Hammed Abass Date: Wed, 22 Jul 2026 21:34:33 +0200 Subject: [PATCH] fix(transfer): finalize delivery completion --- apple/Tests/ProgressDerivationTests.swift | 22 +++ apple/VniDrop/Core/CoreRepository.swift | 2 +- apple/VniDrop/Core/TransferProgress.swift | 4 +- apple/VniDrop/Features/Send/SendScreen.swift | 11 +- crates/vnidrop/src/repository.rs | 110 +++++++++++++- crates/vnidrop/src/runtime/delivery.rs | 135 ++++++++++++++++++ crates/vnidrop/src/runtime/lifecycle.rs | 4 + crates/vnidrop/src/runtime/mod.rs | 8 +- crates/vnidrop/src/runtime/receive.rs | 43 ++---- crates/vnidrop/src/tests/repository.rs | 60 +++++++- crates/vnidrop/src/tests/runtime.rs | 59 +++++++- crates/vnidrop/tests/approval.rs | 33 +++-- .../com/vnidrop/app/core/CoreRepository.kt | 2 +- .../vnidrop/app/feature/send/SendCatalog.kt | 14 +- .../vnidrop/app/feature/send/SendScreen.kt | 1 + .../vnidrop/app/feature/send/SendViewModel.kt | 29 ++++ .../app/feature/settings/SettingsOverview.kt | 14 +- .../com/vnidrop/app/ui/state/AppUiModels.kt | 27 +--- .../com/vnidrop/app/feature/ViewModelsTest.kt | 51 +++++++ .../vnidrop/app/ui/state/AppUiModelsTest.kt | 36 ++++- 20 files changed, 580 insertions(+), 85 deletions(-) create mode 100644 crates/vnidrop/src/runtime/delivery.rs diff --git a/apple/Tests/ProgressDerivationTests.swift b/apple/Tests/ProgressDerivationTests.swift index 431eb1d..4a20940 100644 --- a/apple/Tests/ProgressDerivationTests.swift +++ b/apple/Tests/ProgressDerivationTests.swift @@ -44,6 +44,24 @@ final class ProgressDerivationTests: XCTestCase { XCTAssertEqual(progress?.progress, 0.3) } + func testReceiverCompletionAfterProgressIsTerminal() { + let events = [ + receiverEvent(kind: "completed", json: "{\"connection_id\":1,\"request_id\":1,\"endpoint_id\":\"peer-a\"}"), + receiverEvent(kind: "progress", json: "{\"connection_id\":1,\"request_id\":1,\"endpoint_id\":\"peer-a\",\"end_offset\":100}"), + receiverEvent(kind: "started", json: "{\"connection_id\":1,\"request_id\":1,\"endpoint_id\":\"peer-a\",\"size\":100}"), + ] + + let progress = progressForReceiver( + events: events, + transferId: 1, + remoteEndpointId: "peer-a", + totalSizeHint: 100 + ) + XCTAssertEqual(progress?.kind, "completed") + XCTAssertEqual(progress?.labelKey, "progress_completed") + XCTAssertEqual(progress?.progress, 1) + } + func testStatusLabelKeys() { XCTAssertEqual(statusLabelKey(.sharing), "status_available") XCTAssertEqual(statusLabelKey(.receiving), "status_receiving") @@ -56,4 +74,8 @@ final class ProgressDerivationTests: XCTestCase { direction: "send", phase: phase, kind: kind, dataJson: json ) } + + private func receiverEvent(kind: String, json: String) -> CoreEventModel { + event(phase: "transfer", kind: kind, json: json) + } } diff --git a/apple/VniDrop/Core/CoreRepository.swift b/apple/VniDrop/Core/CoreRepository.swift index 4275d1e..141e54d 100644 --- a/apple/VniDrop/Core/CoreRepository.swift +++ b/apple/VniDrop/Core/CoreRepository.swift @@ -197,7 +197,7 @@ final class CoreRepository: ObservableObject, CoreGateway { guard let transferId = model.transferId else { return } switch model.phase { - case "approval": signalsSubject.send(.approvalChanged(transferId: transferId)) + case "approval", "access": signalsSubject.send(.approvalChanged(transferId: transferId)) case "delivery": signalsSubject.send(.receiverHistoryChanged(transferId: transferId)) default: break } diff --git a/apple/VniDrop/Core/TransferProgress.swift b/apple/VniDrop/Core/TransferProgress.swift index a439335..6c5dabd 100644 --- a/apple/VniDrop/Core/TransferProgress.swift +++ b/apple/VniDrop/Core/TransferProgress.swift @@ -96,14 +96,14 @@ func progressForReceiver( labelKey: "progress_interrupted", progress: nil, detail: nil ) } - if latest.kind == "completed" && !transferEvents.contains(where: { $0.kind == "progress" || $0.kind == "started" }) { + let progress = aggregateReceiverProgress(events: transferEvents, totalSizeHint: totalSizeHint) + if latest.kind == "completed" && (progress.map { $0 >= 0.999 } ?? true) { return TransferProgress( transferId: transferId, phase: "transfer", kind: "completed", labelKey: "progress_completed", progress: 1, detail: nil ) } - let progress = aggregateReceiverProgress(events: transferEvents, totalSizeHint: totalSizeHint) return TransferProgress( transferId: transferId, phase: "transfer", kind: latest.kind, labelKey: "progress_sending", progress: progress, detail: progressDetail(latest) diff --git a/apple/VniDrop/Features/Send/SendScreen.swift b/apple/VniDrop/Features/Send/SendScreen.swift index 368c219..5b56ef7 100644 --- a/apple/VniDrop/Features/Send/SendScreen.swift +++ b/apple/VniDrop/Features/Send/SendScreen.swift @@ -127,11 +127,14 @@ struct SendScreen: View { private func sharingProgress(for transfer: Transfer) -> TransferProgress? { let active = (model.receiversByTransfer[transfer.transferId] ?? []).filter { $0.status == .accepted } if active.isEmpty { return nil } - let fractions = active.compactMap { - progressForReceiver(events: model.coreState.events, transferId: transfer.transferId, - remoteEndpointId: $0.remoteEndpointId, totalSizeHint: transfer.totalSize)?.progress + let fractions: [Double] = active.compactMap { receiver -> Double? in + let progress = progressForReceiver(events: model.coreState.events, transferId: transfer.transferId, + remoteEndpointId: receiver.remoteEndpointId, totalSizeHint: transfer.totalSize) + guard progress?.kind == "started" || progress?.kind == "progress" else { return nil } + return progress?.progress } - let combined = fractions.isEmpty ? nil : fractions.reduce(0, +) / Double(fractions.count) + guard !fractions.isEmpty else { return nil } + let combined = fractions.reduce(0, +) / Double(fractions.count) if active.count == 1 { return TransferProgress(transferId: transfer.transferId, phase: "transfer", kind: "progress", labelKey: "progress_sending", progress: combined) diff --git a/crates/vnidrop/src/repository.rs b/crates/vnidrop/src/repository.rs index 2bb71e2..9715789 100644 --- a/crates/vnidrop/src/repository.rs +++ b/crates/vnidrop/src/repository.rs @@ -20,7 +20,7 @@ use crate::{ util::now_ms, }; -const SCHEMA_VERSION: i64 = 5; +const SCHEMA_VERSION: i64 = 6; #[derive(Debug, Clone)] pub(crate) struct Repository { @@ -78,6 +78,23 @@ pub(crate) struct ReceiverRequestInsert<'a> { pub(crate) app_version: &'a str, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct PendingDeliveryReceipt { + pub(crate) local_transfer_id: u64, + pub(crate) sender_blob_ticket: String, + pub(crate) request_id: String, + pub(crate) sender_transfer_id: u64, + pub(crate) token: String, +} + +pub(crate) struct PendingDeliveryReceiptInsert<'a> { + pub(crate) local_transfer_id: u64, + pub(crate) sender_blob_ticket: &'a str, + pub(crate) request_id: &'a str, + pub(crate) sender_transfer_id: u64, + pub(crate) token: &'a str, +} + impl Repository { pub(crate) async fn open(app_data_dir: &Path) -> Result { let db_path = app_data_dir.join("vnidrop.sqlite3"); @@ -268,6 +285,21 @@ impl Repository { .execute(&self.pool) .await?; + sqlx::query( + r#" + CREATE TABLE IF NOT EXISTS pending_delivery_receipts ( + request_id TEXT PRIMARY KEY, + local_transfer_id INTEGER NOT NULL, + sender_blob_ticket TEXT NOT NULL, + sender_transfer_id INTEGER NOT NULL, + token TEXT NOT NULL, + created_at INTEGER NOT NULL + ); + "#, + ) + .execute(&self.pool) + .await?; + sqlx::query(&format!("PRAGMA user_version = {SCHEMA_VERSION}")) .execute(&self.pool) .await?; @@ -506,6 +538,82 @@ impl Repository { Ok(()) } + pub(crate) async fn complete_receive_with_pending_receipt( + &self, + receipt: PendingDeliveryReceiptInsert<'_>, + ) -> Result<()> { + self.maybe_fail_write()?; + let mut transaction = self.pool.begin().await?; + let updated = sqlx::query( + r#" + UPDATE transfers + SET status = 'done', updated_at = ?1 + WHERE transfer_id = ?2 AND direction = 'receive' AND status = 'receiving' + "#, + ) + .bind(now_ms()) + .bind(to_db_id(receipt.local_transfer_id)?) + .execute(&mut *transaction) + .await?; + require_one_changed(updated.rows_affected(), "complete receive")?; + sqlx::query( + r#" + INSERT INTO pending_delivery_receipts ( + request_id, local_transfer_id, sender_blob_ticket, + sender_transfer_id, token, created_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6) + ON CONFLICT(request_id) DO UPDATE SET + local_transfer_id = excluded.local_transfer_id, + sender_blob_ticket = excluded.sender_blob_ticket, + sender_transfer_id = excluded.sender_transfer_id, + token = excluded.token + "#, + ) + .bind(receipt.request_id) + .bind(to_db_id(receipt.local_transfer_id)?) + .bind(receipt.sender_blob_ticket) + .bind(to_db_id(receipt.sender_transfer_id)?) + .bind(receipt.token) + .bind(now_ms()) + .execute(&mut *transaction) + .await?; + transaction.commit().await?; + Ok(()) + } + + pub(crate) async fn list_pending_delivery_receipts( + &self, + ) -> Result> { + let rows = sqlx::query( + r#" + SELECT local_transfer_id, sender_blob_ticket, request_id, + sender_transfer_id, token + FROM pending_delivery_receipts + ORDER BY created_at ASC + "#, + ) + .fetch_all(&self.pool) + .await?; + Ok(rows + .into_iter() + .map(|row| PendingDeliveryReceipt { + local_transfer_id: row.get::("local_transfer_id") as u64, + sender_blob_ticket: row.get("sender_blob_ticket"), + request_id: row.get("request_id"), + sender_transfer_id: row.get::("sender_transfer_id") as u64, + token: row.get("token"), + }) + .collect()) + } + + pub(crate) async fn delete_pending_delivery_receipt(&self, request_id: &str) -> Result<()> { + sqlx::query("DELETE FROM pending_delivery_receipts WHERE request_id = ?1") + .bind(request_id) + .execute(&self.pool) + .await?; + Ok(()) + } + pub(crate) async fn update_active_share_access_mode( &self, transfer_id: u64, diff --git a/crates/vnidrop/src/runtime/delivery.rs b/crates/vnidrop/src/runtime/delivery.rs new file mode 100644 index 0000000..9a85c40 --- /dev/null +++ b/crates/vnidrop/src/runtime/delivery.rs @@ -0,0 +1,135 @@ +use std::{str::FromStr, sync::Arc, time::Duration}; + +use iroh_blobs::ticket::BlobTicket; +use serde_json::json; + +use super::CoreInner; +use crate::{ + handshake::{DeliveryReceipt, DeliveryReceiptResponse, HandshakeService}, + repository::PendingDeliveryReceipt, +}; + +const DELIVERY_RECEIPT_TIMEOUT: Duration = Duration::from_secs(5); +const DELIVERY_RECEIPT_RETRY_INTERVAL: Duration = Duration::from_secs(5); +const DELIVERY_RECEIPT_MAX_RETRY_INTERVAL: Duration = Duration::from_secs(5 * 60); + +impl CoreInner { + pub(super) async fn spawn_delivery_receipt_task(self: &Arc) { + let core = Arc::downgrade(self); + let task = tokio::spawn(async move { + let mut retry_interval = DELIVERY_RECEIPT_RETRY_INTERVAL; + loop { + let Some(core) = core.upgrade() else { + break; + }; + let has_pending = core.deliver_pending_receipts().await; + if has_pending { + let notified = tokio::select! { + () = core.delivery_receipt_notify.notified() => true, + () = tokio::time::sleep(retry_interval) => false, + }; + if notified { + retry_interval = DELIVERY_RECEIPT_RETRY_INTERVAL; + } else { + retry_interval = retry_interval + .saturating_mul(2) + .min(DELIVERY_RECEIPT_MAX_RETRY_INTERVAL); + } + } else { + core.delivery_receipt_notify.notified().await; + retry_interval = DELIVERY_RECEIPT_RETRY_INTERVAL; + } + } + }); + *self.delivery_receipt_task.lock().await = Some(task); + } + + async fn deliver_pending_receipts(&self) -> bool { + let receipts = match self.repository.list_pending_delivery_receipts().await { + Ok(receipts) => receipts, + Err(error) => { + tracing::warn!(%error, "failed to load pending delivery receipts"); + return true; + } + }; + let has_pending = !receipts.is_empty(); + for receipt in receipts { + self.deliver_pending_receipt(receipt).await; + } + has_pending + } + + async fn deliver_pending_receipt(&self, pending: PendingDeliveryReceipt) { + let blob_ticket = match BlobTicket::from_str(&pending.sender_blob_ticket) { + Ok(ticket) => ticket, + Err(error) => { + tracing::warn!(%error, request_id = %pending.request_id, "discarded invalid pending delivery receipt"); + let _ = self + .repository + .delete_pending_delivery_receipt(&pending.request_id) + .await; + self.emit_transfer( + pending.local_transfer_id, + "receive", + "delivery", + "receipt-rejected", + json!({ "reason": "invalid-sender-ticket" }), + ); + return; + } + }; + let client = HandshakeService::client(self.endpoint.clone(), blob_ticket.addr().clone()); + let receipt = DeliveryReceipt { + request_id: pending.request_id.clone(), + transfer_id: pending.sender_transfer_id, + token: pending.token, + }; + match tokio::time::timeout(DELIVERY_RECEIPT_TIMEOUT, client.report_delivery(receipt)).await + { + Ok(Ok(DeliveryReceiptResponse::Recorded)) => { + if let Err(error) = self + .repository + .delete_pending_delivery_receipt(&pending.request_id) + .await + { + tracing::warn!(%error, request_id = %pending.request_id, "failed to clear recorded delivery receipt"); + return; + } + self.emit_transfer( + pending.local_transfer_id, + "receive", + "delivery", + "receipt-recorded", + json!({ "sender_transfer_id": pending.sender_transfer_id }), + ); + } + Ok(Ok(DeliveryReceiptResponse::Rejected { reason })) => { + let _ = self + .repository + .delete_pending_delivery_receipt(&pending.request_id) + .await; + self.emit_transfer( + pending.local_transfer_id, + "receive", + "delivery", + "receipt-rejected", + json!({ "reason": reason }), + ); + } + Ok(Err(error)) => self.emit_transfer( + pending.local_transfer_id, + "receive", + "delivery", + "receipt-failed", + json!({ "reason": error.to_string() }), + ), + Err(_) => self.emit_transfer( + pending.local_transfer_id, + "receive", + "delivery", + "receipt-failed", + json!({ "reason": "delivery receipt timed out" }), + ), + } + } +} diff --git a/crates/vnidrop/src/runtime/lifecycle.rs b/crates/vnidrop/src/runtime/lifecycle.rs index 2d26bb8..cea2c9c 100644 --- a/crates/vnidrop/src/runtime/lifecycle.rs +++ b/crates/vnidrop/src/runtime/lifecycle.rs @@ -187,6 +187,10 @@ impl CoreInner { // 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 Some(task) = self.delivery_receipt_task.lock().await.take() { + task.abort(); + let _ = task.await; + } if let Err(error) = self.router.shutdown().await { self.emit_endpoint( "shutdown", diff --git a/crates/vnidrop/src/runtime/mod.rs b/crates/vnidrop/src/runtime/mod.rs index 5114634..55e3201 100644 --- a/crates/vnidrop/src/runtime/mod.rs +++ b/crates/vnidrop/src/runtime/mod.rs @@ -7,6 +7,7 @@ //! - [`lifecycle`] — cancel/delete/shutdown/status/access //! - [`provider`] — blob provider events and per-connection send progress +mod delivery; mod facade; mod lifecycle; mod provider; @@ -38,7 +39,7 @@ use iroh_blobs::{ }; use serde_json::json; use tokio::{ - sync::{oneshot, Mutex as TokioMutex, Semaphore}, + sync::{oneshot, Mutex as TokioMutex, Notify, Semaphore}, task::JoinHandle, }; @@ -77,6 +78,8 @@ pub(super) struct CoreInner { pub(super) hash_to_transfer: TokioMutex>>, pub(super) connection_endpoints: TokioMutex>, pub(super) provider_task: TokioMutex>>, + pub(super) delivery_receipt_notify: Notify, + pub(super) delivery_receipt_task: TokioMutex>>, pub(super) shutdown_started: AtomicBool, } @@ -244,6 +247,8 @@ impl CoreInner { hash_to_transfer: TokioMutex::new(restored_hashes), connection_endpoints: TokioMutex::new(HashMap::new()), provider_task: TokioMutex::new(None), + delivery_receipt_notify: Notify::new(), + delivery_receipt_task: TokioMutex::new(None), shutdown_started: AtomicBool::new(false), }); @@ -257,6 +262,7 @@ impl CoreInner { }), ); inner.spawn_provider_event_task(event_rx).await; + inner.spawn_delivery_receipt_task().await; Ok(inner) } diff --git a/crates/vnidrop/src/runtime/receive.rs b/crates/vnidrop/src/runtime/receive.rs index 6b7ddb9..b72223f 100644 --- a/crates/vnidrop/src/runtime/receive.rs +++ b/crates/vnidrop/src/runtime/receive.rs @@ -26,8 +26,8 @@ use crate::{ validated_relative_string, wait_for_writer, write_stream_to_blocking_writer, AtomicOutputFile, }, - handshake::{DeliveryReceipt, DeliveryReceiptResponse, HandshakeResponse, HandshakeService}, - repository::{ReceivedArtifactInsert, TransferUpsert}, + handshake::{DeliveryReceipt, HandshakeResponse, HandshakeService}, + repository::{PendingDeliveryReceiptInsert, ReceivedArtifactInsert, TransferUpsert}, ticket::{parse_transfer_ticket_with_limits, ParsedTransferTicket}, transfer_state::{TransferDirection, TransferStatus}, }; @@ -271,6 +271,7 @@ impl CoreInner { .map_err(VnidropError::filesystem)?; } let sender_addr = parsed.blob_ticket.addr().clone(); + let sender_blob_ticket = parsed.blob_ticket.to_string(); self.emit_transfer(transfer_id, "receive", "network", "connecting", json!({})); // Every VniDrop ticket carries metadata and must complete the handshake. @@ -352,40 +353,18 @@ impl CoreInner { self.export_collection(transfer_id, total_files, target, collection) .await?; self.repository - .transition_transfer_status( - transfer_id, - TransferStatus::Receiving, - TransferStatus::Done, - ) + .complete_receive_with_pending_receipt(PendingDeliveryReceiptInsert { + local_transfer_id: transfer_id, + sender_blob_ticket: &sender_blob_ticket, + request_id: &delivery_receipt.request_id, + sender_transfer_id: delivery_receipt.transfer_id, + token: &delivery_receipt.token, + }) .await .map_err(VnidropError::repository)?; drop(download_tag); self.emit_transfer(transfer_id, "receive", "lifecycle", "done", json!({})); - let sender_transfer_id = delivery_receipt.transfer_id; - let client = HandshakeService::client(self.endpoint.clone(), sender_addr); - match client.report_delivery(delivery_receipt).await { - Ok(DeliveryReceiptResponse::Recorded) => self.emit_transfer( - transfer_id, - "receive", - "delivery", - "receipt-recorded", - json!({ "sender_transfer_id": sender_transfer_id }), - ), - Ok(DeliveryReceiptResponse::Rejected { reason }) => self.emit_transfer( - transfer_id, - "receive", - "delivery", - "receipt-rejected", - json!({ "reason": reason }), - ), - Err(error) => self.emit_transfer( - transfer_id, - "receive", - "delivery", - "receipt-failed", - json!({ "reason": error.to_string() }), - ), - } + self.delivery_receipt_notify.notify_one(); Ok(()) } diff --git a/crates/vnidrop/src/tests/repository.rs b/crates/vnidrop/src/tests/repository.rs index 3edb411..8bade55 100644 --- a/crates/vnidrop/src/tests/repository.rs +++ b/crates/vnidrop/src/tests/repository.rs @@ -1,6 +1,9 @@ use crate::{ api::{CoreEvent, ReceivedLocatorKind}, - repository::{ReceivedArtifactInsert, ReceiverRequestInsert, Repository, TransferUpsert}, + repository::{ + PendingDeliveryReceiptInsert, ReceivedArtifactInsert, ReceiverRequestInsert, Repository, + TransferUpsert, + }, transfer_state::{ReceiverRequestStatus, TransferDirection, TransferStatus}, }; @@ -63,7 +66,7 @@ async fn received_artifacts_survive_history_deletion() { async fn persists_transfers_and_events_across_reopen() { let temp = tempfile::tempdir().unwrap(); let repository = Repository::open(temp.path()).await.unwrap(); - assert_eq!(repository.schema_version().await.unwrap(), 5); + assert_eq!(repository.schema_version().await.unwrap(), 6); repository .insert_transfer(transfer( 7, @@ -112,6 +115,57 @@ async fn persists_transfers_and_events_across_reopen() { assert_eq!(events[0].id, "event-1"); } +#[tokio::test] +async fn receive_completion_persists_delivery_receipt_until_recorded() { + let temp = tempfile::tempdir().unwrap(); + let repository = Repository::open(temp.path()).await.unwrap(); + repository + .start_receive(transfer( + 93, + TransferDirection::Receive, + TransferStatus::Receiving, + )) + .await + .unwrap(); + repository + .complete_receive_with_pending_receipt(PendingDeliveryReceiptInsert { + local_transfer_id: 93, + sender_blob_ticket: "blob-ticket", + request_id: "request-93", + sender_transfer_id: 39, + token: "receipt-token", + }) + .await + .unwrap(); + + let transfer = repository + .list_transfers() + .await + .unwrap() + .into_iter() + .find(|transfer| transfer.transfer_id == 93) + .unwrap(); + assert_eq!(transfer.status, "done"); + drop(repository); + + let reopened = Repository::open(temp.path()).await.unwrap(); + let pending = reopened.list_pending_delivery_receipts().await.unwrap(); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].local_transfer_id, 93); + assert_eq!(pending[0].sender_transfer_id, 39); + assert_eq!(pending[0].request_id, "request-93"); + assert_eq!(pending[0].token, "receipt-token"); + reopened + .delete_pending_delivery_receipt("request-93") + .await + .unwrap(); + assert!(reopened + .list_pending_delivery_receipts() + .await + .unwrap() + .is_empty()); +} + #[tokio::test] async fn receiver_request_can_only_be_resolved_once() { let temp = tempfile::tempdir().unwrap(); @@ -527,7 +581,7 @@ async fn migrates_schema_v2_identity_without_losing_transfer() { pool.close().await; let repository = Repository::open(temp.path()).await.unwrap(); - assert_eq!(repository.schema_version().await.unwrap(), 5); + assert_eq!(repository.schema_version().await.unwrap(), 6); let stored = repository.list_transfers().await.unwrap().remove(0); assert_eq!(stored.transfer_id, 7); assert_eq!(stored.local_id, "legacy-7-send"); diff --git a/crates/vnidrop/src/tests/runtime.rs b/crates/vnidrop/src/tests/runtime.rs index 6591a97..8b3ca76 100644 --- a/crates/vnidrop/src/tests/runtime.rs +++ b/crates/vnidrop/src/tests/runtime.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use iroh_blobs::Hash; use crate::{ - repository::{Repository, TransferUpsert}, + repository::{PendingDeliveryReceiptInsert, Repository, TransferUpsert}, transfer_state::{TransferDirection, TransferStatus}, CoreEvent, CoreEventSink, VnidropCore, VnidropError, }; @@ -98,6 +98,63 @@ fn startup_recovers_interrupted_transfer_and_persists_event() { core.shutdown(); } +#[test] +fn startup_processes_persisted_delivery_receipts() { + let temp = tempfile::tempdir().unwrap(); + let preparation_runtime = tokio::runtime::Runtime::new().unwrap(); + preparation_runtime.block_on(async { + let repository = Repository::open(temp.path()).await.unwrap(); + repository + .start_receive(TransferUpsert { + transfer_id: 94, + peer_id: None, + direction: TransferDirection::Receive, + status: TransferStatus::Receiving, + transfer_name: Some("completed receive"), + content_hash: Some("hash"), + ticket: None, + file_count: 1, + total_size: 5, + access_mode: "approval_required", + }) + .await + .unwrap(); + repository + .complete_receive_with_pending_receipt(PendingDeliveryReceiptInsert { + local_transfer_id: 94, + sender_blob_ticket: "invalid-ticket", + request_id: "request-94", + sender_transfer_id: 49, + token: "receipt-token", + }) + .await + .unwrap(); + }); + drop(preparation_runtime); + + let core = VnidropCore::initialize( + temp.path().to_string_lossy().to_string(), + Arc::new(TestSink), + ) + .unwrap(); + let started = std::time::Instant::now(); + loop { + if core.list_events(Some(94)).unwrap().iter().any(|event| { + event.phase == "delivery" + && event.kind == "receipt-rejected" + && event.data_json.contains("invalid-sender-ticket") + }) { + break; + } + assert!( + started.elapsed() < std::time::Duration::from_secs(2), + "startup did not process the persisted delivery receipt" + ); + std::thread::sleep(std::time::Duration::from_millis(10)); + } + core.shutdown(); +} + #[test] fn startup_fails_persisted_share_when_root_blob_is_missing() { let temp = tempfile::tempdir().unwrap(); diff --git a/crates/vnidrop/tests/approval.rs b/crates/vnidrop/tests/approval.rs index e20a090..a09253b 100644 --- a/crates/vnidrop/tests/approval.rs +++ b/crates/vnidrop/tests/approval.rs @@ -1,6 +1,9 @@ mod support; -use std::sync::Arc; +use std::{ + sync::Arc, + time::{Duration, Instant}, +}; use support::{ receive_with_response, share_path, wait_for_receiver_request, CoreGuard, RecordingSink, @@ -8,6 +11,24 @@ use support::{ }; use vnidrop::{CoreLimits, ShareMetadataInput, ShareSource, SourceKind, TransferAccessMode}; +fn wait_for_completed_delivery( + sender: &vnidrop::VnidropCore, + transfer_id: u64, +) -> Vec { + let started = Instant::now(); + loop { + let requests = sender.list_receiver_requests(transfer_id).unwrap(); + if requests.iter().any(|request| request.status == "completed") { + return requests; + } + assert!( + started.elapsed() < Duration::from_secs(5), + "delivery receipt was not recorded" + ); + std::thread::sleep(Duration::from_millis(10)); + } +} + #[test] fn public_share_receives_without_sender_approval() { let source_dir = tempfile::tempdir().unwrap(); @@ -47,10 +68,7 @@ fn public_share_receives_without_sender_approval() { std::fs::read(output_dir.path().join("public.txt")).unwrap(), b"public content" ); - let deliveries = sender - .core - .list_receiver_requests(share.transfer_id) - .unwrap(); + let deliveries = wait_for_completed_delivery(&sender.core, share.transfer_id); assert_eq!(deliveries.len(), 1); assert_eq!(deliveries[0].receiver_name.as_deref(), Some("Receiver")); assert_eq!(deliveries[0].status, "completed"); @@ -104,10 +122,7 @@ fn approval_required_denies_then_allows_receiver() { std::fs::read(allowed_output.path().join("private.txt")).unwrap(), b"approved content" ); - let completed = sender - .core - .list_receiver_requests(share.transfer_id) - .unwrap(); + let completed = wait_for_completed_delivery(&sender.core, share.transfer_id); assert!(completed .iter() .any(|request| request.status == "completed")); diff --git a/shared/src/commonMain/kotlin/com/vnidrop/app/core/CoreRepository.kt b/shared/src/commonMain/kotlin/com/vnidrop/app/core/CoreRepository.kt index dc8b05a..e1cc450 100644 --- a/shared/src/commonMain/kotlin/com/vnidrop/app/core/CoreRepository.kt +++ b/shared/src/commonMain/kotlin/com/vnidrop/app/core/CoreRepository.kt @@ -50,7 +50,7 @@ class CoreRepository( val transferId = model.transferId if (transferId != null) { when (model.phase) { - "approval" -> _signals.tryEmit(CoreSignal.ApprovalChanged(transferId)) + "approval", "access" -> _signals.tryEmit(CoreSignal.ApprovalChanged(transferId)) "delivery" -> _signals.tryEmit(CoreSignal.ReceiverHistoryChanged(transferId)) } if (model.shouldRefreshTransfers()) { diff --git a/shared/src/commonMain/kotlin/com/vnidrop/app/feature/send/SendCatalog.kt b/shared/src/commonMain/kotlin/com/vnidrop/app/feature/send/SendCatalog.kt index 8a153a1..278b9b9 100644 --- a/shared/src/commonMain/kotlin/com/vnidrop/app/feature/send/SendCatalog.kt +++ b/shared/src/commonMain/kotlin/com/vnidrop/app/feature/send/SendCatalog.kt @@ -35,6 +35,8 @@ import androidx.compose.ui.text.style.TextAlign import androidx.compose.ui.text.style.TextOverflow import androidx.compose.ui.unit.dp import com.vnidrop.app.core.CoreEventModel +import com.vnidrop.app.core.ReceiverDeliveryStatus +import com.vnidrop.app.core.ReceiverRequestModel import com.vnidrop.app.core.Transfer import com.vnidrop.app.core.TransferStatus import com.vnidrop.app.ui.components.PillTone @@ -79,6 +81,7 @@ internal fun SendFloatingAction(onClick: () -> Unit, modifier: Modifier = Modifi internal fun TransferCatalog( transfers: List, transferThumbnails: Map, + receiversByTransfer: Map> = emptyMap(), events: List = emptyList(), windowClass: WindowClass, onOpenComposer: () -> Unit, @@ -107,9 +110,18 @@ internal fun TransferCatalog( ) } items(transfers, key = Transfer::localId) { transfer -> + val activeReceiverEndpointIds = receiversByTransfer[transfer.transferId] + .orEmpty() + .filter { it.status == ReceiverDeliveryStatus.Accepted } + .mapTo(mutableSetOf()) { it.remoteEndpointId } val progress = when (transfer.status) { TransferStatus.Importing -> progressForTransfer(events, transfer.transferId) - TransferStatus.Sharing -> activeSendProgress(events, transfer.transferId, transfer.totalSize) + TransferStatus.Sharing -> activeSendProgress( + events, + transfer.transferId, + activeReceiverEndpointIds, + transfer.totalSize, + ) else -> null } TransferListItem( diff --git a/shared/src/commonMain/kotlin/com/vnidrop/app/feature/send/SendScreen.kt b/shared/src/commonMain/kotlin/com/vnidrop/app/feature/send/SendScreen.kt index 2cade4b..a49c2a7 100644 --- a/shared/src/commonMain/kotlin/com/vnidrop/app/feature/send/SendScreen.kt +++ b/shared/src/commonMain/kotlin/com/vnidrop/app/feature/send/SendScreen.kt @@ -69,6 +69,7 @@ fun SendScreen( TransferCatalog( transfers = outgoingTransfers, transferThumbnails = state.transferThumbnails, + receiversByTransfer = state.receiversByTransfer, events = coreState.events, windowClass = windowClass, onOpenComposer = onOpenComposer, diff --git a/shared/src/commonMain/kotlin/com/vnidrop/app/feature/send/SendViewModel.kt b/shared/src/commonMain/kotlin/com/vnidrop/app/feature/send/SendViewModel.kt index eaa1017..9a2db5f 100644 --- a/shared/src/commonMain/kotlin/com/vnidrop/app/feature/send/SendViewModel.kt +++ b/shared/src/commonMain/kotlin/com/vnidrop/app/feature/send/SendViewModel.kt @@ -8,6 +8,8 @@ import com.vnidrop.app.core.FileSystemService import com.vnidrop.app.core.PickedShareFile import com.vnidrop.app.core.ShareAccessPolicy import com.vnidrop.app.core.ReceiverRequestModel +import com.vnidrop.app.core.TransferDirection +import com.vnidrop.app.core.TransferStatus import com.vnidrop.app.preferences.PreferencesRepository import com.vnidrop.app.ui.feedback.UiMessage import com.vnidrop.app.ui.feedback.UiMessageController @@ -39,6 +41,7 @@ data class SendState( val transferThumbnails: Map = emptyMap(), val detailPanel: TransferDetailPanel? = null, val receiverHistory: List = emptyList(), + val receiversByTransfer: Map> = emptyMap(), val isLoadingReceivers: Boolean = false, val isDeleteConfirmationOpen: Boolean = false, val isDeleting: Boolean = false, @@ -82,15 +85,24 @@ class SendViewModel( if (signal.transferId == _state.value.selectedTransferId) { refreshReceivers(signal.transferId) } + refreshReceiverStatuses(signal.transferId) } is CoreSignal.ApprovalChanged -> { if (signal.transferId == _state.value.selectedTransferId) { refreshReceivers(signal.transferId) } + refreshReceiverStatuses(signal.transferId) } } } } + viewModelScope.launch { + coreState.map { core -> + core.transfers + .filter { it.direction == TransferDirection.Send && it.status in setOf(TransferStatus.Importing, TransferStatus.Sharing) } + .mapTo(mutableSetOf()) { it.transferId } + }.distinctUntilChanged().collect(::syncSharingReceivers) + } viewModelScope.launch { filePreviewRepository.previews.collect { previews -> _state.update { it.copy(transferThumbnails = previews) } @@ -316,4 +328,21 @@ class SendViewModel( ) } } + + private fun syncSharingReceivers(transferIds: Set) { + _state.update { current -> + current.copy(receiversByTransfer = current.receiversByTransfer.filterKeys { it in transferIds }) + } + transferIds.forEach(::refreshReceiverStatuses) + } + + private fun refreshReceiverStatuses(transferId: ULong) { + viewModelScope.launch { + repository.receiverRequests(transferId).onSuccess { requests -> + _state.update { current -> + current.copy(receiversByTransfer = current.receiversByTransfer + (transferId to requests)) + } + } + } + } } diff --git a/shared/src/commonMain/kotlin/com/vnidrop/app/feature/settings/SettingsOverview.kt b/shared/src/commonMain/kotlin/com/vnidrop/app/feature/settings/SettingsOverview.kt index 5d292c4..e5e262a 100644 --- a/shared/src/commonMain/kotlin/com/vnidrop/app/feature/settings/SettingsOverview.kt +++ b/shared/src/commonMain/kotlin/com/vnidrop/app/feature/settings/SettingsOverview.kt @@ -34,13 +34,6 @@ internal fun SettingsOverview( fontWeight = FontWeight.Bold, ) SettingsGroup { - SettingsRow( - icon = AppIcon.Storage, - title = stringResource(Res.string.storage_title), - selected = state.selectedSection == SettingsSection.Storage, - onClick = { onSectionSelected(SettingsSection.Storage) }, - ) - SettingsDivider() SettingsRow( icon = AppIcon.User, title = stringResource(Res.string.preferences_title), @@ -65,6 +58,13 @@ internal fun SettingsOverview( onClick = { onSectionSelected(SettingsSection.Notifications) }, ) SettingsDivider() + SettingsRow( + icon = AppIcon.Storage, + title = stringResource(Res.string.storage_title), + selected = state.selectedSection == SettingsSection.Storage, + onClick = { onSectionSelected(SettingsSection.Storage) }, + ) + SettingsDivider() SettingsRow( icon = AppIcon.Info, title = stringResource(Res.string.about_title), diff --git a/shared/src/commonMain/kotlin/com/vnidrop/app/ui/state/AppUiModels.kt b/shared/src/commonMain/kotlin/com/vnidrop/app/ui/state/AppUiModels.kt index 75288ba..ed2879d 100644 --- a/shared/src/commonMain/kotlin/com/vnidrop/app/ui/state/AppUiModels.kt +++ b/shared/src/commonMain/kotlin/com/vnidrop/app/ui/state/AppUiModels.kt @@ -138,7 +138,8 @@ fun progressForReceiver( detail = null, ) } - if (latest.kind == "completed" && transferEvents.none { it.kind == "progress" || it.kind == "started" }) { + val progress = aggregateReceiverProgress(transferEvents, totalSizeHint) + if (latest.kind == "completed" && (progress == null || progress >= 0.999f)) { return TransferProgress( transferId = transferId, phase = "transfer", @@ -149,7 +150,6 @@ fun progressForReceiver( ) } - val progress = aggregateReceiverProgress(transferEvents, totalSizeHint) return TransferProgress( transferId = transferId, phase = "transfer", @@ -167,36 +167,21 @@ fun progressForReceiver( fun activeSendProgress( events: List, transferId: ULong, + activeReceiverEndpointIds: Set, totalSizeHint: ULong? = null, ): TransferProgress? { + if (activeReceiverEndpointIds.isEmpty()) return null val endpointIds = events .asSequence() .filter { it.transferId == transferId && it.direction == "send" && it.phase == "transfer" } .mapNotNull { findString(it.dataJson, "endpoint_id") } + .filter { it in activeReceiverEndpointIds } .distinct() .toList() - if (endpointIds.isEmpty()) { - // Fall back to connection-scoped events without endpoint attribution. - val relevant = events.filter { - it.transferId == transferId && - it.direction == "send" && - it.phase == "transfer" && - it.kind in setOf("started", "progress") - } - if (relevant.isEmpty()) return null - return TransferProgress( - transferId = transferId, - phase = "transfer", - kind = relevant.first().kind, - label = Res.string.progress_sending, - progress = aggregateReceiverProgress(relevant, totalSizeHint), - detail = progressDetail(relevant.first()), - ) - } + if (endpointIds.isEmpty()) return null return endpointIds .mapNotNull { progressForReceiver(events, transferId, it, totalSizeHint) } .firstOrNull { it.kind == "progress" || it.kind == "started" } - ?: endpointIds.mapNotNull { progressForReceiver(events, transferId, it, totalSizeHint) }.firstOrNull() } fun summarizeProgress(events: List): List = diff --git a/shared/src/commonTest/kotlin/com/vnidrop/app/feature/ViewModelsTest.kt b/shared/src/commonTest/kotlin/com/vnidrop/app/feature/ViewModelsTest.kt index 8ff7f49..28a0e81 100644 --- a/shared/src/commonTest/kotlin/com/vnidrop/app/feature/ViewModelsTest.kt +++ b/shared/src/commonTest/kotlin/com/vnidrop/app/feature/ViewModelsTest.kt @@ -3,9 +3,12 @@ package com.vnidrop.app.feature import com.vnidrop.app.DeviceInfo import com.vnidrop.app.PlatformEnvironment import com.vnidrop.app.core.CoreState +import com.vnidrop.app.core.CoreSignal import com.vnidrop.app.core.PickedShareFile import com.vnidrop.app.core.ReceiveFolder import com.vnidrop.app.core.ReceiveFolderKind +import com.vnidrop.app.core.ReceiverDeliveryStatus +import com.vnidrop.app.core.ReceiverRequestModel import com.vnidrop.app.core.Share import com.vnidrop.app.core.ShareAccessPolicy import com.vnidrop.app.core.Transfer @@ -251,6 +254,38 @@ class ViewModelsTest { assertEquals(listOf(selected), fileSystem.discardedPickedFiles) } + @Test + fun sendViewModelTracksReceiverCompletionForCatalogProgress() = runTest { + Dispatchers.setMain(StandardTestDispatcher(testScheduler)) + val accepted = ReceiverRequestModel( + id = "request-7", + transferId = 7UL, + remoteEndpointId = "peer-a", + transferName = "Photo", + receiverName = "Receiver", + receiverDeviceName = null, + appVersion = "1.0", + status = ReceiverDeliveryStatus.Accepted, + reason = null, + requestedAt = 1L, + respondedAt = 2L, + completedAt = null, + ) + val core = FakeCoreGateway().apply { + mutableState.value = CoreState(isInitialized = true, transfers = listOf(sentTransfer(7UL))) + requests[7UL] = listOf(accepted) + } + val viewModel = SendViewModel(core, FakeFileSystemService(folder), preferences(), FakeFilePreviewRepository(), UiMessageController()) + advanceUntilIdle() + assertEquals(ReceiverDeliveryStatus.Accepted, viewModel.state.value.receiversByTransfer.getValue(7UL).single().status) + + core.requests[7UL] = listOf(accepted.copy(status = ReceiverDeliveryStatus.Completed, completedAt = 3L)) + core.mutableSignals.emit(CoreSignal.ReceiverHistoryChanged(7UL)) + advanceUntilIdle() + + assertEquals(ReceiverDeliveryStatus.Completed, viewModel.state.value.receiversByTransfer.getValue(7UL).single().status) + } + @Test fun sendComposerClosesAfterSuccessfulAtomicShareCreation() = runTest { Dispatchers.setMain(StandardTestDispatcher(testScheduler)) @@ -722,6 +757,22 @@ class ViewModelsTest { updatedAt = 1L, ) + private fun sentTransfer(id: ULong) = Transfer( + localId = "send-$id", + transferId = id, + direction = TransferDirection.Send, + status = TransferStatus.Sharing, + peerId = null, + transferName = "Sent $id", + contentHash = "hash-$id", + fileCount = 1UL, + totalSize = 42UL, + ticket = "ticket-$id", + accessPolicy = ShareAccessPolicy.RequireApproval, + createdAt = 1L, + updatedAt = 1L, + ) + private fun sampleTicketInspection() = com.vnidrop.app.core.TicketInspectionModel( kind = "vnidrop", metadata = com.vnidrop.app.core.TransferMetadataModel(1UL, "Photo", null, "hash", 1UL, 42UL), diff --git a/shared/src/commonTest/kotlin/com/vnidrop/app/ui/state/AppUiModelsTest.kt b/shared/src/commonTest/kotlin/com/vnidrop/app/ui/state/AppUiModelsTest.kt index 1fa0a42..01865c6 100644 --- a/shared/src/commonTest/kotlin/com/vnidrop/app/ui/state/AppUiModelsTest.kt +++ b/shared/src/commonTest/kotlin/com/vnidrop/app/ui/state/AppUiModelsTest.kt @@ -211,9 +211,10 @@ class AppUiModelsTest { direction = "send", ), ) - val progress = activeSendProgress(events, 7UL, totalSizeHint = 100UL) + val progress = activeSendProgress(events, 7UL, setOf("peer-a"), totalSizeHint = 100UL) assertEquals(0.3f, progress?.progress) assertEquals(Res.string.progress_sending, progress?.label) + assertEquals(null, activeSendProgress(events, 7UL, emptySet(), totalSizeHint = 100UL)) } @Test @@ -270,6 +271,39 @@ class AppUiModelsTest { assertEquals(1f, progress?.progress) } + @Test + fun progressForReceiverCompletedAfterProgressUsesCompletedLabel() { + val events = listOf( + event( + id = "done", + phase = "transfer", + kind = "completed", + data = """{"connection_id":1,"request_id":1,"endpoint_id":"peer-a"}""", + direction = "send", + ), + event( + id = "progress", + phase = "transfer", + kind = "progress", + data = """{"connection_id":1,"request_id":1,"endpoint_id":"peer-a","end_offset":100}""", + direction = "send", + ), + event( + id = "started", + phase = "transfer", + kind = "started", + data = """{"connection_id":1,"request_id":1,"endpoint_id":"peer-a","size":100}""", + direction = "send", + ), + ) + + val progress = progressForReceiver(events, 7UL, "peer-a", totalSizeHint = 100UL) + assertEquals("completed", progress?.kind) + assertEquals(Res.string.progress_completed, progress?.label) + assertEquals(1f, progress?.progress) + assertEquals(null, activeSendProgress(events, 7UL, setOf("peer-a"), totalSizeHint = 100UL)) + } + private fun storedTransfer(status: TransferStatus): Transfer = Transfer( localId = "local-1",