diff --git a/crates/lanspread-peer/src/handlers.rs b/crates/lanspread-peer/src/handlers.rs index c6bb619..052957a 100644 --- a/crates/lanspread-peer/src/handlers.rs +++ b/crates/lanspread-peer/src/handlers.rs @@ -4,6 +4,7 @@ use std::{ collections::{HashMap, HashSet}, fmt, future::Future, + net::IpAddr, path::{Path, PathBuf}, sync::Arc, time::Duration, @@ -59,6 +60,7 @@ use crate::{ ReceiveStreamedInstallRequest, StreamInstallReceiveError, StreamInstallReceiveErrorKind, + StreamInstallTotalDeadline, receive_streamed_install, }, transfer_status::{DownloadAttemptReporter, DownloadAttemptStatus}, @@ -166,6 +168,53 @@ const fn stream_install_failure_disposition( } } +const MAX_STREAM_INSTALL_SOURCE_IP_ATTEMPTS: usize = 4; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum StreamInstallSourceAdmission { + Admitted, + AlreadyAttempted, + Exhausted, +} + +#[derive(Default)] +struct StreamInstallSourceAttempts { + source_ips: HashSet, +} + +impl StreamInstallSourceAttempts { + fn admit(&mut self, source_ip: IpAddr) -> StreamInstallSourceAdmission { + if self.source_ips.contains(&source_ip) { + return StreamInstallSourceAdmission::AlreadyAttempted; + } + if self.source_ips.len() >= MAX_STREAM_INSTALL_SOURCE_IP_ATTEMPTS { + return StreamInstallSourceAdmission::Exhausted; + } + self.source_ips.insert(source_ip); + StreamInstallSourceAdmission::Admitted + } +} + +fn admit_stream_install_source( + attempts: &mut StreamInstallSourceAttempts, + source: &PeerEndpoint, + game_id: &str, +) -> StreamInstallSourceAdmission { + let admission = attempts.admit(source.addr.ip()); + match admission { + StreamInstallSourceAdmission::Admitted => {} + StreamInstallSourceAdmission::AlreadyAttempted => log::debug!( + "Skipping streamed-install source {} at {} for {game_id}: endpoint IP was already attempted", + source.peer_id, + source.addr + ), + StreamInstallSourceAdmission::Exhausted => log::warn!( + "Streamed install for {game_id} exhausted its {MAX_STREAM_INSTALL_SOURCE_IP_ATTEMPTS}-source-IP attempt budget" + ), + } + admission +} + #[derive(Debug)] struct StreamDownloadError { reason: Option, @@ -1079,6 +1128,57 @@ fn settle_failed_stream_receive( Ok((disposition, error)) } +enum StreamInstallFailureAction { + Retry { + error: StreamInstallReceiveError, + invalid_source: bool, + }, + Exhausted(StreamInstallReceiveError), + Stop(StreamDownloadError), +} + +fn stream_install_failure_action( + disposition: StreamInstallFailureDisposition, + error_kind: StreamInstallReceiveErrorKind, + error: StreamInstallReceiveError, + total_deadline: StreamInstallTotalDeadline, + source: &PeerEndpoint, + game_id: &str, +) -> StreamInstallFailureAction { + if disposition != StreamInstallFailureDisposition::Stop && total_deadline.is_elapsed() { + log::warn!("Streamed install for {game_id} exhausted its shared total receive deadline"); + return StreamInstallFailureAction::Exhausted(error); + } + + match disposition { + StreamInstallFailureDisposition::RetryAndQuarantine + | StreamInstallFailureDisposition::Retry => { + log::warn!( + "Streamed install attempt from {} at {} failed for {game_id}; trying another peer if available: {error}", + source.peer_id, + source.addr + ); + StreamInstallFailureAction::Retry { + error, + invalid_source: disposition == StreamInstallFailureDisposition::RetryAndQuarantine, + } + } + StreamInstallFailureDisposition::Stop => { + let error = eyre::Report::new(error); + StreamInstallFailureAction::Stop(match error_kind { + StreamInstallReceiveErrorKind::Cancelled => StreamDownloadError::cancelled(error), + StreamInstallReceiveErrorKind::Setup => { + StreamDownloadError::operation_failed(error) + } + StreamInstallReceiveErrorKind::Integrity + | StreamInstallReceiveErrorKind::Transport => { + unreachable!("retryable stream failures cannot stop immediately") + } + }) + } + } +} + fn exhausted_stream_receive_error( id: &str, last_receive_error: Option, @@ -1126,11 +1226,16 @@ async fn receive_streamed_install_from_peers( let mut last_receive_error = None; let mut retry_invalid_source = false; let content_id = manifest.content_id(); + let total_deadline = StreamInstallTotalDeadline::for_manifest(manifest) + .map_err(StreamDownloadError::operation_failed)?; + let mut source_attempts = StreamInstallSourceAttempts::default(); for source in sources { - if cancel_token.is_cancelled() { - return Err(StreamDownloadError::cancelled(eyre::eyre!( - "streamed install for {id} was cancelled" - ))); + if let Err(error) = total_deadline.ensure_active(id, cancel_token) { + if error.kind() == StreamInstallReceiveErrorKind::Cancelled { + return Err(StreamDownloadError::cancelled(eyre::Report::new(error))); + } + last_receive_error = Some(error); + break; } if ctx.content_quarantine.is_quarantined(source, content_id) { log::debug!( @@ -1140,6 +1245,11 @@ async fn receive_streamed_install_from_peers( ); continue; } + match admit_stream_install_source(&mut source_attempts, source, id) { + StreamInstallSourceAdmission::Admitted => {} + StreamInstallSourceAdmission::AlreadyAttempted => continue, + StreamInstallSourceAdmission::Exhausted => break, + } let transaction = begin_stream_receive_attempt( &game_root, @@ -1157,6 +1267,7 @@ async fn receive_streamed_install_from_peers( tx_notify_ui: tx_notify_ui.clone(), quic, cancel_token: cancel_token.clone(), + total_deadline, }) .await; @@ -1166,40 +1277,26 @@ async fn receive_streamed_install_from_peers( let error_kind = err.kind(); let (disposition, err) = settle_failed_stream_receive(ctx, source, content_id, id, transaction, err)?; - - match disposition { - StreamInstallFailureDisposition::RetryAndQuarantine => { - log::warn!( - "Streamed install attempt from {} at {} failed for {id}; trying another peer if available: {err}", - source.peer_id, - source.addr - ); - last_receive_error = Some(err); - retry_invalid_source = true; + match stream_install_failure_action( + disposition, + error_kind, + err, + total_deadline, + source, + id, + ) { + StreamInstallFailureAction::Retry { + error, + invalid_source, + } => { + last_receive_error = Some(error); + retry_invalid_source = invalid_source; } - StreamInstallFailureDisposition::Retry => { - log::warn!( - "Streamed install attempt from {} at {} failed for {id}; trying another peer if available: {err}", - source.peer_id, - source.addr - ); - last_receive_error = Some(err); - } - StreamInstallFailureDisposition::Stop => { - let error = eyre::Report::new(err); - return Err(match error_kind { - StreamInstallReceiveErrorKind::Cancelled => { - StreamDownloadError::cancelled(error) - } - StreamInstallReceiveErrorKind::Setup => { - StreamDownloadError::operation_failed(error) - } - StreamInstallReceiveErrorKind::Integrity - | StreamInstallReceiveErrorKind::Transport => { - unreachable!("retryable stream failures cannot stop immediately") - } - }); + StreamInstallFailureAction::Exhausted(error) => { + last_receive_error = Some(error); + break; } + StreamInstallFailureAction::Stop(error) => return Err(error), } } } @@ -3536,6 +3633,10 @@ mod tests { PeerEndpoint::new(peer_id(seed), addr(port)) } + fn endpoint_at(seed: u8, ip: [u8; 4], port: u16) -> PeerEndpoint { + PeerEndpoint::new(peer_id(seed), SocketAddr::from((ip, port))) + } + fn upsert(db: &mut PeerGameDB, endpoint: PeerEndpoint, game: Option<(&str, ContentId)>) { let ticket = db .begin_candidate_negotiation(endpoint) @@ -3637,6 +3738,47 @@ mod tests { ); } + #[test] + fn streamed_install_same_ip_identities_and_ports_share_one_attempt() { + let mut attempts = StreamInstallSourceAttempts::default(); + let first = endpoint_at(1, [192, 168, 1, 10], 12_000); + let same_ip_sybil = endpoint_at(2, [192, 168, 1, 10], 13_000); + + assert_eq!( + attempts.admit(first.addr.ip()), + StreamInstallSourceAdmission::Admitted + ); + assert_eq!( + attempts.admit(same_ip_sybil.addr.ip()), + StreamInstallSourceAdmission::AlreadyAttempted + ); + assert_eq!(attempts.source_ips.len(), 1); + } + + #[test] + fn streamed_install_stops_after_four_distinct_source_ip_attempts() { + let mut attempts = StreamInstallSourceAttempts::default(); + for seed in 1..=u8::try_from(MAX_STREAM_INSTALL_SOURCE_IP_ATTEMPTS) + .expect("attempt limit should fit u8") + { + let source = endpoint_at(seed, [10, 0, 0, seed], 12_000 + u16::from(seed)); + assert_eq!( + attempts.admit(source.addr.ip()), + StreamInstallSourceAdmission::Admitted + ); + } + + let excess = endpoint_at(100, [10, 0, 0, 100], 13_000); + assert_eq!( + attempts.admit(excess.addr.ip()), + StreamInstallSourceAdmission::Exhausted + ); + assert_eq!( + attempts.source_ips.len(), + MAX_STREAM_INSTALL_SOURCE_IP_ATTEMPTS + ); + } + #[test] fn streamed_install_retries_and_quarantines_only_typed_integrity_failures() { assert_eq!( diff --git a/crates/lanspread-peer/src/stream_install.rs b/crates/lanspread-peer/src/stream_install.rs index 64105b8..7eeed4e 100644 --- a/crates/lanspread-peer/src/stream_install.rs +++ b/crates/lanspread-peer/src/stream_install.rs @@ -62,6 +62,103 @@ const STREAM_INSTALL_PROGRESS_UPDATE_INTERVAL: Duration = Duration::from_millis( const STREAM_CHUNK_SIZE: usize = 256 * 1024; const UNRAR_LISTING_CAPTURE_LIMIT: usize = 64 * 1024 * 1024; const STREAM_INSTALL_INACTIVITY_TIMEOUT: Duration = Duration::from_mins(10); +const STREAM_INSTALL_TOTAL_BASE_TIMEOUT: Duration = Duration::from_mins(10); +/// A streamed install is expected to sustain at least 1 MiB/s of catalog- +/// authorized file progress across its complete retry budget. +const STREAM_INSTALL_MIN_BYTES_PER_SECOND: u64 = 1024 * 1024; +/// Allows one source to fail after sending almost everything and one replacement +/// source to complete without granting every attempted identity a fresh window. +const STREAM_INSTALL_TOTAL_COPY_ALLOWANCE: u64 = 2; + +fn stream_install_total_timeout(expected_file_bytes: u64) -> Duration { + let budgeted_bytes = expected_file_bytes.saturating_mul(STREAM_INSTALL_TOTAL_COPY_ALLOWANCE); + let transfer_seconds = budgeted_bytes.div_ceil(STREAM_INSTALL_MIN_BYTES_PER_SECOND); + STREAM_INSTALL_TOTAL_BASE_TIMEOUT.saturating_add(Duration::from_secs(transfer_seconds)) +} + +#[derive(Clone, Copy, Debug)] +pub(crate) struct StreamInstallTotalDeadline { + expires_at: TokioInstant, + timeout: Duration, +} + +impl StreamInstallTotalDeadline { + pub(crate) fn for_manifest(manifest: &CatalogContentManifest) -> eyre::Result { + let expected_file_bytes = expected_streamed_file_bytes(manifest)?; + Ok(Self::after(stream_install_total_timeout( + expected_file_bytes, + ))) + } + + fn after(timeout: Duration) -> Self { + Self { + expires_at: TokioInstant::now() + timeout, + timeout, + } + } + + #[must_use] + pub(crate) fn is_elapsed(self) -> bool { + TokioInstant::now() >= self.expires_at + } + + fn timeout_error(self, game_id: &str) -> StreamInstallReceiveError { + let timeout = self.timeout; + StreamInstallReceiveError::transport(eyre::eyre!( + "streamed install for {game_id} exceeded its shared total receive deadline of {timeout:?}" + )) + } + + pub(crate) fn ensure_active( + self, + game_id: &str, + cancel_token: &CancellationToken, + ) -> StreamInstallReceiveResult<()> { + if cancel_token.is_cancelled() { + return Err(StreamInstallReceiveError::cancelled( + game_id, + "while receiving across peer attempts", + )); + } + if self.is_elapsed() { + return Err(self.timeout_error(game_id)); + } + Ok(()) + } + + async fn run( + self, + operation: impl Future>, + game_id: &str, + cancel_token: &CancellationToken, + ) -> StreamInstallReceiveResult { + self.ensure_active(game_id, cancel_token)?; + let result = tokio::select! { + biased; + () = cancel_token.cancelled() => { + return Err(StreamInstallReceiveError::cancelled( + game_id, + "while receiving across peer attempts", + )); + } + () = time::sleep_until(self.expires_at) => { + return Err(self.timeout_error(game_id)); + } + result = operation => result, + }; + + if result.as_ref().err().is_some_and(|error| { + matches!( + error.kind(), + StreamInstallReceiveErrorKind::Integrity | StreamInstallReceiveErrorKind::Setup + ) + }) { + return result; + } + self.ensure_active(game_id, cancel_token)?; + result + } +} #[derive(Clone, Copy, Debug)] struct StreamInstallInactivityDeadline { @@ -81,7 +178,10 @@ impl StreamInstallInactivityDeadline { } } - fn reset_after_frame(&mut self) { + fn observe_frame(&mut self, outcome: ReceiveFrameOutcome) { + if outcome != ReceiveFrameOutcome::UsefulFileBytes { + return; + } self.expires_at = TokioInstant::now() + self.timeout; } @@ -973,6 +1073,19 @@ struct ActiveArchive { name: CanonicalCatalogPath, } +fn expected_streamed_file_bytes(manifest: &CatalogContentManifest) -> eyre::Result { + let expected_file_bytes = manifest + .streamed_install_files() + .iter() + .filter(|entry| entry.kind() == CatalogEntryKind::File) + .try_fold(0_u64, |total, entry| total.checked_add(entry.size())) + .ok_or_else(|| eyre::eyre!("catalog streamed file-size total overflow"))?; + if expected_file_bytes > MAX_CATALOG_TOTAL_BYTES { + eyre::bail!("catalog streamed files exceed the {MAX_CATALOG_TOTAL_BYTES}-byte total limit"); + } + Ok(expected_file_bytes) +} + impl CatalogStreamVerifier { fn new(game_id: &str, manifest: Arc) -> eyre::Result { if manifest.game_id() != game_id { @@ -1002,17 +1115,7 @@ impl CatalogStreamVerifier { eyre::bail!("catalog game {game_id} has streamed output but no root .eti archive"); } - let expected_file_bytes = manifest - .streamed_install_files() - .iter() - .filter(|entry| entry.kind() == CatalogEntryKind::File) - .try_fold(0_u64, |total, entry| total.checked_add(entry.size())) - .ok_or_else(|| eyre::eyre!("catalog streamed file-size total overflow"))?; - if expected_file_bytes > MAX_CATALOG_TOTAL_BYTES { - eyre::bail!( - "catalog streamed files exceed the {MAX_CATALOG_TOTAL_BYTES}-byte total limit" - ); - } + let expected_file_bytes = expected_streamed_file_bytes(&manifest)?; let seen_entries = vec![false; manifest.streamed_install_files().len()]; Ok(Self { @@ -1230,6 +1333,7 @@ impl CatalogStreamVerifier { #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum ReceiveFrameOutcome { Continue, + UsefulFileBytes, Complete, } @@ -1336,6 +1440,9 @@ impl StreamInstallReceiveState { }; let length = file.write_chunk(&bytes)?; self.progress.record_bytes(length); + if length > 0 { + return Ok(ReceiveFrameOutcome::UsefulFileBytes); + } } StreamInstallFrame::FileEnd { relative_path } => { self.verifier @@ -1385,10 +1492,26 @@ pub(crate) struct ReceiveStreamedInstallRequest<'a> { pub(crate) tx_notify_ui: PeerEventSender, pub(crate) quic: &'a QuicConnector, pub(crate) cancel_token: CancellationToken, + pub(crate) total_deadline: StreamInstallTotalDeadline, } pub(crate) async fn receive_streamed_install( request: ReceiveStreamedInstallRequest<'_>, +) -> StreamInstallReceiveResult<()> { + let total_deadline = request.total_deadline; + let game_id = request.game_id.to_owned(); + let cancel_token = request.cancel_token.clone(); + total_deadline + .run( + receive_streamed_install_within_deadline(request), + &game_id, + &cancel_token, + ) + .await +} + +async fn receive_streamed_install_within_deadline( + request: ReceiveStreamedInstallRequest<'_>, ) -> StreamInstallReceiveResult<()> { let ReceiveStreamedInstallRequest { endpoint, @@ -1399,6 +1522,7 @@ pub(crate) async fn receive_streamed_install( tx_notify_ui, quic, cancel_token, + total_deadline: _, } = request; let content_id = manifest.content_id(); let mut state = StreamInstallReceiveState::new(game_id, manifest, staging_dir, attempt)?; @@ -1461,11 +1585,10 @@ pub(crate) async fn receive_streamed_install( ) })? .freeze(); - inactivity.reset_after_frame(); let frame = decode_received_stream_install_frame(frame)?; - if state.handle_frame(frame, game_id, endpoint, content_id, &tx_notify_ui)? - == ReceiveFrameOutcome::Complete - { + let outcome = state.handle_frame(frame, game_id, endpoint, content_id, &tx_notify_ui)?; + inactivity.observe_frame(outcome); + if outcome == ReceiveFrameOutcome::Complete { return framed_rx.drain_fin(&cancel_token, inactivity).await; } } @@ -2230,6 +2353,199 @@ mod tests { assert!(error.to_string().contains("no frame progress")); } + #[test] + fn total_receive_timeout_allows_two_catalog_copies_at_one_mib_per_second() { + assert_eq!( + stream_install_total_timeout(0), + STREAM_INSTALL_TOTAL_BASE_TIMEOUT + ); + assert_eq!( + stream_install_total_timeout(STREAM_INSTALL_MIN_BYTES_PER_SECOND), + STREAM_INSTALL_TOTAL_BASE_TIMEOUT + Duration::from_secs(2) + ); + assert_eq!( + stream_install_total_timeout(STREAM_INSTALL_MIN_BYTES_PER_SECOND + 1), + STREAM_INSTALL_TOTAL_BASE_TIMEOUT + Duration::from_secs(3) + ); + } + + #[tokio::test(start_paused = true)] + async fn total_receive_deadline_is_shared_across_source_attempts() { + let cancellation = CancellationToken::new(); + let timeout = Duration::from_secs(10); + let deadline = StreamInstallTotalDeadline::after(timeout); + let started = TokioInstant::now(); + + let first_error = deadline + .run( + async { + time::sleep(Duration::from_secs(6)).await; + Err::<(), _>(StreamInstallReceiveError::transport(eyre::eyre!( + "first source failed" + ))) + }, + "game", + &cancellation, + ) + .await + .expect_err("the first source should fail before the total deadline"); + assert_eq!(first_error.kind(), StreamInstallReceiveErrorKind::Transport); + assert_eq!(started.elapsed(), Duration::from_secs(6)); + + let second_error = deadline + .run( + std::future::pending::>(), + "game", + &cancellation, + ) + .await + .expect_err("the second source must inherit the original deadline"); + assert_eq!( + second_error.kind(), + StreamInstallReceiveErrorKind::Transport + ); + assert!( + second_error + .to_string() + .contains("shared total receive deadline") + ); + assert_eq!(started.elapsed(), timeout); + } + + #[tokio::test(start_paused = true)] + async fn metadata_and_directory_frames_do_not_refresh_receive_inactivity() { + let staging = TempDir::new("lanspread-stream-inactivity-metadata"); + let manifest = test_catalog_manifest( + &["a.eti"], + vec![CatalogExtractedEntry::directory("bin").expect("directory should validate")], + ); + let content_id = manifest.content_id(); + let (event_tx, _event_rx) = crate::peer_event_channel(); + let status = crate::transfer_status::DownloadAttemptStatus::new( + crate::DownloadAttemptKey::next("game".to_owned()), + CancellationToken::new(), + event_tx.clone(), + ); + let mut state = + StreamInstallReceiveState::new("game", manifest, staging.path(), status.reporter()) + .expect("receive state should initialize"); + let mut inactivity = StreamInstallInactivityDeadline::after(Duration::from_secs(10)); + let original_expiry = inactivity.expires_at; + + time::advance(Duration::from_secs(8)).await; + let archive = state + .handle_frame( + StreamInstallFrame::ArchiveBegin { + archive_name: canonical_path("a.eti"), + solid: false, + unpacked_size: 0, + }, + "game", + peer_endpoint(), + content_id, + &event_tx, + ) + .expect("archive metadata should be accepted"); + assert_eq!(archive, ReceiveFrameOutcome::Continue); + inactivity.observe_frame(archive); + + time::advance(Duration::from_secs(1)).await; + let directory = state + .handle_frame( + StreamInstallFrame::Directory { + relative_path: canonical_path("bin"), + }, + "game", + peer_endpoint(), + content_id, + &event_tx, + ) + .expect("catalog directory should be accepted"); + assert_eq!(directory, ReceiveFrameOutcome::Continue); + inactivity.observe_frame(directory); + + assert_eq!(inactivity.expires_at, original_expiry); + } + + #[tokio::test(start_paused = true)] + async fn catalog_checked_staging_bytes_refresh_receive_inactivity() { + let staging = TempDir::new("lanspread-stream-inactivity-file"); + let payload = Bytes::from_static(b"x"); + let manifest = test_catalog_manifest( + &["a.eti"], + vec![ + CatalogExtractedEntry::file( + "payload.bin", + u64::try_from(payload.len()).expect("payload length should fit"), + Blake3Digest::hash(&payload), + ) + .expect("file should validate"), + ], + ); + let content_id = manifest.content_id(); + let (event_tx, _event_rx) = crate::peer_event_channel(); + let status = crate::transfer_status::DownloadAttemptStatus::new( + crate::DownloadAttemptKey::next("game".to_owned()), + CancellationToken::new(), + event_tx.clone(), + ); + let mut state = + StreamInstallReceiveState::new("game", manifest, staging.path(), status.reporter()) + .expect("receive state should initialize"); + state + .handle_frame( + StreamInstallFrame::ArchiveBegin { + archive_name: canonical_path("a.eti"), + solid: false, + unpacked_size: 1, + }, + "game", + peer_endpoint(), + content_id, + &event_tx, + ) + .expect("archive should begin"); + state + .handle_frame( + StreamInstallFrame::FileBegin { + relative_path: canonical_path("payload.bin"), + size: 1, + crc32: crc32_of(&payload), + }, + "game", + peer_endpoint(), + content_id, + &event_tx, + ) + .expect("catalog file should begin"); + let timeout = Duration::from_secs(10); + let mut inactivity = StreamInstallInactivityDeadline::after(timeout); + let original_expiry = inactivity.expires_at; + + time::advance(Duration::from_secs(9)).await; + let chunk = state + .handle_frame( + StreamInstallFrame::FileChunk { + bytes: payload.clone(), + }, + "game", + peer_endpoint(), + content_id, + &event_tx, + ) + .expect("catalog-sized chunk should be written"); + assert_eq!(chunk, ReceiveFrameOutcome::UsefulFileBytes); + inactivity.observe_frame(chunk); + + assert!(inactivity.expires_at > original_expiry); + assert_eq!(inactivity.expires_at, TokioInstant::now() + timeout); + assert_eq!( + std::fs::read(staging.path().join("payload.bin")) + .expect("staged payload should be readable"), + payload + ); + } + #[tokio::test] async fn progress_ticks_do_not_reset_a_pending_frame_deadline() { let cancellation = CancellationToken::new();