cdricms/vnidrop
1
1
Fork 0
mirror of https://github.com/sudosylabs/vnidrop.git synced 2026-08-05 02:29:55 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #28 from sudosylabs/fix/transfer-completion-progress

Browse Source 
fix(transfer): finalize delivery completion
This commit is contained in:
Hammed Abass
2026-07-22 21:54:57 +02:00
committed by GitHub
parent ad9ffd63f5 f0b06ad1cf
commit 90a06e9151
20 changed files with 580 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
apple/Tests/ProgressDerivationTests.swift
View File

@@ -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)
}
}

2
apple/VniDrop/Core/CoreRepository.swift
View File

@@ -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
}

4
apple/VniDrop/Core/TransferProgress.swift
View File

@@ -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)

11
apple/VniDrop/Features/Send/SendScreen.swift
View File

@@ -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)

110
crates/vnidrop/src/repository.rs
View File

@@ -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<Self> {
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<Vec<PendingDeliveryReceipt>> {
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::<i64, _>("local_transfer_id") as u64,
sender_blob_ticket: row.get("sender_blob_ticket"),
request_id: row.get("request_id"),
sender_transfer_id: row.get::<i64, _>("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,

135
crates/vnidrop/src/runtime/delivery.rs Normal file
View File

@@ -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<Self>) {
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" }),
),
}
}
}

4
crates/vnidrop/src/runtime/lifecycle.rs
View File

@@ -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",

8
crates/vnidrop/src/runtime/mod.rs
View File

@@ -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<HashMap<String, HashSet<u64>>>,
pub(super) connection_endpoints: TokioMutex<HashMap<u64, String>>,
pub(super) provider_task: TokioMutex<Option<JoinHandle<()>>>,
pub(super) delivery_receipt_notify: Notify,
pub(super) delivery_receipt_task: TokioMutex<Option<JoinHandle<()>>>,
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)
}

43
crates/vnidrop/src/runtime/receive.rs
View File

@@ -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(())
}

60
crates/vnidrop/src/tests/repository.rs
View File

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

59
crates/vnidrop/src/tests/runtime.rs
View File

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

33
crates/vnidrop/tests/approval.rs
View File

@@ -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<vnidrop::ReceiverRequest> {
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"));

2
shared/src/commonMain/kotlin/com/vnidrop/app/core/CoreRepository.kt
View File

@@ -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()) {

14
shared/src/commonMain/kotlin/com/vnidrop/app/feature/send/SendCatalog.kt
View File

@@ -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<Transfer>,
transferThumbnails: Map<ULong, ByteArray>,
receiversByTransfer: Map<ULong, List<ReceiverRequestModel>> = emptyMap(),
events: List<CoreEventModel> = 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(

1
shared/src/commonMain/kotlin/com/vnidrop/app/feature/send/SendScreen.kt
View File

@@ -69,6 +69,7 @@ fun SendScreen(
TransferCatalog(
transfers = outgoingTransfers,
transferThumbnails = state.transferThumbnails,
receiversByTransfer = state.receiversByTransfer,
events = coreState.events,
windowClass = windowClass,
onOpenComposer = onOpenComposer,

29
shared/src/commonMain/kotlin/com/vnidrop/app/feature/send/SendViewModel.kt
View File

@@ -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<ULong, ByteArray> = emptyMap(),
val detailPanel: TransferDetailPanel? = null,
val receiverHistory: List<ReceiverRequestModel> = emptyList(),
val receiversByTransfer: Map<ULong, List<ReceiverRequestModel>> = 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<ULong>) {
_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))
}
}
}
}
}

14
shared/src/commonMain/kotlin/com/vnidrop/app/feature/settings/SettingsOverview.kt
View File

@@ -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),

27
shared/src/commonMain/kotlin/com/vnidrop/app/ui/state/AppUiModels.kt
View File

@@ -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<CoreEventModel>,
transferId: ULong,
activeReceiverEndpointIds: Set<String>,
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<CoreEventModel>): List<TransferProgress> =

51
shared/src/commonTest/kotlin/com/vnidrop/app/feature/ViewModelsTest.kt
View File

@@ -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),

36
shared/src/commonTest/kotlin/com/vnidrop/app/ui/state/AppUiModelsTest.kt
View File

@@ -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",