From efadc228eb53dbf85fbea3b76d489599d0d3e817 Mon Sep 17 00:00:00 2001 From: Mollusk Date: Sun, 21 Jun 2026 00:07:23 -0400 Subject: [PATCH] fix(files): keep serve connection alive until fetcher has the bytes Chat file fetches failed 100% of the time with "file fetch: read failed: read error: connection lost" (both images and arbitrary files, both directions). Root cause: the FileRouter serve handler called send.finish() and immediately returned Ok(()), which dropped the Connection. In QUIC, finish() only marks the stream's EOF -- it does not wait for the written bytes to be delivered and acknowledged -- so the connection's CONNECTION_CLOSE raced ahead of the still-in-flight stream data and the fetcher's read_to_end aborted. Fix: after finishing, wait on connection.closed() (bounded by FILE_FETCH_TIMEOUT) so the link stays up until the fetcher has read everything and closed the connection itself, which is the signal the transfer landed. Wire-compatible (no protocol change), so version stays 0.3.0; both peers just need the rebuilt binary since either side can be the file server. Adds tests/file_transfer_loopback.rs: a real two-endpoint serve->fetch round-trip over FILES_ALPN with a 2 MiB multi-packet blob (deterministic A/B: 0/20 pass without the fix, 20/20 with it) plus an unknown-id "gone" case. Co-Authored-By: Claude Opus 4.8 --- src/network/iroh_impl.rs | 9 ++ tests/file_transfer_loopback.rs | 141 ++++++++++++++++++++++++++++++++ 2 files changed, 150 insertions(+) create mode 100644 tests/file_transfer_loopback.rs diff --git a/src/network/iroh_impl.rs b/src/network/iroh_impl.rs index ba7e273..f0e62bc 100644 --- a/src/network/iroh_impl.rs +++ b/src/network/iroh_impl.rs @@ -489,6 +489,15 @@ impl iroh::protocol::ProtocolHandler for FileRouter { // Finish either way: an unknown id closes with an empty body, which // the fetcher reads as a zero-length result and treats as "gone". let _ = send.finish(); + // CRITICAL: `finish()` only marks the stream's EOF — it does NOT wait + // for the written bytes to be delivered and acknowledged. If we return + // here the `connection` drops, and its CONNECTION_CLOSE can race ahead + // of the still-in-flight stream data, so the fetcher's read aborts with + // "connection lost". Wait for the fetcher to receive everything and + // close the connection itself (it drops `conn` right after read_to_end); + // that close is our signal the transfer landed. Bounded so a fetcher + // that vanishes can't pin this task forever. + let _ = tokio::time::timeout(FILE_FETCH_TIMEOUT, connection.closed()).await; Ok(()) } } diff --git a/tests/file_transfer_loopback.rs b/tests/file_transfer_loopback.rs new file mode 100644 index 0000000..7d3dedd --- /dev/null +++ b/tests/file_transfer_loopback.rs @@ -0,0 +1,141 @@ +//! End-to-end loopback test for the chat file-transfer plane (`FILES_ALPN`). +//! +//! Spins up two real iroh endpoints on localhost (relay disabled, addresses +//! exchanged directly), registers the production [`FileRouter`] on each, serves a +//! multi-megabyte blob on one side, and fetches it from the other through the +//! real `serve_attachment`/`fetch_attachment` path. +//! +//! This is the regression guard for the "file fetch: read failed: connection +//! lost" bug: the serve handler used to return (and drop the connection) the +//! instant it called `finish()`, so the CONNECTION_CLOSE raced ahead of the +//! still-in-flight stream data and the fetcher's `read_to_end` aborted. A blob +//! large enough to span many packets makes that race deterministic — the fix +//! (waiting on `connection.closed()` before returning) keeps the link up until +//! the fetcher has the bytes. + +use std::sync::Arc; +use std::time::Duration; + +use iroh::address_lookup::memory::MemoryLookup; +use iroh::endpoint::presets; +use iroh::protocol::Router; +use iroh::{Endpoint, RelayMode}; + +use peerspeak::files::{ChatAttachment, AttachmentKind}; +use peerspeak::network::NetworkTransport; +use peerspeak::network::iroh_impl::{FileRouter, IrohTransport}; +use peerspeak::protocol::FILES_ALPN; + +struct Node { + endpoint: Endpoint, + transport: Arc, + _router: Router, + lookup: MemoryLookup, +} + +async fn spawn_node() -> Node { + let lookup = MemoryLookup::new(); + let endpoint = Endpoint::builder(presets::Minimal) + .secret_key(iroh::SecretKey::generate()) + .relay_mode(RelayMode::Disabled) + .address_lookup(lookup.clone()) + .bind() + .await + .expect("bind endpoint"); + + let transport = Arc::new(IrohTransport::new(endpoint.clone())); + // Mirror production: a persistent FileRouter bound to this session's transport + // is what the router accepts inbound file fetches on. + let file_router = FileRouter::new(); + file_router.bind(&transport); + let router = Router::builder(endpoint.clone()) + .accept(FILES_ALPN, file_router) + .spawn(); + + Node { endpoint, transport, _router: router, lookup } +} + +/// A pseudo-random-ish multi-megabyte payload spanning many QUIC packets, so a +/// premature connection close on the serve side reliably corrupts/aborts the read. +fn big_blob() -> Vec { + (0..(2 * 1024 * 1024u32)) + .map(|i| (i.wrapping_mul(2654435761) >> 13) as u8) + .collect() +} + +#[tokio::test] +async fn loopback_attachment_round_trips_intact() { + let server = spawn_node().await; + let client = spawn_node().await; + + // Seed each side with the other's full address so direct dialing works. + server.lookup.add_endpoint_info(client.endpoint.addr()); + client.lookup.add_endpoint_info(server.endpoint.addr()); + + let server_id = server.endpoint.id(); + let client_id = client.endpoint.id(); + + // The serve handler gates on room membership (the audio admission roster), so + // the server must admit the client before it will answer the fetch. + server.transport.admit_audio_sender(client_id); + client.transport.admit_audio_sender(server_id); + + // The fetcher dials the retained full address; seed it so fetch_attachment + // doesn't have to fall back to a bare-id lookup. + client.transport.connect_peer(server.endpoint.addr()).await; + + let blob = big_blob(); + let id = [42u8; 32]; + server.transport.serve_attachment(id, Arc::new(blob.clone())); + + let att = ChatAttachment { + name: "exterior-landscape.jpg".to_string(), + size: blob.len() as u64, + kind: AttachmentKind::Image, + id, + }; + + let fetched = tokio::time::timeout( + Duration::from_secs(30), + client.transport.fetch_attachment(server_id, &att), + ) + .await + .expect("fetch did not time out") + .expect("fetch succeeded"); + + assert_eq!(fetched.len(), blob.len(), "fetched the full blob"); + assert_eq!(fetched, blob, "fetched bytes match served bytes exactly"); +} + +#[tokio::test] +async fn loopback_unknown_id_reports_gone() { + let server = spawn_node().await; + let client = spawn_node().await; + + server.lookup.add_endpoint_info(client.endpoint.addr()); + client.lookup.add_endpoint_info(server.endpoint.addr()); + + let server_id = server.endpoint.id(); + let client_id = client.endpoint.id(); + server.transport.admit_audio_sender(client_id); + client.transport.admit_audio_sender(server_id); + client.transport.connect_peer(server.endpoint.addr()).await; + + // Never served — the handler closes with an empty body and the fetcher must + // surface that as an error, not hang or return empty bytes. + let att = ChatAttachment { + name: "missing.bin".to_string(), + size: 4096, + kind: AttachmentKind::File, + id: [7u8; 32], + }; + + let result = tokio::time::timeout( + Duration::from_secs(30), + client.transport.fetch_attachment(server_id, &att), + ) + .await + .expect("fetch did not time out"); + + assert!(result.is_err(), "unknown id should error, got {result:?}"); +}