From b698fdaac147ec07b071fe8308f6057350de7488 Mon Sep 17 00:00:00 2001 From: kiyo-e Date: Thu, 28 May 2026 00:04:53 +0900 Subject: [PATCH] Add same-room transfer resume --- README.ja.md | 2 + README.md | 2 + cli/src/main.rs | 625 +++++++++++++++++++++++++++---------- docs/signaling-protocol.md | 19 +- src/client/room.tsx | 123 +++++++- 5 files changed, 580 insertions(+), 191 deletions(-) diff --git a/README.ja.md b/README.ja.md index 44bc0c8..fe4815d 100644 --- a/README.ja.md +++ b/README.ja.md @@ -37,6 +37,8 @@ WebRTCを使ったP2Pファイル共有ツール。サーバーを経由せず 転送にはSHA-256チェックサムが含まれます。受信側はダウンロードしたバイト列が送信側のチェックサムと一致した場合だけ完了として扱います。 +転送が途中で切れた場合、受信側のプロセスまたはブラウザページが生きている間は同じルーム内で途中から再開できます。CLIの再起動やページ再読み込み後は最初から転送します。 + ### クイックスタート ```sh diff --git a/README.md b/README.md index 8bad24a..8b4f4d4 100644 --- a/README.md +++ b/README.md @@ -37,6 +37,8 @@ Send and receive files directly from your terminal. Works with browsers and othe Transfers include a SHA-256 checksum. Receivers only mark the file complete after the downloaded bytes match the sender's checksum. +Interrupted transfers can resume within the same room while the receiver process or browser page is still running. Restarting the CLI or reloading the page starts the transfer from the beginning. + ### Quick Start ```sh diff --git a/cli/src/main.rs b/cli/src/main.rs index fc8d294..8d2959a 100644 --- a/cli/src/main.rs +++ b/cli/src/main.rs @@ -15,7 +15,7 @@ use std::env; use std::path::{Path, PathBuf}; use std::sync::Arc; use tokio::fs::File; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt, SeekFrom}; use tokio::sync::{mpsc, Mutex}; use tokio::time::{sleep, timeout, Duration}; use tokio_tungstenite::connect_async; @@ -23,12 +23,12 @@ use tokio_tungstenite::tungstenite::Message; use url::form_urlencoded; use url::Url; use uuid::Uuid; -use webrtc::data_channel::data_channel_state::RTCDataChannelState; use webrtc::api::interceptor_registry::register_default_interceptors; use webrtc::api::media_engine::MediaEngine; use webrtc::api::APIBuilder; use webrtc::data_channel::data_channel_init::RTCDataChannelInit; use webrtc::data_channel::data_channel_message::DataChannelMessage; +use webrtc::data_channel::data_channel_state::RTCDataChannelState; use webrtc::data_channel::RTCDataChannel; use webrtc::ice_transport::ice_candidate::RTCIceCandidateInit; use webrtc::ice_transport::ice_server::RTCIceServer; @@ -58,11 +58,24 @@ enum Command { Send { #[arg(value_name = "FILE", help = "File to send")] file: Option, - #[arg(long = "file", value_name = "PATH", help = "File to send (legacy --file)", hide = true)] + #[arg( + long = "file", + value_name = "PATH", + help = "File to send (legacy --file)", + hide = true + )] file_flag: Option, - #[arg(value_name = "ROOM_ID_OR_URL", help = "Room ID or full room URL (optional, supports #k=...)")] + #[arg( + value_name = "ROOM_ID_OR_URL", + help = "Room ID or full room URL (optional, supports #k=...)" + )] room_input: Option, - #[arg(long = "room-id", value_name = "ROOM_ID_OR_URL", help = "Room ID or full room URL (legacy --room-id)", hide = true)] + #[arg( + long = "room-id", + value_name = "ROOM_ID_OR_URL", + help = "Room ID or full room URL (legacy --room-id)", + hide = true + )] room_id: Option, #[arg(long, value_name = "URL", help = "Override signaling endpoint")] endpoint: Option, @@ -72,15 +85,32 @@ enum Command { stay_open: bool, }, Receive { - #[arg(value_name = "ROOM_ID_OR_URL", help = "Room ID or full room URL (supports #k=...)")] + #[arg( + value_name = "ROOM_ID_OR_URL", + help = "Room ID or full room URL (supports #k=...)" + )] room_input: Option, - #[arg(long = "room-id", value_name = "ROOM_ID_OR_URL", help = "Room ID or full room URL (legacy --room-id)", hide = true)] + #[arg( + long = "room-id", + value_name = "ROOM_ID_OR_URL", + help = "Room ID or full room URL (legacy --room-id)", + hide = true + )] room_id: Option, - #[arg(long, default_value = ".", value_name = "DIR", help = "Output directory")] + #[arg( + long, + default_value = ".", + value_name = "DIR", + help = "Output directory" + )] output_dir: PathBuf, #[arg(long, value_name = "URL", help = "Override signaling endpoint")] endpoint: Option, - #[arg(long, value_name = "KEY", help = "Base64url decryption key (overrides #k=...)")] + #[arg( + long, + value_name = "KEY", + help = "Base64url decryption key (overrides #k=...)" + )] key: Option, #[arg(long, help = "Keep running after a successful receive")] stay_open: bool, @@ -99,28 +129,61 @@ enum ServerMessage { #[serde(rename = "wait")] Wait { position: Option }, #[serde(rename = "start")] - Start { #[serde(rename = "peerId")] peer_id: Option }, + Start { + #[serde(rename = "peerId")] + peer_id: Option, + }, #[serde(rename = "peer-left")] - PeerLeft { #[serde(rename = "peerId")] peer_id: String }, + PeerLeft { + #[serde(rename = "peerId")] + peer_id: String, + }, #[serde(rename = "offer")] - Offer { from: String, sid: u64, sdp: RTCSessionDescription }, + Offer { + from: String, + sid: u64, + sdp: RTCSessionDescription, + }, #[serde(rename = "answer")] - Answer { from: String, sid: u64, sdp: RTCSessionDescription }, + Answer { + from: String, + sid: u64, + sdp: RTCSessionDescription, + }, #[serde(rename = "candidate")] - Candidate { from: String, sid: u64, candidate: RTCIceCandidateInit }, + Candidate { + from: String, + sid: u64, + candidate: RTCIceCandidateInit, + }, } #[derive(Debug, Serialize)] #[serde(tag = "type")] enum ClientMessage { #[serde(rename = "offer")] - Offer { to: String, sid: u64, sdp: RTCSessionDescription }, + Offer { + to: String, + sid: u64, + sdp: RTCSessionDescription, + }, #[serde(rename = "answer")] - Answer { to: String, sid: u64, sdp: RTCSessionDescription }, + Answer { + to: String, + sid: u64, + sdp: RTCSessionDescription, + }, #[serde(rename = "candidate")] - Candidate { to: String, sid: u64, candidate: RTCIceCandidateInit }, + Candidate { + to: String, + sid: u64, + candidate: RTCIceCandidateInit, + }, #[serde(rename = "transfer-done")] - TransferDone { #[serde(rename = "peerId")] peer_id: String }, + TransferDone { + #[serde(rename = "peerId")] + peer_id: String, + }, } #[derive(Debug, Deserialize)] @@ -132,6 +195,7 @@ enum DataMessage { size: u64, mime: String, encrypted: bool, + offset: Option, }, #[serde(rename = "done")] Done { sha256: String }, @@ -139,9 +203,16 @@ enum DataMessage { Capabilities { #[serde(rename = "maxChunkSize")] max_chunk_size: usize, + #[serde(rename = "resumeOffset")] + resume_offset: Option, }, } +struct ReceiverCapabilities { + max_chunk_size: usize, + resume_offset: u64, +} + struct RoomInput { room_id: String, endpoint: Option, @@ -187,6 +258,8 @@ struct ReceiveProgress { output_dir: PathBuf, current_file: Option, final_file: Option, + current_name: Option, + current_mime: Option, file: Option, expected_size: u64, expected_sha256: Option, @@ -218,7 +291,14 @@ async fn main() -> Result<()> { .or(file) .ok_or_else(|| anyhow!("File path is required (usage: send )"))?; let room_input = room_id.or(room_input); - run_send(room_input.as_deref(), &file, endpoint.as_deref(), no_encrypt, stay_open).await + run_send( + room_input.as_deref(), + &file, + endpoint.as_deref(), + no_encrypt, + stay_open, + ) + .await } Command::Receive { room_input, @@ -232,7 +312,15 @@ async fn main() -> Result<()> { let room_input = room_id .or(room_input) .ok_or_else(|| anyhow!("Room ID or URL is required (usage: receive )"))?; - run_receive(&room_input, &output_dir, endpoint.as_deref(), key.as_deref(), stay_open, debug).await + run_receive( + &room_input, + &output_dir, + endpoint.as_deref(), + key.as_deref(), + stay_open, + debug, + ) + .await } } } @@ -292,7 +380,10 @@ async fn run_send( let writer = tokio::spawn(async move { while let Some(msg) = signal_rx.recv().await { let text = serde_json::to_string(&msg).map_err(|err| anyhow!(err))?; - ws_write.send(Message::Text(text)).await.map_err(|err| anyhow!(err))?; + ws_write + .send(Message::Text(text)) + .await + .map_err(|err| anyhow!(err))?; } Ok::<(), anyhow::Error>(()) }); @@ -405,29 +496,15 @@ async fn run_receive( let ws_url = build_ws_url(endpoint_override, &room_id, &client_id)?; log_line("[room] id", &room_id); - log_line("[ws] connecting", &ws_url.to_string()); - let (ws_stream, _) = connect_async(ws_url.to_string()) - .await - .context("connect signaling websocket")?; - let (mut ws_write, mut ws_read) = ws_stream.split(); - - let (signal_tx, mut signal_rx) = mpsc::unbounded_channel::(); let (success_tx, mut success_rx) = mpsc::unbounded_channel::(); let success_tx = if stay_open { None } else { Some(success_tx) }; - - let writer = tokio::spawn(async move { - while let Some(msg) = signal_rx.recv().await { - let text = serde_json::to_string(&msg).map_err(|err| anyhow!(err))?; - ws_write.send(Message::Text(text)).await.map_err(|err| anyhow!(err))?; - } - Ok::<(), anyhow::Error>(()) - }); - - let receiver_state: Arc>> = Arc::new(Mutex::new(None)); + let (reconnect_tx, mut reconnect_rx) = mpsc::unbounded_channel::(); let progress = Arc::new(Mutex::new(ReceiveProgress { output_dir: output_dir.to_path_buf(), current_file: None, final_file: None, + current_name: None, + current_mime: None, file: None, expected_size: 0, expected_sha256: None, @@ -441,120 +518,166 @@ async fn run_receive( success_tx, })); - let mut completed = None; loop { - tokio::select! { - msg = ws_read.next() => { - let msg = match msg { - Some(msg) => msg.context("websocket read")?, - None => break, - }; - if let Message::Text(text) = msg { - let parsed: ServerMessage = match serde_json::from_str(&text) { - Ok(msg) => msg, - Err(_) => continue, + log_line("[ws] connecting", &ws_url.to_string()); + let (ws_stream, _) = connect_async(ws_url.to_string()) + .await + .context("connect signaling websocket")?; + let (mut ws_write, mut ws_read) = ws_stream.split(); + let (signal_tx, mut signal_rx) = mpsc::unbounded_channel::(); + let receiver_state: Arc>> = Arc::new(Mutex::new(None)); + + let writer = tokio::spawn(async move { + while let Some(msg) = signal_rx.recv().await { + let text = serde_json::to_string(&msg).map_err(|err| anyhow!(err))?; + ws_write + .send(Message::Text(text)) + .await + .map_err(|err| anyhow!(err))?; + } + Ok::<(), anyhow::Error>(()) + }); + + let mut completed = None; + let mut should_reconnect = false; + loop { + tokio::select! { + msg = ws_read.next() => { + let msg = match msg { + Some(msg) => msg.context("websocket read")?, + None => break, }; + if let Message::Text(text) = msg { + let parsed: ServerMessage = match serde_json::from_str(&text) { + Ok(msg) => msg, + Err(_) => continue, + }; - match parsed { - ServerMessage::Role { role, cid } => { - log_line("[ws] role", &format!("{role} ({cid})")); - if role != "answerer" { - return Err(anyhow!("This command must be the answerer; connect after the sender.")); + match parsed { + ServerMessage::Role { role, cid } => { + log_line("[ws] role", &format!("{role} ({cid})")); + if role != "answerer" { + return Err(anyhow!("This command must be the answerer; connect after the sender.")); + } } - } - ServerMessage::Peers { count } => { - log_line("[ws] peers", &count.to_string()); - } - ServerMessage::Wait { position } => { - let label = position.map(|p| p.to_string()).unwrap_or_else(|| "waiting".to_string()); - log_line("[ws] queue", &label); - } - ServerMessage::Start { .. } => { - let pc = create_peer_connection().await?; - let tx = signal_tx.clone(); - let receiver_state_for_ice = receiver_state.clone(); - pc.on_ice_candidate(Box::new(move |candidate| { - let tx = tx.clone(); - let receiver_state = receiver_state_for_ice.clone(); - Box::pin(async move { - if let Some(candidate) = candidate { - let candidate = candidate.to_json().unwrap_or_default(); - let guard = receiver_state.lock().await; - if let Some(state) = guard.as_ref() { - if let (Some(peer_id), Some(sid)) = (state.peer_id.clone(), state.active_sid) { - let _ = tx.send(ClientMessage::Candidate { to: peer_id, sid, candidate }); + ServerMessage::Peers { count } => { + log_line("[ws] peers", &count.to_string()); + } + ServerMessage::Wait { position } => { + let label = position.map(|p| p.to_string()).unwrap_or_else(|| "waiting".to_string()); + log_line("[ws] queue", &label); + } + ServerMessage::Start { .. } => { + let pc = create_peer_connection().await?; + let tx = signal_tx.clone(); + let receiver_state_for_ice = receiver_state.clone(); + pc.on_ice_candidate(Box::new(move |candidate| { + let tx = tx.clone(); + let receiver_state = receiver_state_for_ice.clone(); + Box::pin(async move { + if let Some(candidate) = candidate { + let candidate = candidate.to_json().unwrap_or_default(); + let guard = receiver_state.lock().await; + if let Some(state) = guard.as_ref() { + if let (Some(peer_id), Some(sid)) = (state.peer_id.clone(), state.active_sid) { + let _ = tx.send(ClientMessage::Candidate { to: peer_id, sid, candidate }); + } } } - } - }) - })); - - let rx_progress = progress.clone(); - pc.on_data_channel(Box::new(move |dc| { - let rx_progress = rx_progress.clone(); - Box::pin(async move { - debug_log(&rx_progress, "[rtc] datachannel", "open").await; - wire_receiver_channel(dc, rx_progress).await; - }) - })); - - *receiver_state.lock().await = Some(ReceiverState { - pc, - peer_id: None, - active_sid: None, - pending_candidates: Vec::new(), - remote_desc_set: false, - }); - } - ServerMessage::Offer { from, sid, sdp } => { - let mut guard = receiver_state.lock().await; - let state = guard.as_mut().ok_or_else(|| anyhow!("Receiver not initialized"))?; - state.peer_id = Some(from.clone()); - state.active_sid = Some(sid); - state.pc.set_remote_description(sdp).await?; - state.remote_desc_set = true; - flush_receiver_candidates(state).await?; - - let answer = state.pc.create_answer(None).await?; - state.pc.set_local_description(answer).await?; - if let Some(local) = state.pc.local_description().await { - let _ = signal_tx.send(ClientMessage::Answer { to: from, sid, sdp: local }); + }) + })); + + let reconnect_on_state = reconnect_tx.clone(); + pc.on_peer_connection_state_change(Box::new(move |state: RTCPeerConnectionState| { + let reconnect_on_state = reconnect_on_state.clone(); + Box::pin(async move { + log_line("[rtc] connectionState", &format!("{:?}", state)); + if matches!(state, RTCPeerConnectionState::Disconnected | RTCPeerConnectionState::Failed) { + let _ = reconnect_on_state.send(false); + } + }) + })); + + let rx_progress = progress.clone(); + let reconnect_for_dc = reconnect_tx.clone(); + pc.on_data_channel(Box::new(move |dc| { + let rx_progress = rx_progress.clone(); + let reconnect_for_dc = reconnect_for_dc.clone(); + Box::pin(async move { + debug_log(&rx_progress, "[rtc] datachannel", "open").await; + wire_receiver_channel(dc, rx_progress, reconnect_for_dc).await; + }) + })); + + *receiver_state.lock().await = Some(ReceiverState { + pc, + peer_id: None, + active_sid: None, + pending_candidates: Vec::new(), + remote_desc_set: false, + }); } - } - ServerMessage::Candidate { from: _, sid, candidate } => { - let mut guard = receiver_state.lock().await; - if let Some(state) = guard.as_mut() { - handle_receiver_candidate(state, sid, candidate).await?; + ServerMessage::Offer { from, sid, sdp } => { + let mut guard = receiver_state.lock().await; + let state = guard.as_mut().ok_or_else(|| anyhow!("Receiver not initialized"))?; + state.peer_id = Some(from.clone()); + state.active_sid = Some(sid); + state.pc.set_remote_description(sdp).await?; + state.remote_desc_set = true; + flush_receiver_candidates(state).await?; + + let answer = state.pc.create_answer(None).await?; + state.pc.set_local_description(answer).await?; + if let Some(local) = state.pc.local_description().await { + let _ = signal_tx.send(ClientMessage::Answer { to: from, sid, sdp: local }); + } } + ServerMessage::Candidate { from: _, sid, candidate } => { + let mut guard = receiver_state.lock().await; + if let Some(state) = guard.as_mut() { + handle_receiver_candidate(state, sid, candidate).await?; + } + } + _ => {} } - _ => {} } } - } - result = success_rx.recv(), if !stay_open => { - let ok = result.unwrap_or(false); - if ok { - log_line("[recv] completed", "transfer done"); - } else { - log_line("[recv] error", "receive failed"); + force_reconnect = reconnect_rx.recv() => { + if force_reconnect.unwrap_or(false) || can_resume_receive(&progress).await { + should_reconnect = true; + break; + } } - if let Some(state) = receiver_state.lock().await.take() { - let _ = state.pc.close().await; + result = success_rx.recv(), if !stay_open => { + let ok = result.unwrap_or(false); + if ok { + log_line("[recv] completed", "transfer done"); + } else { + log_line("[recv] error", "receive failed"); + } + completed = Some(ok); + break; } - completed = Some(ok); - break; } } - } - if let Some(ok) = completed { + writer.abort(); - if !ok { - return Err(anyhow!("receive failed")); + if let Some(state) = receiver_state.lock().await.take() { + let _ = state.pc.close().await; } - return Ok(()); + if let Some(ok) = completed { + if !ok { + return Err(anyhow!("receive failed")); + } + return Ok(()); + } + if should_reconnect { + log_line("[recv] resume", "reconnecting"); + sleep(Duration::from_millis(500)).await; + continue; + } + break; } - drop(signal_tx); - writer.await??; Ok(()) } @@ -621,7 +744,7 @@ async fn create_offerer_peer( let dc_for_open = dc.clone(); let crypto = crypto.clone(); let success_tx = success_tx.clone(); - let (capability_tx, capability_rx) = mpsc::unbounded_channel::(); + let (capability_tx, capability_rx) = mpsc::unbounded_channel::(); let capability_rx = Arc::new(Mutex::new(capability_rx)); dc.on_message(Box::new(move |msg: DataChannelMessage| { let capability_tx = capability_tx.clone(); @@ -633,8 +756,15 @@ async fn create_offerer_peer( Ok(text) => text, Err(_) => return, }; - if let Ok(DataMessage::Capabilities { max_chunk_size }) = serde_json::from_str::(&text) { - let _ = capability_tx.send(max_chunk_size); + if let Ok(DataMessage::Capabilities { + max_chunk_size, + resume_offset, + }) = serde_json::from_str::(&text) + { + let _ = capability_tx.send(ReceiverCapabilities { + max_chunk_size, + resume_offset: resume_offset.unwrap_or(0), + }); } }) })); @@ -655,20 +785,26 @@ async fn create_offerer_peer( guard.sending = true; drop(guard); - let receiver_max = timeout(Duration::from_secs(2), async { + let capabilities = timeout(Duration::from_secs(2), async { capability_rx.lock().await.recv().await }) - .await - .ok() - .flatten() - .unwrap_or(CLI_SAFE_CHUNK_SIZE); - let chunk_size = choose_chunk_size(receiver_max); - - if let Err(err) = send_file(&dc, &file_info, crypto, chunk_size).await { + .await + .ok() + .flatten() + .unwrap_or(ReceiverCapabilities { + max_chunk_size: CLI_SAFE_CHUNK_SIZE, + resume_offset: 0, + }); + let chunk_size = choose_chunk_size(capabilities.max_chunk_size); + let resume_offset = capabilities.resume_offset.min(file_info.size); + + if let Err(err) = send_file(&dc, &file_info, crypto, chunk_size, resume_offset).await { log_line("[send] error", &format!("{err:#}")); return; } - let _ = send_tx.send(ClientMessage::TransferDone { peer_id: send_peer_id }); + let _ = send_tx.send(ClientMessage::TransferDone { + peer_id: send_peer_id, + }); if let Some(tx) = success_tx.as_ref() { let _ = tx.send(()); } @@ -680,7 +816,10 @@ async fn create_offerer_peer( Ok(peer) } -async fn send_offer(peer: Arc, signal_tx: mpsc::UnboundedSender) -> Result<()> { +async fn send_offer( + peer: Arc, + signal_tx: mpsc::UnboundedSender, +) -> Result<()> { let mut guard = peer.state.lock().await; if guard.active_sid.is_some() { return Ok(()); @@ -714,13 +853,19 @@ async fn handle_answer(peer: Arc, sid: u64, sdp: RTCSessionDescript Ok(()) } -async fn handle_offer_candidate(peer: Arc, sid: u64, candidate: RTCIceCandidateInit) -> Result<()> { +async fn handle_offer_candidate( + peer: Arc, + sid: u64, + candidate: RTCIceCandidateInit, +) -> Result<()> { let mut guard = peer.state.lock().await; if guard.remote_desc_set { drop(guard); peer.pc.add_ice_candidate(candidate).await?; } else { - guard.pending_candidates.push(PendingCandidate { sid, candidate }); + guard + .pending_candidates + .push(PendingCandidate { sid, candidate }); } Ok(()) } @@ -747,11 +892,17 @@ async fn flush_offer_candidates(peer: Arc) -> Result<()> { Ok(()) } -async fn handle_receiver_candidate(state: &mut ReceiverState, sid: u64, candidate: RTCIceCandidateInit) -> Result<()> { +async fn handle_receiver_candidate( + state: &mut ReceiverState, + sid: u64, + candidate: RTCIceCandidateInit, +) -> Result<()> { if state.remote_desc_set { state.pc.add_ice_candidate(candidate).await?; } else { - state.pending_candidates.push(PendingCandidate { sid, candidate }); + state + .pending_candidates + .push(PendingCandidate { sid, candidate }); } Ok(()) } @@ -775,40 +926,89 @@ async fn flush_receiver_candidates(state: &mut ReceiverState) -> Result<()> { Ok(()) } -async fn wire_receiver_channel(dc: Arc, progress: Arc>) { +async fn wire_receiver_channel( + dc: Arc, + progress: Arc>, + reconnect_tx: mpsc::UnboundedSender, +) { let dc_for_open = dc.clone(); + let progress_for_open = progress.clone(); dc.on_open(Box::new(move || { let dc = dc_for_open.clone(); + let progress = progress_for_open.clone(); Box::pin(async move { - let _ = send_capabilities(&dc).await; + let _ = send_capabilities(&dc, &progress).await; }) })); if dc.ready_state() == RTCDataChannelState::Open { - let _ = send_capabilities(&dc).await; + let _ = send_capabilities(&dc, &progress).await; } + let progress_for_close = progress.clone(); + let reconnect_for_close = reconnect_tx.clone(); + dc.on_close(Box::new(move || { + let progress = progress_for_close.clone(); + let reconnect_tx = reconnect_for_close.clone(); + Box::pin(async move { + if can_resume_receive(&progress).await { + let _ = reconnect_tx.send(false); + } + }) + })); + + let dc_for_message = dc.clone(); dc.on_message(Box::new(move |msg: DataChannelMessage| { let progress = progress.clone(); + let dc = dc_for_message.clone(); + let reconnect_tx = reconnect_tx.clone(); Box::pin(async move { if msg.is_string { if let Ok(text) = String::from_utf8(msg.data.to_vec()) { if let Ok(parsed) = serde_json::from_str::(&text) { match parsed { - DataMessage::Meta { name, size, mime, encrypted } => { + DataMessage::Meta { + name, + size, + mime, + encrypted, + offset, + } => { let mut guard = progress.lock().await; if encrypted && guard.crypto.is_none() { log_line("[recv] error", "encrypted files need a decryption key"); notify_receive_done(&mut guard, false); return; } - guard.encrypted = encrypted; let safe_name = sanitize_file_name(&name); let final_path = guard.output_dir.join(&safe_name); let path = guard.output_dir.join(format!("{safe_name}.partial")); + let offset = offset.unwrap_or(0).min(size); + let can_resume = offset > 0 + && guard.current_name.as_deref() == Some(safe_name.as_str()) + && guard.current_mime.as_deref() == Some(mime.as_str()) + && guard.expected_size == size + && guard.encrypted == encrypted + && guard.received == offset + && guard.file.is_some(); + if offset > 0 && !can_resume { + log_line("[recv] resume", "offset mismatch; retrying from start"); + reset_receive_progress(&mut guard); + drop(guard); + let _ = reconnect_tx.send(true); + let _ = dc.close().await; + return; + } + if can_resume { + log_line("[recv] resume", &format!("continuing at {offset} bytes")); + return; + } + guard.encrypted = encrypted; match File::create(&path).await { Ok(file) => { guard.current_file = Some(path); guard.final_file = Some(final_path); + guard.current_name = Some(safe_name.clone()); + guard.current_mime = Some(mime.clone()); guard.file = Some(file); guard.expected_size = size; guard.expected_sha256 = None; @@ -816,7 +1016,10 @@ async fn wire_receiver_channel(dc: Arc, progress: Arc { log_line("[recv] error", &format!("{err:#}")); @@ -871,7 +1074,11 @@ async fn wire_receiver_channel(dc: Arc, progress: Arc, progress: Arc Result<()> { +async fn send_capabilities( + dc: &RTCDataChannel, + progress: &Arc>, +) -> Result<()> { + let resume_offset = { + let guard = progress.lock().await; + if guard.current_file.is_some() && guard.received > 0 && guard.received < guard.expected_size { + guard.received + } else { + 0 + } + }; let capabilities = serde_json::json!({ "type": "capabilities", "maxChunkSize": CLI_SAFE_CHUNK_SIZE, + "resumeOffset": resume_offset, }); dc.send_text(capabilities.to_string()).await?; Ok(()) @@ -901,6 +1120,7 @@ async fn send_file( info: &FileInfo, crypto: Option>, max_frame_size: usize, + resume_offset: u64, ) -> Result<()> { let encrypted = crypto.is_some(); let meta = serde_json::json!({ @@ -909,6 +1129,7 @@ async fn send_file( "size": info.size, "mime": info.mime, "encrypted": encrypted, + "offset": resume_offset, }); let meta_text = serde_json::to_string(&meta)?; dc.send_text(meta_text).await?; @@ -919,14 +1140,20 @@ async fn send_file( max_frame_size }; let mut file = File::open(&info.path).await?; + if resume_offset > 0 { + file.seek(SeekFrom::Start(resume_offset)).await?; + } let mut hasher = Sha256::new(); + let sent_from_start = resume_offset == 0; let mut buffer = vec![0u8; chunk_size]; loop { let read = file.read(&mut buffer).await?; if read == 0 { break; } - hasher.update(&buffer[..read]); + if sent_from_start { + hasher.update(&buffer[..read]); + } let payload = if let Some(crypto) = crypto.as_ref() { Bytes::from(encrypt_frame(crypto, &buffer[..read])?) } else { @@ -935,7 +1162,11 @@ async fn send_file( dc.send(&payload).await?; } - let sha256 = bytes_to_hex(&hasher.finalize()); + let sha256 = if sent_from_start { + bytes_to_hex(&hasher.finalize()) + } else { + file_sha256(&info.path).await? + }; let done = serde_json::json!({ "type": "done", "sha256": sha256 }); dc.send_text(serde_json::to_string(&done)?).await?; wait_for_drain(dc).await; @@ -962,6 +1193,20 @@ async fn load_file_info(path: &Path) -> Result { }) } +async fn file_sha256(path: &Path) -> Result { + let mut file = File::open(path).await?; + let mut hasher = Sha256::new(); + let mut buffer = vec![0u8; PREFERRED_CHUNK_SIZE]; + loop { + let read = file.read(&mut buffer).await?; + if read == 0 { + break; + } + hasher.update(&buffer[..read]); + } + Ok(bytes_to_hex(&hasher.finalize())) +} + async fn finalize_receive(progress: &mut ReceiveProgress) { if progress.file.is_none() { return; @@ -972,7 +1217,10 @@ async fn finalize_receive(progress: &mut ReceiveProgress) { if progress.received != progress.expected_size { log_line( "[recv] error", - &format!("size mismatch: got {} bytes, expected {} bytes", progress.received, progress.expected_size), + &format!( + "size mismatch: got {} bytes, expected {} bytes", + progress.received, progress.expected_size + ), ); notify_receive_done(progress, false); return; @@ -1008,6 +1256,8 @@ async fn finalize_receive(progress: &mut ReceiveProgress) { Some(path) => path, None => return, }; + progress.current_name = None; + progress.current_mime = None; if let Err(err) = tokio::fs::rename(&partial_path, &final_path).await { log_line("[recv] error", &format!("{err:#}")); notify_receive_done(progress, false); @@ -1024,6 +1274,29 @@ fn notify_receive_done(progress: &mut ReceiveProgress, ok: bool) { } } +fn reset_receive_progress(progress: &mut ReceiveProgress) { + progress.file = None; + progress.current_file = None; + progress.final_file = None; + progress.current_name = None; + progress.current_mime = None; + progress.expected_size = 0; + progress.expected_sha256 = None; + progress.hasher = None; + progress.received = 0; + progress.next_progress_percent = 10; + progress.saw_first_chunk = false; + progress.encrypted = false; +} + +async fn can_resume_receive(progress: &Arc>) -> bool { + let guard = progress.lock().await; + guard.current_file.is_some() + && guard.file.is_some() + && guard.received > 0 + && guard.received < guard.expected_size +} + fn choose_chunk_size(receiver_max: usize) -> usize { receiver_max.clamp(CLI_SAFE_CHUNK_SIZE, PREFERRED_CHUNK_SIZE) } @@ -1111,7 +1384,11 @@ fn parse_room_url(url: &Url) -> Result { Some(fragment) => parse_key_fragment(fragment)?, None => None, }; - Ok(RoomInput { room_id, endpoint, key }) + Ok(RoomInput { + room_id, + endpoint, + key, + }) } fn extract_room_id_from_url(url: &Url) -> Result { @@ -1162,7 +1439,9 @@ fn decrypt_frame(crypto: &Aes256Gcm, frame: &[u8]) -> Result> { } let (nonce_bytes, ciphertext) = frame.split_at(AES_NONCE_LEN); let nonce = Nonce::from_slice(nonce_bytes); - crypto.decrypt(nonce, ciphertext).map_err(|err| anyhow!(err)) + crypto + .decrypt(nonce, ciphertext) + .map_err(|err| anyhow!(err)) } fn b64url_encode(value: &[u8]) -> String { @@ -1211,13 +1490,19 @@ fn build_ws_url(endpoint: Option<&str>, room_id: &str, client_id: &str) -> Resul "ws" => "ws", other => return Err(anyhow!("Unsupported endpoint scheme: {other}")), }; - url.set_scheme(scheme).map_err(|_| anyhow!("Invalid endpoint scheme"))?; + url + .set_scheme(scheme) + .map_err(|_| anyhow!("Invalid endpoint scheme"))?; url.set_path(&format!("/ws/{room_id}")); url.set_query(Some(&format!("cid={client_id}"))); Ok(url) } -fn build_room_url_with_key(endpoint: Option<&str>, room_id: &str, key: Option<&[u8]>) -> Result { +fn build_room_url_with_key( + endpoint: Option<&str>, + room_id: &str, + key: Option<&[u8]>, +) -> Result { let mut url = base_endpoint_url(endpoint)?; url.set_path(&format!("/r/{room_id}")); url.set_query(None); @@ -1246,7 +1531,9 @@ fn base_endpoint_url(endpoint: Option<&str>) -> Result { "ws" => "http".to_string(), other => return Err(anyhow!("Unsupported endpoint scheme: {other}")), }; - url.set_scheme(&scheme).map_err(|_| anyhow!("Invalid endpoint scheme"))?; + url + .set_scheme(&scheme) + .map_err(|_| anyhow!("Invalid endpoint scheme"))?; url.set_path(""); url.set_query(None); url.set_fragment(None); @@ -1277,7 +1564,9 @@ async fn create_room(endpoint: Option<&str>, creator_cid: Option<&str>) -> Resul .send() .await .context("create room request")?; - let response = response.error_for_status().context("create room response")?; + let response = response + .error_for_status() + .context("create room response")?; let body: RoomResponse = response.json().await.context("parse room response")?; Ok(body.room_id) } diff --git a/docs/signaling-protocol.md b/docs/signaling-protocol.md index 0dc9c81..0905262 100644 --- a/docs/signaling-protocol.md +++ b/docs/signaling-protocol.md @@ -211,11 +211,12 @@ Once WebRTC connection is established: ### File Transfer Protocol ``` -Sender Receiver - │ │ - │◄─── { type: "capabilities", maxChunkSize } ─│ - │ │ - │──── { type: "meta", name, size, ... } ──►│ +Sender Receiver + │ │ + │◄─── { type: "capabilities", maxChunkSize, │ + │ resumeOffset? } ─────────────────────────│ + │ │ + │──── { type: "meta", name, size, offset?, ... } ►│ │ │ │──── [binary chunk 1] ───────────────────►│ │──── [binary chunk 2] ───────────────────►│ @@ -232,7 +233,8 @@ Sender Receiver name: string, // File name size: number, // File size in bytes mime: string, // MIME type - encrypted: boolean // Whether chunks are encrypted + encrypted: boolean, // Whether chunks are encrypted + offset?: number // Plaintext byte offset to start sending from } ``` @@ -241,12 +243,15 @@ Sender Receiver ```typescript { type: "capabilities", - maxChunkSize: number // Maximum binary message size this receiver accepts + maxChunkSize: number, // Maximum binary message size this receiver accepts + resumeOffset?: number // Plaintext bytes already received in this process/page } ``` Receivers send capabilities when the DataChannel opens. Senders choose the transfer chunk size from the receiver limit and their own preferred maximum. CLI receivers using the normal webrtc-rs `on_message` API advertise 16 KiB; browser receivers advertise a larger browser-friendly limit. +If a transfer is interrupted while the receiver process/page is still alive, the receiver may reconnect to the same room with the same client ID and advertise `resumeOffset`. The sender then repeats `meta` with `offset` and starts sending from that plaintext byte offset. This is an in-memory resume only; browser reloads and CLI restarts start over from byte 0. + #### Completion Message ```typescript diff --git a/src/client/room.tsx b/src/client/room.tsx index 82bbe43..4aba3d3 100644 --- a/src/client/room.tsx +++ b/src/client/room.tsx @@ -33,11 +33,12 @@ type IncomingMeta = { size: number; mime: string; encrypted: boolean; + offset?: number; }; type DoneMessage = { type: "done"; sha256: string }; -type CapabilitiesMessage = { type: "capabilities"; maxChunkSize: number }; +type CapabilitiesMessage = { type: "capabilities"; maxChunkSize: number; resumeOffset?: number }; type DataMessage = IncomingMeta | DoneMessage | CapabilitiesMessage; @@ -71,6 +72,7 @@ type OffererPeer = { sending: boolean; sent: boolean; maxChunkSize: number | null; + resumeOffset: number; }; const clientId = getClientId(); @@ -122,6 +124,7 @@ function RoomApp({ roomId, maxConcurrent }: RoomAppProps) { const recvChunksRef = useRef([]); const recvBytesRef = useRef(0); const downloadUrlRef = useRef(null); + const receiverReconnectTimerRef = useRef(null); useEffect(() => { roleRef.current = role; @@ -229,20 +232,22 @@ function RoomApp({ roomId, maxConcurrent }: RoomAppProps) { if (!dc) return; const encrypted = !!cryptoKeyRef.current; + const startOffset = clampOffset(peer.resumeOffset, file.size); const meta: OutgoingMeta = { type: "meta", name: file.name, size: file.size, mime: file.type || "application/octet-stream", encrypted, + offset: startOffset, }; log("[send] starting:", meta.name, "size:", meta.size, "peer:", peer.peerId); dc.send(JSON.stringify(meta)); setStatus(t.status.sending); - setSendProgress({ sent: 0, total: file.size }); + setSendProgress({ sent: startOffset, total: file.size }); - let sent = 0; + let sent = startOffset; const negotiatedChunkSize = peer.maxChunkSize ?? CLI_SAFE_CHUNK_SIZE; dc.bufferedAmountLowThreshold = 4 * 1024 * 1024; @@ -266,7 +271,7 @@ function RoomApp({ roomId, maxConcurrent }: RoomAppProps) { }; const chunkSize = encrypted ? negotiatedChunkSize - 12 - 16 : negotiatedChunkSize; - let offset = 0; + let offset = startOffset; while (offset < file.size) { const slice = file.slice(offset, offset + chunkSize); const buf = await slice.arrayBuffer(); @@ -316,7 +321,47 @@ function RoomApp({ roomId, maxConcurrent }: RoomAppProps) { }, [selectedFile, trySendAll]); useEffect(() => { - const boot = async () => { + let disposed = false; + + const receiverResumeOffset = () => { + const meta = incomingMetaRef.current; + if (!meta) return 0; + if (recvBytesRef.current <= 0 || recvBytesRef.current >= meta.size) return 0; + return recvBytesRef.current; + }; + + const closeReceiverConnection = () => { + receiverDcRef.current?.close(); + receiverPcRef.current?.close(); + receiverDcRef.current = null; + receiverPcRef.current = null; + receiverPeerIdRef.current = null; + receiverRemoteDescSetRef.current = false; + receiverPendingCandidatesRef.current = []; + receiverActiveSidRef.current = null; + }; + + const scheduleReceiverReconnect = (force = false) => { + if (disposed || roleRef.current !== "answerer") return; + if (!force && receiverResumeOffset() === 0) return; + if (receiverReconnectTimerRef.current != null) return; + + setStatus(t.status.connecting); + receiverReconnectTimerRef.current = window.setTimeout(() => { + receiverReconnectTimerRef.current = null; + startBoot(); + }, 500); + closeReceiverConnection(); + wsRef.current?.close(); + }; + + const startBoot = () => { + boot().catch((e) => { + setStatus(t.error.generic.replace("{message}", String(e?.message || e))); + }); + }; + + async function boot() { setStatus(t.status.connecting); const keyParam = new URLSearchParams(location.hash.slice(1)).get("k"); @@ -343,6 +388,7 @@ function RoomApp({ roomId, maxConcurrent }: RoomAppProps) { const m = safeJson(ev.data) as DataMessage | null; if (!m || m.type !== "capabilities") return; peer.maxChunkSize = chooseChunkSize(m.maxChunkSize); + peer.resumeOffset = clampOffset(m.resumeOffset ?? 0, selectedFileRef.current?.size ?? 0); void trySendPeer(peer, "capabilities"); }; }; @@ -367,6 +413,7 @@ function RoomApp({ roomId, maxConcurrent }: RoomAppProps) { sending: false, sent: false, maxChunkSize: null, + resumeOffset: 0, }; pc.onicecandidate = (ev) => { @@ -436,16 +483,25 @@ function RoomApp({ roomId, maxConcurrent }: RoomAppProps) { ch.binaryType = "arraybuffer"; ch.onopen = () => { log("[rtc] datachannel open (receiver)"); - ch.send(JSON.stringify({ type: "capabilities", maxChunkSize: WEB_PREFERRED_CHUNK_SIZE } satisfies CapabilitiesMessage)); + ch.send(JSON.stringify({ + type: "capabilities", + maxChunkSize: WEB_PREFERRED_CHUNK_SIZE, + resumeOffset: receiverResumeOffset(), + } satisfies CapabilitiesMessage)); setStatus(t.status.dataChannelReady); }; ch.onclose = () => { log("[rtc] datachannel close (receiver)"); - setStatus(t.status.dataChannelClosed); + if (receiverResumeOffset() > 0) { + scheduleReceiverReconnect(); + } else { + setStatus(t.status.dataChannelClosed); + } }; ch.onerror = () => { console.warn("[rtc] datachannel error (receiver)"); - setStatus(t.status.dataChannelError); + scheduleReceiverReconnect(); + if (receiverResumeOffset() === 0) setStatus(t.status.dataChannelError); }; ch.onmessage = async (ev) => { @@ -456,11 +512,29 @@ function RoomApp({ roomId, maxConcurrent }: RoomAppProps) { if (m.type === "meta") { log("[recv] starting:", m.name, "size:", m.size); + const offset = clampOffset(m.offset ?? 0, m.size); + const canResume = + offset > 0 && + incomingMetaRef.current && + isSameTransfer(incomingMetaRef.current, m) && + offset === recvBytesRef.current; + + if (offset > 0 && !canResume) { + incomingMetaRef.current = null; + recvChunksRef.current = []; + recvBytesRef.current = 0; + setRecvProgress({ got: 0, total: m.size }); + scheduleReceiverReconnect(true); + return; + } + incomingMetaRef.current = m; - recvChunksRef.current = []; - recvBytesRef.current = 0; + if (!canResume) { + recvChunksRef.current = []; + recvBytesRef.current = 0; + } setDownload(null); - setRecvProgress({ got: 0, total: m.size }); + setRecvProgress({ got: recvBytesRef.current, total: m.size }); if (m.encrypted && !cryptoKeyRef.current) { setStatus(t.status.missingKey); @@ -515,6 +589,9 @@ function RoomApp({ roomId, maxConcurrent }: RoomAppProps) { pc.onconnectionstatechange = () => { log("[rtc] connectionState:", pc.connectionState, "(receiver)"); + if (pc.connectionState === "failed" || pc.connectionState === "disconnected") { + scheduleReceiverReconnect(); + } }; pc.ondatachannel = (ev) => { @@ -658,15 +735,20 @@ function RoomApp({ roomId, maxConcurrent }: RoomAppProps) { } }; - ws.onclose = () => setStatus(t.status.disconnected); + ws.onclose = () => { + if (receiverReconnectTimerRef.current == null) setStatus(t.status.disconnected); + }; ws.onerror = () => console.warn("[ws] error"); - }; + } - boot().catch((e) => { - setStatus(t.error.generic.replace("{message}", String(e?.message || e))); - }); + startBoot(); return () => { + disposed = true; + if (receiverReconnectTimerRef.current != null) { + window.clearTimeout(receiverReconnectTimerRef.current); + receiverReconnectTimerRef.current = null; + } wsRef.current?.close(); receiverDcRef.current?.close(); receiverPcRef.current?.close(); @@ -1018,6 +1100,15 @@ function chooseChunkSize(receiverMax: number) { return Math.max(CLI_SAFE_CHUNK_SIZE, Math.min(WEB_PREFERRED_CHUNK_SIZE, Math.floor(receiverMax))); } +function clampOffset(value: number, size: number) { + if (!Number.isFinite(value) || value <= 0) return 0; + return Math.min(size, Math.floor(value)); +} + +function isSameTransfer(a: IncomingMeta, b: IncomingMeta) { + return a.name === b.name && a.size === b.size && a.mime === b.mime && a.encrypted === b.encrypted; +} + async function copyText(s: string) { try { await navigator.clipboard.writeText(s);