diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 64f2f54dac..bbcceefc93 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -427,6 +427,7 @@ jobs: run: | cargo test --locked -p openbitfun-services-core --no-default-features --features workspace-runtime --lib workspace::tests:: cargo test --locked -p openbitfun-services-integrations --no-default-features --features remote-ssh-concrete --lib remote_ssh::manager::tests::workspace_ + cargo test --locked -p openbitfun-services-integrations --no-default-features --features remote-connect --lib remote_connect::html_preview cargo test --locked -p tool-runtime --no-default-features --lib fs::read_file::tests:: cargo test --locked -p tool-runtime --no-default-features --test tool_io_contracts diff --git a/src/apps/desktop/AGENTS.md b/src/apps/desktop/AGENTS.md index 2dfc724dfb..d1c058b95b 100644 --- a/src/apps/desktop/AGENTS.md +++ b/src/apps/desktop/AGENTS.md @@ -356,6 +356,15 @@ capability negotiation and encrypted loopback Relay transfer, run: cargo test --locked -p openbitfun-desktop --lib html_preview ``` +For encrypted peer resource-load timings, run the same command with +`--lib html_preview_encrypted_peer_loading_benchmark -- --ignored --nocapture`. +The isolated Relay fixture adds 40 ms per resource RPC and compares the previous +serial reader with cold and warm HTTP gateway loads of the same HTML/assets. +It reports three samples and verifies payload bytes; it does not time native +rendering or use a physical remote host. Cache and prefetch limits belong to +`services-integrations::remote_connect::html_preview::transfer`; the Desktop +adapter owns HTTP authentication, explicit-refresh headers and peer RPC wiring. + Use `node tests/e2e/scripts/run-html-preview.mjs` after building Desktop and Web UI to verify CSS, JavaScript modules, images and concurrent preview cookies in native browser child views. The fixture uses isolated storage; see `tests/e2e/AGENTS.md`. diff --git a/src/apps/desktop/src/api/html_preview_api.rs b/src/apps/desktop/src/api/html_preview_api.rs index b6a94e8edb..3e1c174931 100644 --- a/src/apps/desktop/src/api/html_preview_api.rs +++ b/src/apps/desktop/src/api/html_preview_api.rs @@ -12,18 +12,18 @@ use axum::{ routing::any, Router, }; -use base64::{engine::general_purpose::STANDARD as BASE64, Engine}; +use futures::TryStreamExt; use openbitfun_core::service::filesystem::html_preview::resolve_html_preview_source; use openbitfun_product_domains::{ html_preview::{ HtmlPreviewError, HtmlPreviewErrorCode as Code, HtmlPreviewResource, - HtmlPreviewResourceReply, HtmlPreviewResourceRequest, MAX_PREVIEW_BYTES, - PREVIEW_CHUNK_BYTES, + HtmlPreviewResourceReply, HtmlPreviewResourceRequest, }, remote_surface::PeerHostCapability, }; use openbitfun_services_integrations::remote_connect::{ - html_preview::HtmlPreviewSource, RemoteCommand, RemoteResponse, + html_preview::transfer::{HtmlPreviewReader as PreviewReader, HtmlPreviewTransfer}, + RemoteCommand, RemoteResponse, }; use serde::{Deserialize, Serialize}; use std::sync::Arc; @@ -60,27 +60,6 @@ pub struct HtmlPreviewReleaseRequest { pub session_id: String, } -#[async_trait::async_trait] -trait PreviewReader: Send + Sync { - fn is_current(&self) -> bool { - true - } - async fn read( - &self, - request: &HtmlPreviewResourceRequest, - ) -> Result; -} - -#[async_trait::async_trait] -impl PreviewReader for HtmlPreviewSource { - async fn read( - &self, - request: &HtmlPreviewResourceRequest, - ) -> Result { - HtmlPreviewSource::read(self, request).await - } -} - struct PeerPreviewReader { device_id: String, account_generation: u64, @@ -177,7 +156,7 @@ struct PreviewState { cookie_name: String, entry_path: String, request: HtmlPreviewResourceRequest, - reader: Arc, + transfer: Arc, closed: watch::Receiver, } @@ -266,18 +245,22 @@ async fn start_preview( reader: Arc, mut request: HtmlPreviewResourceRequest, ) -> Result { - let entry = reader.read(&request).await.map_err(|error| error.message)?; - validate_chunk(&entry, &request).map_err(|error| error.message)?; + let (shutdown, closed) = watch::channel(false); + let transfer = Arc::new(HtmlPreviewTransfer::new(reader, closed.clone())); + let entry = transfer + .open(request.clone(), false, None) + .await + .map_err(|error| error.message)? + .metadata; request.expected_source = Some(entry.source); let token = Uuid::new_v4().simple().to_string(); let session_id = Uuid::new_v4().simple().to_string(); - let (shutdown, closed) = watch::channel(false); let state = PreviewState { token: token.clone(), cookie_name: format!("openbitfun_html_preview_{session_id}"), entry_path: entry.entry_path, request, - reader, + transfer, closed: closed.clone(), }; let listener = TcpListener::bind(("127.0.0.1", 0)) @@ -320,68 +303,6 @@ pub async fn html_preview_release(request: HtmlPreviewReleaseRequest) -> Result< Ok(()) } -impl PreviewState { - async fn read( - &self, - request: &HtmlPreviewResourceRequest, - ) -> Result { - let mut closed = self.closed.clone(); - if *closed.borrow() || !self.reader.is_current() { - return Err(HtmlPreviewError::unavailable( - "HTML preview is no longer available", - )); - } - tokio::select! { - _ = closed.wait_for(|closed| *closed) => Err(HtmlPreviewError::unavailable("HTML preview was closed")), - result = self.reader.read(request) => { - let resource = result?; - if *self.closed.borrow() || !self.reader.is_current() { return Err(HtmlPreviewError::unavailable("HTML preview is no longer available")); } - validate_chunk(&resource, request)?; - Ok(resource) - } - } - } -} - -fn validate_chunk( - file: &HtmlPreviewResource, - request: &HtmlPreviewResourceRequest, -) -> Result<(), HtmlPreviewError> { - if file.size > MAX_PREVIEW_BYTES - || file.offset != request.offset - || file.offset > file.size - || file.source.is_empty() - || file.revision.is_empty() - || request - .expected_source - .as_ref() - .is_some_and(|source| source != &file.source) - || request - .expected_revision - .as_ref() - .is_some_and(|revision| revision != &file.revision) - || file.content_base64.len() > PREVIEW_CHUNK_BYTES.div_ceil(3) * 4 - { - return Err(HtmlPreviewError::unavailable( - "Invalid HTML preview resource chunk", - )); - } - let bytes = BASE64 - .decode(&file.content_base64) - .map_err(|_| HtmlPreviewError::unavailable("Invalid HTML preview resource encoding"))?; - let expected = if request.metadata_only { - 0 - } else { - (file.size - file.offset).min(PREVIEW_CHUNK_BYTES as u64) as usize - }; - if bytes.len() != expected { - return Err(HtmlPreviewError::unavailable( - "Incomplete HTML preview resource chunk", - )); - } - Ok(()) -} - async fn preview_handler( State(state): State, method: Method, @@ -404,7 +325,7 @@ async fn preview_handler( .any(|(key, value)| key == state.cookie_name && value == state.token) }); if *state.closed.borrow() - || !state.reader.is_current() + || !state.transfer.is_current() || (method != Method::GET && method != Method::HEAD) { return response(StatusCode::NOT_FOUND, "Not found"); @@ -441,44 +362,27 @@ async fn preview_handler( let mut request = state.request.clone(); request.resource_path = Some(relative); request.metadata_only = method == Method::HEAD; - let first = match state.read(&request).await { - Ok(file) => file, + let bypass_cache = force_resource_reload(&headers); + let resource = match state + .transfer + .open( + request, + bypass_cache, + uri.path_and_query().map(|path| path.as_str().to_string()), + ) + .await + { + Ok(resource) => resource, Err(error) => return preview_error_response(error), }; - let size = first.size; - let content_type = first.mime_type.clone(); - request.expected_revision = Some(first.revision.clone()); - let body = if method == Method::HEAD { - Body::empty() - } else { - let stream = futures::stream::try_unfold( - (state, request, Some(first)), - move |(state, mut request, first)| async move { - let file = match first { - Some(file) => file, - None => state - .read(&request) - .await - .map_err(|error| std::io::Error::other(error.message))?, - }; - if file.offset == file.size { - return Ok::<_, std::io::Error>(None); - } - let bytes = BASE64 - .decode(file.content_base64) - .map_err(std::io::Error::other)?; - request.offset += bytes.len() as u64; - // An empty sentinel ends the stream without another RPC at EOF. - let next = (request.offset == size).then(|| HtmlPreviewResource { - offset: size, - content_base64: String::new(), - ..file - }); - Ok(Some((bytes, (state, request, next)))) - }, - ); - Body::from_stream(stream) - }; + let size = resource.metadata.size; + let content_type = resource.metadata.mime_type; + let body = Body::from_stream( + resource + .body + .map_ok(|bytes| bytes.to_vec()) + .map_err(|error| std::io::Error::other(error.message)), + ); Response::builder() .status(StatusCode::OK) .header(header::CONTENT_TYPE, content_type) @@ -495,6 +399,27 @@ async fn preview_handler( }) } +fn force_resource_reload(headers: &HeaderMap) -> bool { + [header::CACHE_CONTROL, header::PRAGMA].iter().any(|name| { + headers + .get_all(name) + .iter() + .filter_map(|value| value.to_str().ok()) + .any(|value| { + value.split(',').any(|directive| { + let (name, value) = directive + .trim() + .split_once('=') + .unwrap_or((directive.trim(), "")); + name.trim().eq_ignore_ascii_case("no-cache") + || name.trim().eq_ignore_ascii_case("no-store") + || (name.trim().eq_ignore_ascii_case("max-age") + && value.trim().trim_matches('"') == "0") + }) + }) + }) +} + fn preview_error_response(error: HtmlPreviewError) -> Response { let status = match error.code { Code::InvalidPath => StatusCode::BAD_REQUEST, diff --git a/src/apps/desktop/src/api/html_preview_api/tests.rs b/src/apps/desktop/src/api/html_preview_api/tests.rs index e02e538d46..3a549663b9 100644 --- a/src/apps/desktop/src/api/html_preview_api/tests.rs +++ b/src/apps/desktop/src/api/html_preview_api/tests.rs @@ -1,4 +1,6 @@ use super::*; +use base64::{engine::general_purpose::STANDARD as BASE64, Engine}; +use openbitfun_product_domains::html_preview::PREVIEW_CHUNK_BYTES; use std::sync::{ atomic::{AtomicBool, Ordering}, Mutex, @@ -392,3 +394,180 @@ pub(crate) async fn verify_encrypted_peer_gateway( assert_eq!(reply.bytes().await.unwrap().as_ref(), expected); release(preview).await; } + +pub(crate) async fn benchmark_encrypted_peer_gateway( + device_id: &str, + mut input: HtmlPreviewResourceRequest, + files: Vec<(String, Vec)>, +) { + use futures::{stream, StreamExt}; + let reader = Arc::new(PeerPreviewReader { + device_id: device_id.into(), + account_generation: account_context_generation(), + }); + reader.ensure_supported().await.unwrap(); + input.expected_source = Some(reader.read(&input).await.unwrap().source); + let client = reqwest::Client::builder() + .no_proxy() + .redirect(reqwest::redirect::Policy::none()) + .build() + .unwrap(); + let mut serial = Vec::new(); + let mut cold = Vec::new(); + let mut warm = Vec::new(); + for _ in 0..3 { + let start = std::time::Instant::now(); + for group in [&files[..1], &files[1..]] { + stream::iter(group.iter()) + .map(|(path, expected)| { + let reader = reader.clone(); + let mut request = input.clone(); + request.metadata_only = false; + request.resource_path = Some(path.clone()); + async move { + let mut bytes = Vec::new(); + loop { + let file = reader.read(&request).await.unwrap(); + bytes.extend(BASE64.decode(file.content_base64).unwrap()); + request.expected_revision = Some(file.revision); + request.offset = bytes.len() as u64; + if request.offset == file.size { + break; + } + } + assert_eq!(&bytes, expected); + } + }) + .buffer_unordered(6) + .collect::>() + .await; + } + serial.push(start.elapsed().as_millis()); + let preview = start_preview(reader.clone(), input.clone()).await.unwrap(); + let bootstrap = client.get(&preview.url).send().await.unwrap(); + let cookie = bootstrap.headers()[header::SET_COOKIE] + .to_str() + .unwrap() + .split(';') + .next() + .unwrap(); + let origin = reqwest::Url::parse(&preview.url).unwrap(); + for times in [&mut cold, &mut warm] { + let start = std::time::Instant::now(); + for group in [&files[..1], &files[1..]] { + stream::iter(group.iter()) + .map(|(path, expected)| { + let client = client.clone(); + let url = origin.join(&format!("/{path}")).unwrap(); + async move { + let response = client + .get(url) + .header(header::COOKIE, cookie) + .send() + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response.bytes().await.unwrap().as_ref(), expected); + } + }) + .buffer_unordered(6) + .collect::>() + .await; + } + times.push(start.elapsed().as_millis()); + } + release(preview).await; + } + println!("HTML_PREVIEW_BENCH transport=encrypted-peer-40ms bytes={} serial_ms={serial:?} cold_ms={cold:?} warm_ms={warm:?}", files.iter().map(|(_, bytes)| bytes.len()).sum::()); +} + +#[tokio::test] +async fn http_cache_revalidates_and_explicit_reload_bypasses_content_cache() { + let (preview, reader, client) = fixture(vec![21; PREVIEW_CHUNK_BYTES + 3]).await; + let (cookie, entry) = bootstrap(&client, &preview).await; + for control in [ + None, + None, + Some("max-age=0"), + Some("no-cache"), + Some("no-store"), + ] { + let before = reader.requests.lock().unwrap().len(); + let mut request = client.get(&entry).header(header::COOKIE, &cookie); + if let Some(control) = control { + request = request.header(header::CACHE_CONTROL, control); + } + assert_eq!( + request.send().await.unwrap().bytes().await.unwrap().len(), + PREVIEW_CHUNK_BYTES + 3 + ); + let requests = reader.requests.lock().unwrap(); + if before > 1 && control.is_none() { + assert_eq!(requests.len() - before, 1); + assert!(requests.last().unwrap().metadata_only); + } else { + assert_eq!(requests.len() - before, 2); + assert!(!requests.last().unwrap().metadata_only); + } + } + reader.unavailable.store(true, Ordering::SeqCst); + assert_eq!( + client + .get(&entry) + .header(header::COOKIE, &cookie) + .send() + .await + .unwrap() + .status(), + StatusCode::BAD_GATEWAY + ); + assert_eq!( + client.get(&entry).send().await.unwrap().status(), + StatusCode::NOT_FOUND + ); + release(preview).await; +} + +#[test] +fn reload_directives_handle_repeated_headers_and_case() { + for (name, value) in [ + (header::CACHE_CONTROL, "No-Cache"), + (header::CACHE_CONTROL, "max-age = \"0\""), + (header::CACHE_CONTROL, "private, NO-STORE"), + (header::PRAGMA, "no-cache"), + ] { + let mut headers = HeaderMap::new(); + headers.append(header::CACHE_CONTROL, "max-age=60".parse().unwrap()); + headers.append(name, value.parse().unwrap()); + assert!(force_resource_reload(&headers)); + } + let mut headers = HeaderMap::new(); + headers.insert(header::CACHE_CONTROL, "max-age=60".parse().unwrap()); + assert!(!force_resource_reload(&headers)); +} + +#[tokio::test] +async fn http_cache_busting_queries_have_distinct_entries_without_changing_file_paths() { + let (preview, reader, client) = fixture(vec![31; PREVIEW_CHUNK_BYTES + 7]).await; + let (cookie, entry) = bootstrap(&client, &preview).await; + for (query, content_reads) in [("v=1", 2), ("v=1", 1), ("v=2", 2), ("v=2", 1), ("v=1", 1)] { + let before = reader.requests.lock().unwrap().len(); + let response = client + .get(format!("{entry}?{query}")) + .header(header::COOKIE, &cookie) + .send() + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + response.bytes().await.unwrap().len(), + PREVIEW_CHUNK_BYTES + 7 + ); + let requests = reader.requests.lock().unwrap(); + assert_eq!(requests.len() - before, content_reads, "{query}"); + assert!(requests[before..] + .iter() + .all(|request| request.resource_path.as_deref() == Some("site/dist/index #1.html"))); + } + release(preview).await; +} diff --git a/src/apps/desktop/src/api/remote_connect_api/relay_routes_tests.rs b/src/apps/desktop/src/api/remote_connect_api/relay_routes_tests.rs index d1d2508e59..b93d0c69f3 100644 --- a/src/apps/desktop/src/api/remote_connect_api/relay_routes_tests.rs +++ b/src/apps/desktop/src/api/remote_connect_api/relay_routes_tests.rs @@ -466,6 +466,16 @@ async fn dual_relay_rpc_persistence_and_teardown_are_independent() { #[tokio::test] async fn html_preview_streams_assets_over_encrypted_peer_rpc() { + peer_preview_fixture(false).await; +} + +#[tokio::test] +#[ignore = "Performance experiment: encrypted loopback Relay with 40 ms per resource RPC"] +async fn html_preview_encrypted_peer_loading_benchmark() { + peer_preview_fixture(true).await; +} + +async fn peer_preview_fixture(benchmark: bool) { use openbitfun_product_domains::html_preview::{ HtmlPreviewResourceReply, HtmlPreviewResourceRequest, PREVIEW_CHUNK_BYTES, }; @@ -495,12 +505,41 @@ async fn html_preview_streams_assets_over_encrypted_peer_rpc() { "", ) .unwrap(); - let bytes = vec![b'x'; PREVIEW_CHUNK_BYTES * 2 + 19]; + let bytes = vec![ + b'x'; + if benchmark { + 2 * 1024 * 1024 + } else { + PREVIEW_CHUNK_BYTES * 2 + 19 + } + ]; std::fs::write(root.join("site/dist/assets/large.js"), &bytes).unwrap(); let entry = root .join("site/dist/index.html") .to_string_lossy() .into_owned(); + let files = vec![ + ( + "site/dist/index.html".to_string(), + std::fs::read(root.join("site/dist/index.html")).unwrap(), + ), + ( + "site/dist/assets/app.css".to_string(), + vec![b'c'; 96 * 1024], + ), + ("site/dist/assets/large.js".to_string(), bytes.clone()), + ( + "site/dist/assets/image.png".to_string(), + vec![53; 512 * 1024], + ), + ( + "site/dist/assets/font.woff2".to_string(), + vec![19; 128 * 1024], + ), + ]; + for (path, content) in &files { + std::fs::write(root.join(path), content).unwrap(); + } let source = HtmlPreviewSource::new( SessionFileTarget { fs: Arc::new(openbitfun_services_core::workspace::LocalWorkspaceFs), @@ -523,7 +562,10 @@ async fn html_preview_streams_assets_over_encrypted_peer_rpc() { let relay_url = host.url(); let requests = Arc::new(std::sync::atomic::AtomicUsize::new(0)); let serving_requests = requests.clone(); + let source = Arc::new(source); + let target = Arc::new(target); let serving = tokio::spawn(async move { + let mut handlers = tokio::task::JoinSet::new(); while let Some(event) = events.recv().await { let RelayEvent::DeviceMessageReceived { source_device_id, @@ -537,73 +579,94 @@ async fn html_preview_streams_assets_over_encrypted_peer_rpc() { if correlation_id.is_empty() { continue; } - let plaintext = target - .decrypt_from_peer(&relay_url, &source_device_id, &encrypted_data, &nonce) - .await - .unwrap(); - let RemoteCommand::HostInvoke { command, args } = - serde_json::from_str(&plaintext).unwrap() - else { - panic!("unexpected command") - }; - let value = match command.as_str() { - "peer_mode_ping" => { - serde_json::json!({"host_type":"cli", "capabilities":{"workspace_html_preview_v1":true}}) - } - "html_preview_read_resource" => { - let request: HtmlPreviewResourceRequest = - serde_json::from_value(args["request"].clone()).unwrap(); - assert_eq!(request.workspace_id, "fixture-workspace"); - serving_requests.fetch_add(1, std::sync::atomic::Ordering::Relaxed); - serde_json::to_value(HtmlPreviewResourceReply::from( - source.read(&request).await, - )) - .unwrap() - } - _ => panic!("unexpected host invoke {command}"), - }; - let response = RemoteResponse::HostInvokeResult { - ok: true, - value: Some(value), - error: None, - }; - let (data, nonce) = target - .encrypt_for_peer( - &relay_url, - &source_device_id, - &serde_json::to_string(&response).unwrap(), - ) - .await - .unwrap(); - serving_client - .send_device_message(&source_device_id, &correlation_id, &data, &nonce) - .await - .unwrap(); + while let Some(result) = handlers.try_join_next() { + result.unwrap(); + } + let target = target.clone(); + let source = source.clone(); + let relay_url = relay_url.clone(); + let serving_client = serving_client.clone(); + let serving_requests = serving_requests.clone(); + handlers.spawn(async move { + let plaintext = target + .decrypt_from_peer(&relay_url, &source_device_id, &encrypted_data, &nonce) + .await + .unwrap(); + let RemoteCommand::HostInvoke { command, args } = + serde_json::from_str(&plaintext).unwrap() + else { + panic!("unexpected command") + }; + let value = match command.as_str() { + "peer_mode_ping" => { + serde_json::json!({"host_type":"cli", "capabilities":{"workspace_html_preview_v1":true}}) + } + "html_preview_read_resource" => { + let request: HtmlPreviewResourceRequest = + serde_json::from_value(args["request"].clone()).unwrap(); + assert_eq!(request.workspace_id, "fixture-workspace"); + if benchmark { tokio::time::sleep(std::time::Duration::from_millis(40)).await; } + serving_requests.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + serde_json::to_value(HtmlPreviewResourceReply::from( + source.read(&request).await, + )) + .unwrap() + } + _ => panic!("unexpected host invoke {command}"), + }; + let response = RemoteResponse::HostInvokeResult { + ok: true, + value: Some(value), + error: None, + }; + let (data, nonce) = target + .encrypt_for_peer( + &relay_url, + &source_device_id, + &serde_json::to_string(&response).unwrap(), + ) + .await + .unwrap(); + serving_client + .send_device_message(&source_device_id, &correlation_id, &data, &nonce) + .await + .unwrap(); + }); } }); - tokio::time::timeout( - std::time::Duration::from_secs(20), - crate::api::html_preview_api::tests::verify_encrypted_peer_gateway( + let request = HtmlPreviewResourceRequest { + workspace_id: "fixture-workspace".into(), + file_path: entry, + resource_path: None, + metadata_only: true, + offset: 0, + expected_source: None, + expected_revision: None, + }; + if benchmark { + crate::api::html_preview_api::tests::benchmark_encrypted_peer_gateway( "preview-peer", - HtmlPreviewResourceRequest { - workspace_id: "fixture-workspace".into(), - file_path: entry, - resource_path: None, - metadata_only: true, - offset: 0, - expected_source: None, - expected_revision: None, - }, - &bytes, - ), - ) - .await - .unwrap(); - assert_eq!( - requests.load(std::sync::atomic::Ordering::Relaxed), - 4, - "entry metadata and three encrypted chunks" - ); + request, + files, + ) + .await; + } else { + tokio::time::timeout( + std::time::Duration::from_secs(20), + crate::api::html_preview_api::tests::verify_encrypted_peer_gateway( + "preview-peer", + request, + &bytes, + ), + ) + .await + .unwrap(); + assert_eq!( + requests.load(std::sync::atomic::Ordering::Relaxed), + 4, + "entry metadata and three encrypted chunks" + ); + } client.disconnect().await; serving.abort(); host.stop().await; diff --git a/src/crates/services/services-integrations/AGENTS.md b/src/crates/services/services-integrations/AGENTS.md index ee2fe4fb37..6866609a87 100644 --- a/src/crates/services/services-integrations/AGENTS.md +++ b/src/crates/services/services-integrations/AGENTS.md @@ -206,10 +206,11 @@ For models.dev snapshot reuse, last-valid recovery, and background refresh: cargo test --locked -p openbitfun-services-integrations --no-default-features --features models-dev --lib models_dev::tests ``` -For workspace-confined HTML resources and revision-checked chunk reads, run: +For workspace-confined HTML resources, revision-checked chunk reads, bounded +prefetch, cache invalidation and cancellation, run: ```bash -cargo test --locked -p openbitfun-services-integrations --no-default-features --features remote-connect --lib remote_connect::html_preview::tests +cargo test --locked -p openbitfun-services-integrations --no-default-features --features remote-connect --lib remote_connect::html_preview ``` To use a real OpenSSH SFTP executable over loopback SSH, set @@ -218,3 +219,12 @@ To use a real OpenSSH SFTP executable over loopback SSH, set This checks the production remote filesystem provider, root-relative assets, parent-relative resources and workspace boundaries; it does not establish multi-device relay or physical-host coverage. + +For the reproducible resource-loading experiment, use the same OpenSSH fixture +with `--lib openssh_html_preview_loading_benchmark -- --ignored --nocapture`. +It adds 40 ms to each complete SFTP response without serializing replies, loads +HTML followed by four assets (about 2.72 MiB), checks every byte, and reports +three serial-baseline / prefetched-cold / revalidated-warm samples. The baseline +retains the previous serial chunk algorithm with the same filesystem provider; +browser rendering and public-network variability are outside this measurement. +Do not turn timing ratios into CI assertions. diff --git a/src/crates/services/services-integrations/src/remote_connect/html_preview.rs b/src/crates/services/services-integrations/src/remote_connect/html_preview.rs index 8a05a28573..5381cc0b9e 100644 --- a/src/crates/services/services-integrations/src/remote_connect/html_preview.rs +++ b/src/crates/services/services-integrations/src/remote_connect/html_preview.rs @@ -12,6 +12,8 @@ use sha2::{Digest, Sha256}; use std::{path::Path, sync::Arc}; use tokio::io::{AsyncReadExt, AsyncSeekExt}; +pub mod transfer; + pub struct HtmlPreviewSource { fs: Arc, root: String, @@ -190,11 +192,19 @@ impl HtmlPreviewSource { "Symbolic links are not available in remote HTML preview", )); } - if current != requested && metadata.kind != WorkspacePathKind::Directory { + if current == requested { + if metadata.kind != WorkspacePathKind::File { + return Err(not_found()); + } + // Reuse the final no-follow lookup instead of paying for + // another SFTP round trip to the same path. + return Ok((requested, metadata)); + } + if metadata.kind != WorkspacePathKind::Directory { return Err(not_found()); } } - requested + return Err(not_found()); } else { let canonical = tokio::fs::canonicalize(&requested) .await diff --git a/src/crates/services/services-integrations/src/remote_connect/html_preview/tests.rs b/src/crates/services/services-integrations/src/remote_connect/html_preview/tests.rs index be39ffb174..4cb59f25f1 100644 --- a/src/crates/services/services-integrations/src/remote_connect/html_preview/tests.rs +++ b/src/crates/services/services-integrations/src/remote_connect/html_preview/tests.rs @@ -355,3 +355,80 @@ async fn head_skips_content_and_oversized_or_truncated_files_fail() { .size = Some(MAX_PREVIEW_BYTES + 1); assert_eq!(source.read(&input).await.unwrap_err().code, Code::TooLarge); } + +#[tokio::test] +async fn cached_remote_assets_recheck_path_policy_and_owner_availability() { + use super::transfer::HtmlPreviewTransfer; + use futures::TryStreamExt; + let (source, fs) = fixture().await; + let (_close, closed) = tokio::sync::watch::channel(false); + let transfer = Arc::new(HtmlPreviewTransfer::new(Arc::new(source), closed)); + let input = request(Some("assets/app.js")); + transfer + .open(input.clone(), false, None) + .await + .unwrap() + .body + .try_collect::>() + .await + .unwrap(); + let reads = fs.reads.lock().unwrap().len(); + fs.nodes + .lock() + .unwrap() + .get_mut("/repo/site/dist/assets") + .unwrap() + .metadata + .kind = WorkspacePathKind::Symlink; + assert!(matches!( + transfer.open(input.clone(), false, None).await, + Err(HtmlPreviewError { + code: Code::Forbidden, + .. + }) + )); + assert_eq!( + fs.reads.lock().unwrap().len(), + reads, + "A cached body cannot bypass no-follow checks" + ); + fs.nodes + .lock() + .unwrap() + .get_mut("/repo/site/dist/assets") + .unwrap() + .metadata + .kind = WorkspacePathKind::Directory; + transfer + .open(input.clone(), false, None) + .await + .unwrap() + .body + .try_collect::>() + .await + .unwrap(); + fs.nodes + .lock() + .unwrap() + .remove("/repo/site/dist/assets/app.js"); + // The root has another app.js. It is a different resolved path/revision, + // so the former site's cached bytes cannot be reused for that fallback. + let response = transfer.open(input.clone(), false, None).await.unwrap(); + let body = response + .body + .try_fold(Vec::new(), |mut all, bytes| async move { + all.extend_from_slice(&bytes); + Ok(all) + }) + .await + .unwrap(); + assert_eq!(body, b"wrong-project"); + fs.offline.store(true, Ordering::SeqCst); + assert!(matches!( + transfer.open(input, false, None).await, + Err(HtmlPreviewError { + code: Code::Unavailable, + .. + }) + )); +} diff --git a/src/crates/services/services-integrations/src/remote_connect/html_preview/transfer.rs b/src/crates/services/services-integrations/src/remote_connect/html_preview/transfer.rs new file mode 100644 index 0000000000..d0c8f2a57c --- /dev/null +++ b/src/crates/services/services-integrations/src/remote_connect/html_preview/transfer.rs @@ -0,0 +1,377 @@ +//! Controller-side resource transfer shared by SSH and encrypted peer readers. +//! No filesystem or transport fallback is permitted. Cached bytes always need +//! a successful owner-side metadata read before they can be served again. + +use super::HtmlPreviewSource; +use base64::{engine::general_purpose::STANDARD as BASE64, Engine}; +use futures::{stream, stream::BoxStream, StreamExt}; +use openbitfun_product_domains::html_preview::{ + HtmlPreviewError, HtmlPreviewResource, HtmlPreviewResourceRequest, MAX_PREVIEW_BYTES, + PREVIEW_CHUNK_BYTES, +}; +use std::{ + collections::HashMap, + sync::{Arc, Mutex, Weak}, + time::Duration, +}; +use tokio::sync::{watch, OwnedSemaphorePermit, Semaphore}; +use tokio::time::Instant; + +/// Each response can hold at most four completed/in-flight chunks ahead of its +/// consumer. The per-preview limits also cover concurrent browser resource URLs. +const PREFETCH_CHUNKS: usize = 4; +const ACTIVE_RESOURCES: usize = 8; +const ACTIVE_READS: usize = 8; +const CACHE_BYTES: usize = 16 * 1024 * 1024; +const CACHE_RESOURCE_BYTES: u64 = 4 * 1024 * 1024; +const CACHE_ENTRIES: usize = 64; +// SFTP v3 timestamps can have one-second resolution. Never extend this lifetime +// on a metadata hit; explicit browser reloads bypass the cache as well. +const CACHE_LIFETIME: Duration = Duration::from_secs(5); + +#[async_trait::async_trait] +pub trait HtmlPreviewReader: Send + Sync { + fn is_current(&self) -> bool { + true + } + async fn read( + &self, + request: &HtmlPreviewResourceRequest, + ) -> Result; +} + +#[async_trait::async_trait] +impl HtmlPreviewReader for HtmlPreviewSource { + async fn read( + &self, + request: &HtmlPreviewResourceRequest, + ) -> Result { + HtmlPreviewSource::read(self, request).await + } +} + +pub type PreviewBody = BoxStream<'static, Result, HtmlPreviewError>>; + +pub struct PreviewResource { + /// Metadata only; content is delivered by the bounded body stream. + pub metadata: HtmlPreviewResource, + pub body: PreviewBody, +} + +struct CachedResource { + metadata: HtmlPreviewResource, + chunks: Vec>, + created: Instant, +} + +#[derive(Default)] +struct ResourceCache { + entries: HashMap>, + bytes: usize, +} + +impl ResourceCache { + fn remove(&mut self, key: &str) { + if let Some(entry) = self.entries.remove(key) { + self.bytes -= entry.metadata.size as usize; + } + } + + fn get(&mut self, key: &str) -> Option> { + if self.entries.get(key)?.created.elapsed() >= CACHE_LIFETIME { + self.remove(key); + return None; + } + self.entries.get(key).cloned() + } + + fn insert(&mut self, key: String, metadata: HtmlPreviewResource, chunks: Vec>) { + self.remove(&key); + let size = metadata.size as usize; + while self.bytes + size > CACHE_BYTES || self.entries.len() >= CACHE_ENTRIES { + let Some(oldest) = self + .entries + .iter() + .min_by_key(|(_, entry)| entry.created) + .map(|(key, _)| key.clone()) + else { + break; + }; + self.remove(&oldest); + } + self.bytes += size; + self.entries.insert( + key, + Arc::new(CachedResource { + metadata, + chunks, + created: Instant::now(), + }), + ); + } +} + +/// One instance belongs to one preview and immutable reader/account/workspace +/// binding. Closing that preview drops its cache and cancels queued/in-flight IO. +pub struct HtmlPreviewTransfer { + reader: Arc, + closed: watch::Receiver, + resources: Arc, + reads: Arc, + cache: Mutex, + loads: Mutex>>>, +} + +impl HtmlPreviewTransfer { + pub fn new(reader: Arc, closed: watch::Receiver) -> Self { + Self { + reader, + closed, + resources: Arc::new(Semaphore::new(ACTIVE_RESOURCES)), + reads: Arc::new(Semaphore::new(ACTIVE_READS)), + cache: Default::default(), + loads: Default::default(), + } + } + + pub fn is_current(&self) -> bool { + !*self.closed.borrow() && self.reader.is_current() + } + + async fn while_open( + &self, + future: impl std::future::Future>, + ) -> Result { + if !self.is_current() { + return Err(closed_error()); + } + let mut closed = self.closed.clone(); + tokio::select! { + _ = closed.wait_for(|closed| *closed) => Err(closed_error()), + result = future => { + if !self.is_current() { return Err(closed_error()); } + result + } + } + } + + async fn acquire( + &self, + semaphore: &Arc, + ) -> Result { + self.while_open(async { + semaphore + .clone() + .acquire_owned() + .await + .map_err(|_| closed_error()) + }) + .await + } + + async fn read( + &self, + request: &HtmlPreviewResourceRequest, + ) -> Result<(HtmlPreviewResource, Arc<[u8]>), HtmlPreviewError> { + let _permit = self.acquire(&self.reads).await?; + let mut resource = self.while_open(self.reader.read(request)).await?; + let bytes = validate_chunk(&resource, request)?; + resource.content_base64 = String::new(); + Ok((resource, bytes.into())) + } + + pub async fn open( + self: &Arc, + mut request: HtmlPreviewResourceRequest, + bypass_cache: bool, + // HTTP adapters include the query here, never in the filesystem path. + cache_key: Option, + ) -> Result { + let permit = self.acquire(&self.resources).await?; + // Preview HTTP responses always start at offset zero; never reinterpret + // a chunk request as a complete resource response. + if request.offset != 0 { + return Err(HtmlPreviewError::unavailable( + "Invalid preview resource offset", + )); + } + if request.metadata_only { + let (metadata, _) = self.read(&request).await?; + return Ok(PreviewResource { + metadata, + body: stream::empty().boxed(), + }); + } + let key = cache_key.unwrap_or_else(|| request.resource_path.clone().unwrap_or_default()); + let lock = { + let mut loads = self.loads.lock().unwrap(); + loads.retain(|_, lock| lock.strong_count() > 0); + let lock = loads.get(&key).and_then(Weak::upgrade).unwrap_or_default(); + loads.insert(key.clone(), Arc::downgrade(&lock)); + lock + }; + let load_guard = self + .while_open(async { Ok(lock.lock_owned().await) }) + .await?; + let cached = { + let mut cache = self.cache.lock().unwrap(); + if bypass_cache { + cache.remove(&key); + } + cache.get(&key) + }; + if let Some(cached) = cached { + let mut probe = request.clone(); + probe.metadata_only = true; + // A version mismatch is a normal cache miss; offline/forbidden is an + // error and must never serve stale bytes from the controller cache. + let current = self.read(&probe).await; + if let Ok((metadata, _)) = ¤t { + if cached.created.elapsed() < CACHE_LIFETIME + && same_resource(metadata, &cached.metadata) + { + let transfer = self.clone(); + let metadata = metadata.clone(); + let body = + stream::try_unfold( + (cached, 0, permit, load_guard), + move |(cached, index, permit, load_guard)| { + let transfer = transfer.clone(); + async move { + if !transfer.is_current() { + return Err(closed_error()); + } + Ok(cached.chunks.get(index).cloned().map(|bytes| { + (bytes, (cached, index + 1, permit, load_guard)) + })) + } + }, + ) + .boxed(); + return Ok(PreviewResource { metadata, body }); + } + } + self.cache.lock().unwrap().remove(&key); + current?; + } + let (metadata, first) = self.read(&request).await?; + request.expected_source = Some(metadata.source.clone()); + request.expected_revision = Some(metadata.revision.clone()); + let size = metadata.size; + let reader = self.clone(); + let expected = metadata.clone(); + let rest = stream::iter((PREVIEW_CHUNK_BYTES as u64..size).step_by(PREVIEW_CHUNK_BYTES)) + .map(move |offset| { + let reader = reader.clone(); + let expected = expected.clone(); + let mut request = request.clone(); + request.offset = offset; + async move { + let (resource, bytes) = reader.read(&request).await?; + if !same_resource(&resource, &expected) { + return Err(HtmlPreviewError::unavailable( + "HTML preview resource changed during transfer", + )); + } + Ok(bytes) + } + }) + .buffered(PREFETCH_CHUNKS); + let chunks = stream::once(async move { Ok(first) }).chain(rest).boxed(); + let transfer = self.clone(); + let cache_metadata = metadata.clone(); + let collect = (size <= CACHE_RESOURCE_BYTES).then(Vec::new); + let body = stream::try_unfold( + (chunks, collect, 0_u64, permit, load_guard), + move |(mut chunks, mut collect, mut received, permit, load_guard)| { + let transfer = transfer.clone(); + let key = key.clone(); + let metadata = cache_metadata.clone(); + async move { + let next = transfer + .while_open(async { chunks.next().await.transpose() }) + .await?; + let Some(bytes) = next else { + return Ok(None); + }; + received += bytes.len() as u64; + if let Some(collected) = &mut collect { + collected.push(bytes.clone()); + } + // Only complete, validated resources enter the cache. No + // detached prefetch task survives a dropped HTTP response. + if received == size { + if let Some(collected) = collect.take() { + transfer + .cache + .lock() + .unwrap() + .insert(key, metadata, collected); + } + } + Ok(Some(( + bytes, + (chunks, collect, received, permit, load_guard), + ))) + } + }, + ) + .boxed(); + Ok(PreviewResource { metadata, body }) + } +} + +fn same_resource(first: &HtmlPreviewResource, next: &HtmlPreviewResource) -> bool { + first.source == next.source + && first.revision == next.revision + && first.size == next.size + && first.resource_path == next.resource_path + && first.entry_path == next.entry_path + && first.mime_type == next.mime_type +} + +fn closed_error() -> HtmlPreviewError { + HtmlPreviewError::unavailable("HTML preview is no longer available") +} + +fn validate_chunk( + file: &HtmlPreviewResource, + request: &HtmlPreviewResourceRequest, +) -> Result, HtmlPreviewError> { + if file.size > MAX_PREVIEW_BYTES + || file.offset != request.offset + || file.offset > file.size + || file.source.is_empty() + || file.revision.is_empty() + || request + .expected_source + .as_ref() + .is_some_and(|source| source != &file.source) + || request + .expected_revision + .as_ref() + .is_some_and(|revision| revision != &file.revision) + || file.content_base64.len() > PREVIEW_CHUNK_BYTES.div_ceil(3) * 4 + { + return Err(HtmlPreviewError::unavailable( + "Invalid HTML preview resource chunk", + )); + } + let bytes = BASE64 + .decode(&file.content_base64) + .map_err(|_| HtmlPreviewError::unavailable("Invalid HTML preview resource encoding"))?; + let expected = if request.metadata_only { + 0 + } else { + (file.size - file.offset).min(PREVIEW_CHUNK_BYTES as u64) as usize + }; + if bytes.len() != expected { + return Err(HtmlPreviewError::unavailable( + "Incomplete HTML preview resource chunk", + )); + } + Ok(bytes) +} + +#[cfg(test)] +pub(crate) mod tests; diff --git a/src/crates/services/services-integrations/src/remote_connect/html_preview/transfer/tests.rs b/src/crates/services/services-integrations/src/remote_connect/html_preview/transfer/tests.rs new file mode 100644 index 0000000000..6ab7c44834 --- /dev/null +++ b/src/crates/services/services-integrations/src/remote_connect/html_preview/transfer/tests.rs @@ -0,0 +1,455 @@ +use super::*; +use futures::{FutureExt, TryStreamExt}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + +struct Reader { + size: usize, + version: AtomicUsize, + current: AtomicBool, + offline: AtomicBool, + active: AtomicUsize, + peak: AtomicUsize, + requests: Mutex>, + completed: Mutex>, +} + +struct ActiveRead<'a>(&'a AtomicUsize); +impl Drop for ActiveRead<'_> { + fn drop(&mut self) { + self.0.fetch_sub(1, Ordering::SeqCst); + } +} + +#[async_trait::async_trait] +impl HtmlPreviewReader for Reader { + fn is_current(&self) -> bool { + self.current.load(Ordering::SeqCst) + } + async fn read( + &self, + request: &HtmlPreviewResourceRequest, + ) -> Result { + self.requests.lock().unwrap().push(request.clone()); + let active = self.active.fetch_add(1, Ordering::SeqCst) + 1; + self.peak.fetch_max(active, Ordering::SeqCst); + let _guard = ActiveRead(&self.active); + let index = request.offset as usize / PREVIEW_CHUNK_BYTES; + tokio::time::sleep(Duration::from_millis((5 - index % 4) as u64 * 10)).await; + if self.offline.load(Ordering::SeqCst) { + return Err(HtmlPreviewError::unavailable("Owner is offline")); + } + let version = self.version.load(Ordering::SeqCst); + let length = if request.metadata_only { + 0 + } else { + (self.size - request.offset as usize).min(PREVIEW_CHUNK_BYTES) + }; + self.completed.lock().unwrap().push(request.offset); + Ok(HtmlPreviewResource { + entry_path: "index.html".into(), + resource_path: request + .resource_path + .clone() + .unwrap_or_else(|| "index.html".into()), + source: "workspace:provider".into(), + revision: format!("version-{version}"), + mime_type: "application/octet-stream".into(), + size: self.size as u64, + offset: request.offset, + content_base64: BASE64.encode(vec![(index + version) as u8; length]), + }) + } +} + +fn request(path: &str) -> HtmlPreviewResourceRequest { + HtmlPreviewResourceRequest { + workspace_id: "workspace".into(), + file_path: "/repo/index.html".into(), + resource_path: Some(path.into()), + metadata_only: false, + offset: 0, + expected_source: Some("workspace:provider".into()), + expected_revision: None, + } +} + +fn fixture(size: usize) -> (Arc, Arc, watch::Sender) { + let reader = Arc::new(Reader { + size, + version: AtomicUsize::new(0), + current: AtomicBool::new(true), + offline: AtomicBool::new(false), + active: AtomicUsize::new(0), + peak: AtomicUsize::new(0), + requests: Default::default(), + completed: Default::default(), + }); + let (close, closed) = watch::channel(false); + ( + Arc::new(HtmlPreviewTransfer::new(reader.clone(), closed)), + reader, + close, + ) +} + +async fn load( + transfer: &Arc, + path: &str, + bypass: bool, +) -> Result, HtmlPreviewError> { + transfer + .open(request(path), bypass, None) + .await? + .body + .try_fold(Vec::new(), |mut all, bytes| async move { + all.extend_from_slice(&bytes); + Ok(all) + }) + .await +} + +#[tokio::test(start_paused = true)] +async fn out_of_order_prefetch_preserves_bytes_and_stays_bounded() { + let size = PREVIEW_CHUNK_BYTES * 10 + 7; + let (transfer, reader, _close) = fixture(size); + let bytes = load(&transfer, "app.js", false).await.unwrap(); + assert_eq!(bytes.len(), size); + for (index, chunk) in bytes.chunks(PREVIEW_CHUNK_BYTES).enumerate() { + assert!(chunk.iter().all(|byte| *byte == index as u8)); + } + assert_eq!(reader.peak.load(Ordering::SeqCst), PREFETCH_CHUNKS); + assert_eq!(reader.active.load(Ordering::SeqCst), 0); + assert!(reader + .completed + .lock() + .unwrap() + .windows(2) + .any(|pair| pair[0] > pair[1])); + assert_eq!(reader.requests.lock().unwrap().len(), 11); +} + +#[tokio::test(start_paused = true)] +async fn concurrent_resources_share_read_budget_and_duplicate_loads_coalesce() { + let (transfer, reader, _close) = fixture(PREVIEW_CHUNK_BYTES * 6); + let loads = (0..20).map(|index| { + let transfer = transfer.clone(); + async move { + load(&transfer, &format!("asset-{}.js", index % 5), false) + .await + .unwrap() + } + }); + for bytes in futures::future::join_all(loads).await { + assert_eq!(bytes.len(), reader.size); + } + assert_eq!(reader.peak.load(Ordering::SeqCst), ACTIVE_READS); + let requests = reader.requests.lock().unwrap(); + assert_eq!( + requests + .iter() + .filter(|request| !request.metadata_only) + .count(), + 5 * 6 + ); + assert_eq!( + requests + .iter() + .filter(|request| request.metadata_only) + .count(), + 15 + ); + assert!(transfer.cache.lock().unwrap().bytes <= CACHE_BYTES); +} + +#[tokio::test(start_paused = true)] +async fn cached_bytes_require_live_metadata_and_changed_revisions_reload() { + let (transfer, reader, _close) = fixture(PREVIEW_CHUNK_BYTES * 2 + 3); + let first = load(&transfer, "app.js", false).await.unwrap(); + let count = reader.requests.lock().unwrap().len(); + assert_eq!(load(&transfer, "app.js", false).await.unwrap(), first); + assert_eq!(reader.requests.lock().unwrap().len(), count + 1); + assert!( + reader + .requests + .lock() + .unwrap() + .last() + .unwrap() + .metadata_only + ); + reader.version.store(1, Ordering::SeqCst); + assert_ne!(load(&transfer, "app.js", false).await.unwrap(), first); + reader.offline.store(true, Ordering::SeqCst); + assert!(load(&transfer, "app.js", false) + .await + .unwrap_err() + .message + .contains("offline")); + assert!(transfer.cache.lock().unwrap().entries.is_empty()); + reader.offline.store(false, Ordering::SeqCst); + assert!(load(&transfer, "app.js", false).await.is_ok()); + reader.current.store(false, Ordering::SeqCst); + assert!(load(&transfer, "app.js", false).await.is_err()); +} + +#[tokio::test(start_paused = true)] +async fn refresh_and_fixed_expiry_force_content_reads_and_head_never_reads_content() { + let (transfer, reader, _close) = fixture(17); + load(&transfer, "app.js", false).await.unwrap(); + load(&transfer, "app.js", true).await.unwrap(); + assert!( + !reader + .requests + .lock() + .unwrap() + .last() + .unwrap() + .metadata_only + ); + tokio::time::advance(CACHE_LIFETIME / 2).await; + load(&transfer, "app.js", false).await.unwrap(); + assert!( + reader + .requests + .lock() + .unwrap() + .last() + .unwrap() + .metadata_only + ); + tokio::time::advance(CACHE_LIFETIME / 2).await; + load(&transfer, "app.js", false).await.unwrap(); + assert!( + !reader + .requests + .lock() + .unwrap() + .last() + .unwrap() + .metadata_only + ); + let mut head = request("app.js"); + head.metadata_only = true; + let response = transfer.open(head, false, None).await.unwrap(); + assert_eq!(response.metadata.size, 17); + assert_eq!(response.body.count().await, 0); + assert!( + reader + .requests + .lock() + .unwrap() + .last() + .unwrap() + .metadata_only + ); +} + +#[tokio::test(start_paused = true)] +async fn cache_expiry_during_slow_revalidation_forces_a_fresh_body() { + let (transfer, reader, _close) = fixture(17); + load(&transfer, "app.js", false).await.unwrap(); + tokio::time::advance(CACHE_LIFETIME - Duration::from_millis(25)).await; + load(&transfer, "app.js", false).await.unwrap(); + let requests = reader.requests.lock().unwrap(); + assert_eq!( + requests.len(), + 3, + "Initial body, delayed metadata, then a fresh body" + ); + assert!(requests[1].metadata_only); + assert!(!requests[2].metadata_only); +} + +#[tokio::test(start_paused = true)] +async fn slow_consumers_and_dropped_bodies_do_not_leave_prefetch_running() { + let (transfer, reader, _close) = fixture(PREVIEW_CHUNK_BYTES * 20); + let mut response = transfer + .open(request("large.js"), false, None) + .await + .unwrap(); + response.body.next().await.unwrap().unwrap(); + // Poll once to launch exactly one window, then stop consuming. + assert!(response.body.next().now_or_never().is_none()); + assert_eq!(reader.active.load(Ordering::SeqCst), PREFETCH_CHUNKS); + tokio::time::advance(Duration::from_secs(1)).await; + assert_eq!(reader.requests.lock().unwrap().len(), 1 + PREFETCH_CHUNKS); + drop(response); + assert_eq!(reader.active.load(Ordering::SeqCst), 0); + assert!(transfer.cache.lock().unwrap().entries.is_empty()); + assert_eq!(transfer.resources.available_permits(), ACTIVE_RESOURCES); + assert_eq!(transfer.reads.available_permits(), ACTIVE_READS); +} + +#[tokio::test(start_paused = true)] +async fn closing_preview_cancels_waiters_and_active_reads() { + let (transfer, reader, close) = fixture(PREVIEW_CHUNK_BYTES * 20); + let mut response = transfer + .open(request("large.js"), false, None) + .await + .unwrap(); + response.body.next().await.unwrap().unwrap(); + assert!(response.body.next().now_or_never().is_none()); + close.send(true).unwrap(); + assert!(response.body.next().await.unwrap().is_err()); + drop(response); + assert_eq!(reader.active.load(Ordering::SeqCst), 0); + assert!(load(&transfer, "another.js", false).await.is_err()); +} + +#[tokio::test(start_paused = true)] +async fn closing_preview_releases_queued_resource_admission() { + let (transfer, reader, close) = fixture(17); + let mut responses = Vec::new(); + for index in 0..ACTIVE_RESOURCES { + responses.push( + transfer + .open(request(&index.to_string()), false, None) + .await + .unwrap(), + ); + } + assert_eq!(transfer.resources.available_permits(), 0); + let mut waiting = Box::pin(transfer.open(request("queued.js"), false, None)); + assert!(waiting.as_mut().now_or_never().is_none()); + assert_eq!(reader.requests.lock().unwrap().len(), ACTIVE_RESOURCES); + close.send(true).unwrap(); + assert!(waiting.await.is_err()); + for mut response in responses { + assert!(response.body.next().await.unwrap().is_err()); + } + assert_eq!(transfer.resources.available_permits(), ACTIVE_RESOURCES); +} + +#[tokio::test(start_paused = true)] +async fn account_change_during_cache_revalidation_rejects_late_bytes() { + let (transfer, reader, _close) = fixture(17); + load(&transfer, "app.js", false).await.unwrap(); + let mut loading = Box::pin(load(&transfer, "app.js", false)); + assert!(loading.as_mut().now_or_never().is_none()); + reader.current.store(false, Ordering::SeqCst); + assert!(loading.await.is_err()); + assert!(transfer.cache.lock().unwrap().entries.is_empty()); +} + +#[tokio::test(start_paused = true)] +async fn partial_or_changed_transfers_never_enter_cache() { + let (transfer, reader, _close) = fixture(PREVIEW_CHUNK_BYTES * 3); + let mut response = transfer.open(request("app.js"), false, None).await.unwrap(); + response.body.next().await.unwrap().unwrap(); + reader.version.store(1, Ordering::SeqCst); + assert!(response.body.next().await.unwrap().is_err()); + assert!(transfer.cache.lock().unwrap().entries.is_empty()); +} + +#[tokio::test(start_paused = true)] +async fn cache_evicts_by_bytes_and_entry_count_and_skips_large_resources() { + let (transfer, _, _close) = fixture(CACHE_RESOURCE_BYTES as usize); + for index in 0..5 { + load(&transfer, &index.to_string(), false).await.unwrap(); + } + { + let cache = transfer.cache.lock().unwrap(); + assert_eq!(cache.bytes, CACHE_BYTES); + assert_eq!(cache.entries.len(), 4); + assert!(!cache.entries.contains_key("0")); + } + let (transfer, _, _close) = fixture(0); + for index in 0..65 { + load(&transfer, &index.to_string(), false).await.unwrap(); + } + assert_eq!(transfer.cache.lock().unwrap().entries.len(), CACHE_ENTRIES); + let (transfer, _, _close) = fixture(CACHE_RESOURCE_BYTES as usize + 1); + load(&transfer, "large.js", false).await.unwrap(); + assert!(transfer.cache.lock().unwrap().entries.is_empty()); +} + +#[tokio::test(start_paused = true)] +async fn previews_with_identical_paths_do_not_share_cache() { + let (a, reader_a, _close_a) = fixture(4); + let (b, reader_b, _close_b) = fixture(4); + reader_b.version.store(2, Ordering::SeqCst); + assert_ne!( + load(&a, "app.js", false).await.unwrap(), + load(&b, "app.js", false).await.unwrap() + ); + assert_eq!(reader_a.requests.lock().unwrap().len(), 1); + assert_eq!(reader_b.requests.lock().unwrap().len(), 1); +} + +/// Same resource order and browser-style concurrency for the previous serial +/// chunk loop and the production transfer service. Timing assertions belong in +/// reports, not CI; byte equality and the production IO boundary are mandatory. +#[cfg(feature = "remote-ssh-concrete")] +pub(crate) async fn benchmark_loading( + reader: Arc, + mut input: HtmlPreviewResourceRequest, + files: Vec<(String, Vec)>, + label: &str, +) { + input.metadata_only = true; + input.expected_source = Some(reader.read(&input).await.unwrap().source); + input.metadata_only = false; + let mut serial = Vec::new(); + let mut cold = Vec::new(); + let mut warm = Vec::new(); + for _ in 0..3 { + let start = Instant::now(); + for group in [&files[..1], &files[1..]] { + stream::iter(group.iter()) + .map(|(path, expected)| { + let reader = reader.clone(); + let mut request = input.clone(); + request.resource_path = Some(path.clone()); + async move { + let mut bytes = Vec::new(); + loop { + let file = reader.read(&request).await.unwrap(); + bytes.extend(BASE64.decode(&file.content_base64).unwrap()); + request.expected_revision = Some(file.revision); + request.offset = bytes.len() as u64; + if request.offset == file.size { + break; + } + } + assert_eq!(&bytes, expected); + } + }) + .buffer_unordered(6) + .collect::>() + .await; + } + serial.push(start.elapsed().as_millis()); + let (_close, closed) = watch::channel(false); + let transfer = Arc::new(HtmlPreviewTransfer::new(reader.clone(), closed)); + for times in [&mut cold, &mut warm] { + let start = Instant::now(); + for group in [&files[..1], &files[1..]] { + stream::iter(group.iter()) + .map(|(path, expected)| { + let transfer = transfer.clone(); + let mut request = input.clone(); + request.resource_path = Some(path.clone()); + async move { + let bytes = transfer + .open(request, false, None) + .await + .unwrap() + .body + .try_fold(Vec::new(), |mut all, bytes| async move { + all.extend_from_slice(&bytes); + Ok(all) + }) + .await + .unwrap(); + assert_eq!(&bytes, expected); + } + }) + .buffer_unordered(6) + .collect::>() + .await; + } + times.push(start.elapsed().as_millis()); + } + } + println!("HTML_PREVIEW_BENCH transport={label} bytes={} serial_ms={serial:?} cold_ms={cold:?} warm_ms={warm:?}", files.iter().map(|(_, bytes)| bytes.len()).sum::()); +} diff --git a/src/crates/services/services-integrations/src/remote_ssh/manager/tests/workspace_sftp.rs b/src/crates/services/services-integrations/src/remote_ssh/manager/tests/workspace_sftp.rs index 90c0ff6570..8b4cec0dca 100644 --- a/src/crates/services/services-integrations/src/remote_ssh/manager/tests/workspace_sftp.rs +++ b/src/crates/services/services-integrations/src/remote_ssh/manager/tests/workspace_sftp.rs @@ -24,6 +24,7 @@ struct State { fail_close: AtomicBool, requested: tokio::sync::Notify, release: tokio::sync::Notify, + response_delay_ms: std::sync::atomic::AtomicU64, } struct SftpServer { @@ -249,6 +250,7 @@ impl russh::server::Handler for SshServer { let mut stdin = child.stdin.take().unwrap(); let mut stdout = child.stdout.take().unwrap(); let (mut reader, mut writer) = tokio::io::split(channel.into_stream()); + let state = self.state.clone(); tokio::spawn(async move { use tokio::io::AsyncWriteExt; let input = async { @@ -256,8 +258,40 @@ impl russh::server::Handler for SshServer { stdin.shutdown().await }; let output = async { - tokio::io::copy(&mut stdout, &mut writer).await?; - writer.shutdown().await + // Delay complete SFTP response packets independently. A + // sleeping copy loop would serialize unrelated replies and + // misrepresent a network that supports pipelined requests. + let (send, mut receive) = tokio::sync::mpsc::channel(32); + let read = async move { + loop { + let size = match stdout.read_u32().await { + Ok(size) => size, + Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => { + break + } + Err(error) => return Err(error), + }; + let mut packet = vec![0; size as usize + 4]; + packet[..4].copy_from_slice(&size.to_be_bytes()); + stdout.read_exact(&mut packet[4..]).await?; + let ready = tokio::time::Instant::now() + + std::time::Duration::from_millis( + state.response_delay_ms.load(Ordering::SeqCst), + ); + if send.send((ready, packet)).await.is_err() { + break; + } + } + Ok::<_, std::io::Error>(()) + }; + let write = async { + while let Some((ready, packet)) = receive.recv().await { + tokio::time::sleep_until(ready).await; + writer.write_all(&packet).await?; + } + writer.shutdown().await + }; + tokio::try_join!(read, write).map(|_| ()) }; tokio::select! { _ = async { tokio::try_join!(input, output) } => {}, @@ -877,6 +911,64 @@ async fn openssh_html_preview_reads_nested_site_resources() { ); } +#[cfg(feature = "remote-connect")] +#[tokio::test] +#[ignore = "Performance experiment: set OPENBITFUN_TEST_SFTP_SERVER; adds 40 ms to each SFTP reply"] +async fn openssh_html_preview_loading_benchmark() { + use crate::remote_connect::{ + file_projection::SessionFileTarget, + html_preview::{transfer::tests::benchmark_loading, HtmlPreviewSource}, + }; + use crate::remote_ssh::workspace_services::RemoteWorkspaceFs; + use openbitfun_product_domains::html_preview::HtmlPreviewResourceRequest; + let executable = std::env::var("OPENBITFUN_TEST_SFTP_SERVER").unwrap(); + let f = Fixture::with_openssh(Some(executable)).await; + f.state.response_delay_ms.store(40, Ordering::SeqCst); + let root = f._dir.path().canonicalize().unwrap(); + std::fs::create_dir_all(root.join("site/assets")).unwrap(); + let files = vec![ + ("site/index.html".to_string(), b"".to_vec()), + ("site/assets/app.css".to_string(), vec![b'c'; 96 * 1024]), + ("site/assets/app.js".to_string(), vec![b'j'; 2 * 1024 * 1024]), + ("site/assets/image.png".to_string(), vec![53; 512 * 1024]), + ("site/assets/font.woff2".to_string(), vec![19; 128 * 1024]), + ]; + for (path, bytes) in &files { + std::fs::write(root.join(path), bytes).unwrap(); + } + let service = crate::remote_ssh::RemoteFileService::new(Arc::new(tokio::sync::RwLock::new( + Some(f.manager.as_ref().clone()), + ))); + let entry = root.join("site/index.html").to_string_lossy().into_owned(); + let source = HtmlPreviewSource::new( + SessionFileTarget { + fs: Arc::new(RemoteWorkspaceFs::new("sftp-test".into(), service)), + root: root.to_string_lossy().into_owned(), + path: entry.clone(), + remote: true, + }, + "ssh-workspace", + ) + .await + .unwrap(); + benchmark_loading( + Arc::new(source), + HtmlPreviewResourceRequest { + workspace_id: "ssh-workspace".into(), + file_path: entry, + resource_path: None, + metadata_only: false, + offset: 0, + expected_source: None, + expected_revision: None, + }, + files, + "ssh-sftp-40ms", + ) + .await; + assert!(f.state.channels.load(Ordering::SeqCst) > 0); +} + #[cfg(unix)] #[tokio::test] #[ignore = "Set OPENBITFUN_TEST_SFTP_SERVER to the OpenSSH sftp-server executable"] diff --git a/src/web-ui/src/app/scenes/browser/README.md b/src/web-ui/src/app/scenes/browser/README.md index 424925479d..20beba29c6 100644 --- a/src/web-ui/src/app/scenes/browser/README.md +++ b/src/web-ui/src/app/scenes/browser/README.md @@ -128,6 +128,21 @@ the selected HTML, then under the workspace root. This supports a nested `dist` without breaking workspace-prefixed links. Resources stay inside the workspace; remote symlinks are rejected. The service streams files up to 64 MiB in 256 KiB chunks and checks workspace identity and file revision throughout a transfer. +The controller prefetches up to four chunks per resource, preserves byte order, +and limits each preview to eight active resources and eight remote reads. Dropped +responses cancel their pending prefetches. SSH and peer resources use the same +transfer service without changing the `workspace_html_preview_v1` wire contract. + +Each preview retains at most 64 complete resources / 16 MiB in memory, with a +4 MiB per-resource limit. Cache hits first revalidate the selected owner's +source, resolved path, size, revision and MIME type; missing, forbidden or offline +resources return errors instead of cached content. Cache entries expire five +seconds after insertion even if revalidated. Keys include URL queries so version +parameters bust the cache without becoming part of the remote filesystem path. +The fixed lifetime limits stale reuse on filesystems +with coarse timestamps. Explicit browser reload directives (`no-cache`, +`no-store`, `max-age=0`) force new content reads. The HTTP response remains +`no-store`: no persistent browser cache or cross-preview content store is added. Each preview has a distinct authorization cookie and is released when its panel closes or its device activation changes. A disconnect returns a resource error; reloading after reconnect retries the same workspace.