From a6e686f20a260d5570e8157e25048156788279db Mon Sep 17 00:00:00 2001 From: rorychatt Date: Mon, 10 Aug 2026 11:15:53 +0200 Subject: [PATCH 1/6] [00140] Enable axum's multipart feature for server side uploads The Multipart extractor the upload endpoint needs is behind a feature flag. Workspace-wide because rusty-server and rusty-desktop share the dependency. --- Cargo.lock | 33 +++++++++++++++++++++++++++++++++ Cargo.toml | 2 +- 2 files changed, 34 insertions(+), 1 deletion(-) diff --git a/Cargo.lock b/Cargo.lock index 5413dea..f77624e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -119,6 +119,7 @@ dependencies = [ "matchit", "memchr", "mime", + "multer", "percent-encoding", "pin-project-lite", "serde_core", @@ -605,6 +606,15 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" +[[package]] +name = "encoding_rs" +version = "0.8.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75030f3c4f45dafd7586dd6780965a8c7e8e285a5ecb86713e63a79c5b2766f3" +dependencies = [ + "cfg-if", +] + [[package]] name = "equivalent" version = "1.0.2" @@ -1592,6 +1602,23 @@ dependencies = [ "windows-sys 0.60.2", ] +[[package]] +name = "multer" +version = "3.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "83e87776546dc87511aa5ee218730c92b666d7264ab6ed41f9d215af9cd5224b" +dependencies = [ + "bytes", + "encoding_rs", + "futures-util", + "http", + "httparse", + "memchr", + "mime", + "spin", + "version_check", +] + [[package]] name = "ndk" version = "0.9.0" @@ -2433,6 +2460,12 @@ dependencies = [ "system-deps", ] +[[package]] +name = "spin" +version = "0.9.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3763264f6b73151db08c50ff20d7d8a0b8796e021cdea7ceedad07b80155fa0e" + [[package]] name = "stable_deref_trait" version = "1.2.1" diff --git a/Cargo.toml b/Cargo.toml index 7ee0431..d734014 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,7 +12,7 @@ repository = "https://github.com/Ivy-Interactive/Rusty-Framework" serde = { version = "1", features = ["derive"] } serde_json = "1" tokio = { version = "1", features = ["full"] } -axum = { version = "0.8", features = ["ws"] } +axum = { version = "0.8", features = ["ws", "multipart"] } futures = "0.3" uuid = { version = "1", features = ["v4"] } tracing = "0.1" From f0988dfc7eb5b17abd48ad231feba39e76f617f9 Mon Sep 17 00:00:00 2001 From: rorychatt Date: Mon, 10 Aug 2026 11:15:58 +0200 Subject: [PATCH 2/6] [00140] Add the UploadService registry and register it per connection Slots are keyed by connection so an upload URL only resolves for the session that created it, and dropping the handle a view holds unregisters the slot. The cancel flag is shared with the caller so a view can cancel a body that is already arriving, before the handle exists. --- rusty/src/server/mod.rs | 1 + rusty/src/server/session.rs | 12 + rusty/src/server/upload.rs | 692 ++++++++++++++++++++++++++++++++++++ 3 files changed, 705 insertions(+) create mode 100644 rusty/src/server/upload.rs diff --git a/rusty/src/server/mod.rs b/rusty/src/server/mod.rs index 20bceb6..c7a8e3a 100644 --- a/rusty/src/server/mod.rs +++ b/rusty/src/server/mod.rs @@ -1,5 +1,6 @@ pub mod download; pub mod session; +pub mod upload; pub mod ws; pub use ws::{RustyServer, DEFAULT_BIND_ADDRESS}; diff --git a/rusty/src/server/session.rs b/rusty/src/server/session.rs index 8527758..cda3ab6 100644 --- a/rusty/src/server/session.rs +++ b/rusty/src/server/session.rs @@ -11,6 +11,7 @@ use crate::core::signals::{ServerSignals, SignalRegistry}; use crate::views::view::View; use super::download::DownloadService; +use super::upload::UploadService; use super::ws::FuncView; /// Per-connection state holding an isolated Runtime and Reconciler. @@ -108,6 +109,7 @@ impl AppSessionStore { // Per-connection. services.register(Arc::new(SignalRegistry::new())); services.register(Arc::new(DownloadService::new(connection_id.to_string()))); + services.register(Arc::new(UploadService::new(connection_id.to_string()))); let (resolved_id, view) = match self.apps.resolve(app_id) { Some(descriptor) => (descriptor.id.clone(), descriptor.create_view()), @@ -386,6 +388,7 @@ mod tests { assert!(services.get::().is_some()); assert!(services.get::().is_some()); assert!(services.get::().is_some()); + assert!(services.get::().is_some()); // The runtime hands the same registry to every BuildContext it creates. let runtime_services = session_arc.read().await.runtime.services().clone(); @@ -436,6 +439,15 @@ mod tests { assert_eq!(downloads_a.connection_id(), "conn-a"); assert_eq!(downloads_b.connection_id(), "conn-b"); assert!(!Arc::ptr_eq(&downloads_a, &downloads_b)); + + // Uploads are scoped the same way: a POST URL minted for one connection + // must not resolve against another's registry. + let uploads_a = a.read().await.services.get::().unwrap(); + let uploads_b = b.read().await.services.get::().unwrap(); + + assert_eq!(uploads_a.connection_id(), "conn-a"); + assert_eq!(uploads_b.connection_id(), "conn-b"); + assert!(!Arc::ptr_eq(&uploads_a, &uploads_b)); } #[tokio::test] diff --git a/rusty/src/server/upload.rs b/rusty/src/server/upload.rs new file mode 100644 index 0000000..e82f35f --- /dev/null +++ b/rusty/src/server/upload.rs @@ -0,0 +1,692 @@ +use std::collections::HashMap; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, Weak}; + +use axum::http::StatusCode; +use bytes::Bytes; +use uuid::Uuid; + +/// Default cap on an upload request body, applied to the upload route only. +/// +/// axum's own `DefaultBodyLimit` is 2 MiB, which would reject nearly every real +/// upload with an opaque 413 that no [`UploadError`] explains. Override with +/// [`RustyServer::with_max_upload_bytes`](crate::server::RustyServer::with_max_upload_bytes). +pub const DEFAULT_MAX_UPLOAD_BYTES: u64 = 32 * 1024 * 1024; + +/// How much larger than the file a multipart request body is allowed to be when +/// the endpoint rejects an oversize upload from `Content-Length` alone. +/// +/// `Content-Length` covers the boundary lines, the part headers and the file name +/// as well as the bytes, so it is only ever an *upper* bound on the file's own +/// size. Rejecting on `Content-Length > max_bytes` would therefore turn a file of +/// exactly `max_bytes` into a 413. The allowance keeps the early rejection for the +/// case it exists for — a body far too big to be worth reading — while the exact +/// limit is still enforced chunk by chunk as the body arrives. +pub const MULTIPART_ENVELOPE_ALLOWANCE: u64 = 8 * 1024; + +/// What an upload slot will accept. Every field is optional — the default +/// constraints accept any file of any size. +/// +/// The browser also enforces `accept` through the file picker's own filter, but a +/// client can POST whatever it likes, so the server checks again. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct UploadConstraints { + /// Accepted MIME types and extensions: `"image/*"`, `"application/pdf"`, + /// `".csv"`. Empty accepts everything. + pub accept: Vec, + pub max_bytes: Option, + pub min_bytes: Option, +} + +impl UploadConstraints { + pub fn new() -> Self { + UploadConstraints::default() + } + + /// Restrict the accepted types. Each pattern is a MIME type (`"image/png"`), a + /// MIME wildcard (`"image/*"`) or a file extension (`".csv"`). + pub fn accept(mut self, patterns: I) -> Self + where + I: IntoIterator, + S: Into, + { + self.accept = patterns.into_iter().map(Into::into).collect(); + self + } + + pub fn max_bytes(mut self, max_bytes: u64) -> Self { + self.max_bytes = Some(max_bytes); + self + } + + pub fn min_bytes(mut self, min_bytes: u64) -> Self { + self.min_bytes = Some(min_bytes); + self + } + + /// Whether a file with this MIME type and name passes the `accept` list. + pub fn accepts(&self, mime_type: &str, file_name: &str) -> bool { + accepts(&self.accept, mime_type, file_name) + } +} + +/// One file that arrived over the upload endpoint, held in memory. +#[derive(Clone, PartialEq, Eq)] +pub struct UploadedFile { + pub file_name: String, + pub mime_type: String, + pub content: Bytes, +} + +impl UploadedFile { + pub fn len(&self) -> usize { + self.content.len() + } + + pub fn is_empty(&self) -> bool { + self.content.is_empty() + } +} + +/// Prints the size instead of the bytes: an upload is routinely megabytes, and a +/// `{:?}` in a log line should not dump them. +impl std::fmt::Debug for UploadedFile { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("UploadedFile") + .field("file_name", &self.file_name) + .field("mime_type", &self.mime_type) + .field("len", &self.content.len()) + .finish() + } +} + +/// Why an upload did not produce a file. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum UploadError { + /// The request carried no `file` field. + NoFile, + TooLarge { + limit: u64, + actual: u64, + }, + TooSmall { + limit: u64, + actual: u64, + }, + RejectedMimeType { + mime_type: String, + accept: Vec, + }, + /// The view called `Upload::cancel` while the body was still arriving. + Cancelled, + /// The connection broke or the multipart body was malformed. + Transport(String), +} + +impl UploadError { + /// The status code the endpoint answers with. + /// + /// All of these are 4xx: every variant is something about the *request*, even + /// `Transport`, which here only ever means a body the server could not parse + /// or finish reading. + pub fn status_code(&self) -> StatusCode { + match self { + UploadError::TooLarge { .. } => StatusCode::PAYLOAD_TOO_LARGE, + UploadError::TooSmall { .. } | UploadError::RejectedMimeType { .. } => { + StatusCode::UNSUPPORTED_MEDIA_TYPE + } + UploadError::NoFile | UploadError::Cancelled | UploadError::Transport(_) => { + StatusCode::BAD_REQUEST + } + } + } +} + +impl std::fmt::Display for UploadError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + UploadError::NoFile => write!(f, "the request contained no file field"), + UploadError::TooLarge { limit, actual } => { + write!(f, "file is {actual} bytes, over the {limit} byte limit") + } + UploadError::TooSmall { limit, actual } => { + write!(f, "file is {actual} bytes, under the {limit} byte minimum") + } + UploadError::RejectedMimeType { mime_type, accept } => { + write!(f, "{mime_type} is not one of {}", accept.join(", ")) + } + UploadError::Cancelled => write!(f, "the upload was cancelled"), + UploadError::Transport(message) => write!(f, "{message}"), + } + } +} + +impl std::error::Error for UploadError {} + +/// What the endpoint reports back to the view as a body arrives. +#[derive(Debug, Clone)] +pub enum UploadEvent { + /// `total` is the request's `Content-Length` when it sent one, so it includes + /// the multipart envelope and is an upper bound on the file's own size. + Progress { + received: u64, + total: Option, + }, + Completed(UploadedFile), + Failed(UploadError), +} + +/// Receives every [`UploadEvent`] for one slot. Called from the HTTP task, not +/// from a build. +pub type UploadObserver = Arc; + +/// One registered upload slot. +struct UploadEntry { + observer: UploadObserver, + constraints: UploadConstraints, + cancel: Arc, +} + +/// A slot resolved for one in-flight request. +/// +/// [`UploadService::slot`] clones this out from under the registry lock, so the +/// request never holds the lock while reading a body. +#[derive(Clone)] +pub struct UploadSlot { + pub constraints: UploadConstraints, + observer: UploadObserver, + cancel: Arc, +} + +impl UploadSlot { + /// Report an event to the view that registered the slot. + pub fn emit(&self, event: UploadEvent) { + (self.observer)(event); + } + + /// Whether the view has asked for the in-flight upload to stop. + pub fn is_cancelled(&self) -> bool { + self.cancel.load(Ordering::SeqCst) + } +} + +impl std::fmt::Debug for UploadSlot { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("UploadSlot") + .field("constraints", &self.constraints) + .field("cancelled", &self.is_cancelled()) + .finish() + } +} + +/// Per-connection registry of upload slots reachable over HTTP. +/// +/// The mirror image of [`DownloadService`](super::download::DownloadService): +/// slots are scoped to a connection so one session's upload URLs are not +/// guessable from another, and a slot lives exactly as long as its handle. +pub struct UploadService { + connection_id: String, + entries: Mutex>, +} + +/// Removes its slot on drop, so a view's upload URL dies with the view. +pub struct UploadHandle { + service: Weak, + upload_id: Uuid, + cancel: Arc, +} + +impl UploadHandle { + pub fn upload_id(&self) -> Uuid { + self.upload_id + } + + /// Ask an in-flight request to stop. The endpoint notices between chunks. + pub fn cancel(&self) { + self.cancel.store(true, Ordering::SeqCst); + } + + /// The shared cancellation flag, so a hook can raise it without the handle. + pub fn cancel_flag(&self) -> Arc { + Arc::clone(&self.cancel) + } +} + +impl Drop for UploadHandle { + fn drop(&mut self) { + if let Some(service) = self.service.upgrade() { + service.remove(self.upload_id); + } + } +} + +impl std::fmt::Debug for UploadHandle { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("UploadHandle") + .field("upload_id", &self.upload_id) + .finish() + } +} + +impl UploadService { + pub fn new(connection_id: impl Into) -> Self { + UploadService { + connection_id: connection_id.into(), + entries: Mutex::new(HashMap::new()), + } + } + + pub fn connection_id(&self) -> &str { + &self.connection_id + } + + /// Register an upload slot and return its handle plus the URL to POST to. + /// The slot stays available until the handle is dropped. + pub fn add_upload( + self: &Arc, + observer: UploadObserver, + constraints: UploadConstraints, + ) -> (UploadHandle, String) { + self.add_upload_with_cancel(observer, constraints, Arc::new(AtomicBool::new(false))) + } + + /// [`add_upload`](Self::add_upload) with a caller-supplied cancellation flag, + /// for a hook that must be able to cancel from a build without waiting for + /// the mount effect to hand back a handle. + pub fn add_upload_with_cancel( + self: &Arc, + observer: UploadObserver, + constraints: UploadConstraints, + cancel: Arc, + ) -> (UploadHandle, String) { + let upload_id = Uuid::new_v4(); + self.entries.lock().unwrap().insert( + upload_id, + UploadEntry { + observer, + constraints, + cancel: Arc::clone(&cancel), + }, + ); + + let url = format!("/rusty/upload/{}/{}", self.connection_id, upload_id); + let handle = UploadHandle { + service: Arc::downgrade(self), + upload_id, + cancel, + }; + (handle, url) + } + + /// Resolve a slot for one request, cloning it out from under the lock. + /// + /// Unlike `DownloadService::take` this does **not** remove the entry: a slot is + /// reusable across files (pick a file, then pick another), and its lifetime + /// belongs to the handle rather than to a single request. + pub fn slot(&self, upload_id: Uuid) -> Option { + let entries = self.entries.lock().unwrap(); + let entry = entries.get(&upload_id)?; + Some(UploadSlot { + constraints: entry.constraints.clone(), + observer: Arc::clone(&entry.observer), + cancel: Arc::clone(&entry.cancel), + }) + } + + /// Number of upload slots currently registered. + pub fn len(&self) -> usize { + self.entries.lock().unwrap().len() + } + + pub fn is_empty(&self) -> bool { + self.len() == 0 + } + + fn remove(&self, upload_id: Uuid) { + self.entries.lock().unwrap().remove(&upload_id); + } +} + +impl std::fmt::Debug for UploadService { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("UploadService") + .field("connection_id", &self.connection_id) + .field("uploads", &self.len()) + .finish() + } +} + +/// Whether `mime_type`/`file_name` satisfies an HTML `accept` list. +/// +/// An empty list accepts everything. Entries are matched case-insensitively; a +/// `type/*` entry matches on the type, a `.ext` entry on the file name's suffix, +/// and anything else on the exact MIME type with its parameters (`; charset=..`) +/// stripped. +pub fn accepts(accept: &[String], mime_type: &str, file_name: &str) -> bool { + if accept.is_empty() { + return true; + } + + let base_mime = mime_type + .split(';') + .next() + .unwrap_or("") + .trim() + .to_ascii_lowercase(); + let lower_name = file_name.to_ascii_lowercase(); + + accept.iter().any(|pattern| { + let pattern = pattern.trim().to_ascii_lowercase(); + if pattern.is_empty() { + false + } else if pattern == "*" || pattern == "*/*" { + true + } else if let Some(type_prefix) = pattern.strip_suffix("/*") { + base_mime + .split('/') + .next() + .is_some_and(|actual| actual == type_prefix) + } else if pattern.starts_with('.') { + lower_name.ends_with(&pattern) + } else { + base_mime == pattern + } + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::AtomicUsize; + + /// An observer that records every event it sees. + fn recording_observer() -> (UploadObserver, Arc>>) { + let events = Arc::new(Mutex::new(Vec::new())); + let sink = Arc::clone(&events); + let observer: UploadObserver = Arc::new(move |event| sink.lock().unwrap().push(event)); + (observer, events) + } + + fn a_file(name: &str) -> UploadedFile { + UploadedFile { + file_name: name.to_string(), + mime_type: "text/csv".to_string(), + content: Bytes::from_static(b"id,name"), + } + } + + #[test] + fn test_add_upload_returns_a_connection_scoped_url() { + let service = Arc::new(UploadService::new("conn-1")); + let (observer, _events) = recording_observer(); + + let (_handle, url) = service.add_upload(observer, UploadConstraints::new()); + + assert!( + url.starts_with("/rusty/upload/conn-1/"), + "unexpected url: {url}" + ); + let id = url.rsplit('/').next().unwrap(); + assert!(Uuid::parse_str(id).is_ok(), "not a uuid: {id}"); + assert_eq!(service.len(), 1); + } + + #[test] + fn test_dropping_the_handle_unregisters_the_slot() { + let service = Arc::new(UploadService::new("conn-1")); + let (observer, _events) = recording_observer(); + let (handle, _url) = service.add_upload(observer, UploadConstraints::new()); + let id = handle.upload_id(); + assert!(service.slot(id).is_some()); + + drop(handle); + + assert_eq!(service.len(), 0); + assert!( + service.slot(id).is_none(), + "an unregistered slot must not resolve" + ); + } + + #[test] + fn test_slot_does_not_remove_the_entry() { + let service = Arc::new(UploadService::new("conn-1")); + let (observer, events) = recording_observer(); + let (handle, _url) = service.add_upload(observer, UploadConstraints::new().max_bytes(1024)); + let id = handle.upload_id(); + + // Two sequential uploads through one slot, as a view that lets the user + // pick a second file would do. + let first = service.slot(id).expect("first upload"); + assert_eq!(first.constraints.max_bytes, Some(1024)); + first.emit(UploadEvent::Completed(a_file("one.csv"))); + + let second = service + .slot(id) + .expect("second upload through the same slot"); + second.emit(UploadEvent::Completed(a_file("two.csv"))); + + assert_eq!(service.len(), 1, "the slot must survive a completed upload"); + assert_eq!(events.lock().unwrap().len(), 2); + } + + #[test] + fn test_unknown_id_does_not_resolve() { + let service = Arc::new(UploadService::new("conn-1")); + assert!(service.slot(Uuid::new_v4()).is_none()); + assert!(service.is_empty()); + } + + #[test] + fn test_cancel_flag_is_shared_between_handle_and_slot() { + let service = Arc::new(UploadService::new("conn-1")); + let (observer, _events) = recording_observer(); + let (handle, _url) = service.add_upload(observer, UploadConstraints::new()); + + let slot = service.slot(handle.upload_id()).unwrap(); + assert!(!slot.is_cancelled()); + + handle.cancel(); + assert!(slot.is_cancelled(), "the flag is shared, not copied"); + // And a slot resolved after the cancel sees it too. + assert!(service.slot(handle.upload_id()).unwrap().is_cancelled()); + } + + #[test] + fn test_caller_supplied_cancel_flag_is_the_one_the_slot_sees() { + let service = Arc::new(UploadService::new("conn-1")); + let (observer, _events) = recording_observer(); + let cancel = Arc::new(AtomicBool::new(false)); + + let (handle, _url) = + service.add_upload_with_cancel(observer, UploadConstraints::new(), Arc::clone(&cancel)); + + cancel.store(true, Ordering::SeqCst); + assert!(service.slot(handle.upload_id()).unwrap().is_cancelled()); + } + + #[test] + fn test_observer_receives_every_event() { + let service = Arc::new(UploadService::new("conn-1")); + let (observer, events) = recording_observer(); + let (handle, _url) = service.add_upload(observer, UploadConstraints::new()); + let slot = service.slot(handle.upload_id()).unwrap(); + + slot.emit(UploadEvent::Progress { + received: 4, + total: Some(8), + }); + slot.emit(UploadEvent::Failed(UploadError::NoFile)); + + let events = events.lock().unwrap(); + assert_eq!(events.len(), 2); + assert!(matches!( + events[0], + UploadEvent::Progress { + received: 4, + total: Some(8) + } + )); + assert!(matches!( + events[1], + UploadEvent::Failed(UploadError::NoFile) + )); + } + + #[test] + fn test_slots_are_isolated_per_service() { + let a = Arc::new(UploadService::new("conn-a")); + let b = Arc::new(UploadService::new("conn-b")); + let (observer, _events) = recording_observer(); + let (handle, _url) = a.add_upload(observer, UploadConstraints::new()); + + assert!( + b.slot(handle.upload_id()).is_none(), + "another connection must not resolve this slot" + ); + assert_eq!(a.connection_id(), "conn-a"); + assert_eq!(b.connection_id(), "conn-b"); + } + + #[test] + fn test_accepts_matches_wildcards_exact_types_and_extensions() { + // (accept, mime, name, expected) + let cases: &[(&[&str], &str, &str, bool)] = &[ + (&[], "application/x-anything", "weird.bin", true), + (&["image/*"], "image/png", "logo.png", true), + (&["image/*"], "text/plain", "notes.txt", false), + (&["IMAGE/*"], "image/jpeg", "photo.jpg", true), + (&["image/*"], "IMAGE/JPEG", "photo.jpg", true), + (&["application/pdf"], "application/pdf", "a.pdf", true), + (&["application/pdf"], "application/x-pdf", "a.pdf", false), + (&["text/csv"], "text/csv; charset=utf-8", "a.csv", true), + (&[".csv"], "application/octet-stream", "export.csv", true), + (&[".csv"], "application/octet-stream", "export.CSV", true), + (&[".CSV"], "application/octet-stream", "export.csv", true), + (&[".csv"], "application/octet-stream", "export.tsv", false), + ( + &["image/*", ".csv"], + "application/octet-stream", + "a.csv", + true, + ), + (&["*/*"], "application/zip", "a.zip", true), + (&["image/png"], "image/png", "", true), + ]; + + for (accept, mime, name, expected) in cases { + let accept: Vec = accept.iter().map(|s| s.to_string()).collect(); + assert_eq!( + accepts(&accept, mime, name), + *expected, + "accepts({accept:?}, {mime:?}, {name:?})" + ); + } + } + + #[test] + fn test_constraints_builder_carries_its_accept_list() { + let constraints = UploadConstraints::new() + .accept([".csv", "text/csv"]) + .max_bytes(100) + .min_bytes(1); + + assert_eq!(constraints.max_bytes, Some(100)); + assert_eq!(constraints.min_bytes, Some(1)); + assert!(constraints.accepts("application/octet-stream", "data.csv")); + assert!(!constraints.accepts("image/png", "logo.png")); + } + + #[test] + fn test_upload_error_status_codes() { + assert_eq!( + UploadError::TooLarge { + limit: 1, + actual: 2 + } + .status_code(), + StatusCode::PAYLOAD_TOO_LARGE + ); + assert_eq!( + UploadError::TooSmall { + limit: 2, + actual: 1 + } + .status_code(), + StatusCode::UNSUPPORTED_MEDIA_TYPE + ); + assert_eq!( + UploadError::RejectedMimeType { + mime_type: "image/png".into(), + accept: vec![".csv".into()] + } + .status_code(), + StatusCode::UNSUPPORTED_MEDIA_TYPE + ); + assert_eq!(UploadError::NoFile.status_code(), StatusCode::BAD_REQUEST); + assert_eq!( + UploadError::Cancelled.status_code(), + StatusCode::BAD_REQUEST + ); + assert_eq!( + UploadError::Transport("broken".into()).status_code(), + StatusCode::BAD_REQUEST + ); + } + + #[test] + fn test_upload_error_messages_name_the_numbers() { + assert!(UploadError::TooLarge { + limit: 10, + actual: 99 + } + .to_string() + .contains("99")); + assert!(UploadError::RejectedMimeType { + mime_type: "image/png".into(), + accept: vec!["text/csv".into(), ".csv".into()] + } + .to_string() + .contains("text/csv, .csv")); + } + + #[test] + fn test_uploaded_file_debug_omits_the_bytes() { + let file = UploadedFile { + file_name: "secret.bin".into(), + mime_type: "application/octet-stream".into(), + content: Bytes::from_static(b"sensitive-payload"), + }; + let rendered = format!("{file:?}"); + + assert!(rendered.contains("secret.bin"), "got {rendered}"); + assert!(rendered.contains("len: 17"), "got {rendered}"); + assert!(!rendered.contains("sensitive"), "got {rendered}"); + assert_eq!(file.len(), 17); + assert!(!file.is_empty()); + } + + #[test] + fn test_observer_is_shared_not_cloned_per_slot() { + let service = Arc::new(UploadService::new("conn-1")); + let calls = Arc::new(AtomicUsize::new(0)); + let observer: UploadObserver = { + let calls = Arc::clone(&calls); + Arc::new(move |_| { + calls.fetch_add(1, Ordering::SeqCst); + }) + }; + let (handle, _url) = service.add_upload(observer, UploadConstraints::new()); + + service + .slot(handle.upload_id()) + .unwrap() + .emit(UploadEvent::Failed(UploadError::Cancelled)); + service + .slot(handle.upload_id()) + .unwrap() + .emit(UploadEvent::Failed(UploadError::Cancelled)); + + assert_eq!(calls.load(Ordering::SeqCst), 2); + } +} From 9108d814353265d1e85404888aa36fa784715a65 Mon Sep 17 00:00:00 2001 From: rorychatt Date: Mon, 10 Aug 2026 11:16:10 +0200 Subject: [PATCH 3/6] [00140] Add the use_stream and use_stream_text hooks Drives a futures::Stream into view state chunk by chunk on a spawned task, with an optional retry budget and a ring cap on how much is retained. restart() bumps a generation State so the effect re-registers, and both restart and the effect cleanup abort the previous task before anything new writes the state. --- rusty-macros/src/hook_rules.rs | 2 + rusty/src/hooks/mod.rs | 4 + rusty/src/hooks/use_stream.rs | 1070 ++++++++++++++++++++++++++++++++ rusty/src/lib.rs | 4 +- 4 files changed, 1078 insertions(+), 2 deletions(-) create mode 100644 rusty/src/hooks/use_stream.rs diff --git a/rusty-macros/src/hook_rules.rs b/rusty-macros/src/hook_rules.rs index d4da60f..b24300f 100644 --- a/rusty-macros/src/hook_rules.rs +++ b/rusty-macros/src/hook_rules.rs @@ -47,6 +47,8 @@ pub(crate) const SLOT_CONSUMING_HOOKS: &[&str] = &[ "use_ref", "use_signal", "use_state", + "use_stream", + "use_stream_text", "use_trigger", "use_trigger_unit", ]; diff --git a/rusty/src/hooks/mod.rs b/rusty/src/hooks/mod.rs index c06d06a..8b3bfa1 100644 --- a/rusty/src/hooks/mod.rs +++ b/rusty/src/hooks/mod.rs @@ -15,6 +15,7 @@ pub mod use_ref; pub mod use_service; pub mod use_signal; pub mod use_state; +pub mod use_stream; pub mod use_trigger; pub use deps::{deps_changed, DynEq}; @@ -33,4 +34,7 @@ pub use use_ref::{use_ref, Ref}; pub use use_service::{try_use_service, use_service}; pub use use_signal::{signal_registry, use_receiver_id, use_signal}; pub use use_state::{use_state, State}; +pub use use_stream::{ + use_stream, use_stream_text, StreamOptions, StreamResult, StreamStatus, TextStreamResult, +}; pub use use_trigger::{use_trigger, use_trigger_unit}; diff --git a/rusty/src/hooks/use_stream.rs b/rusty/src/hooks/use_stream.rs new file mode 100644 index 0000000..3a1267c --- /dev/null +++ b/rusty/src/hooks/use_stream.rs @@ -0,0 +1,1070 @@ +use std::future::Future; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use futures::{Stream, StreamExt}; +use tokio::task::JoinHandle; + +use crate::core::query_cache::QueryError; +use crate::hooks::deps::DynEq; +use crate::hooks::use_effect::use_effect_with_deps; +use crate::hooks::use_ref::use_ref; +use crate::hooks::use_state::{use_state, State}; +use crate::views::view::BuildContext; + +/// Where a consumed stream is in its lifecycle. +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub enum StreamStatus { + /// Nothing is running: either `auto_start: false` and no `restart` yet, or + /// [`StreamResult::stop`] was called. + #[default] + Idle, + /// Chunks are arriving. A retry stays `Streaming` — a recovered hiccup is not + /// something a view should have to render. + Streaming, + /// The stream ended without an error. + Done, + /// The stream failed and every retry was used up. + Error(String), +} + +/// How [`use_stream`] consumes its stream. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct StreamOptions { + /// Open the stream on mount. `false` waits for [`StreamResult::restart`]. + pub auto_start: bool, + /// Reopen the stream this many times after a failure. `0` means give up on the + /// first error. + pub max_retries: u32, + /// How long to wait before each retry. + pub retry_delay: Duration, + /// Cap how many chunks are retained, oldest dropped first. `None` keeps all of + /// them — fine for an LLM response, wrong for an endless feed. + pub max_chunks: Option, +} + +impl Default for StreamOptions { + fn default() -> Self { + StreamOptions { + auto_start: true, + max_retries: 0, + retry_delay: Duration::from_secs(1), + max_chunks: None, + } + } +} + +impl StreamOptions { + pub fn new() -> Self { + StreamOptions::default() + } + + pub fn auto_start(mut self, auto_start: bool) -> Self { + self.auto_start = auto_start; + self + } + + pub fn max_retries(mut self, max_retries: u32) -> Self { + self.max_retries = max_retries; + self + } + + pub fn retry_delay(mut self, retry_delay: Duration) -> Self { + self.retry_delay = retry_delay; + self + } + + pub fn max_chunks(mut self, max_chunks: usize) -> Self { + self.max_chunks = Some(max_chunks); + self + } +} + +/// The chunks a view has received so far, plus the controls to stop and reopen. +/// +/// Cheap to clone — every field is `Arc`-backed, so an event handler can capture it. +pub struct StreamResult { + /// Chunks in arrival order, capped by + /// [`StreamOptions::max_chunks`](StreamOptions#structfield.max_chunks). + pub chunks: State>, + pub status: State, + restart: Arc, + stop: Arc, +} + +impl StreamResult { + /// Abort whatever is running, drop the chunks, reset the status to `Idle`, and + /// reopen the stream from the factory. Also how an `auto_start: false` stream is + /// started. + pub fn restart(&self) { + (self.restart)(); + } + + /// Abort whatever is running and go back to `Idle`, keeping the chunks received + /// so far. + pub fn stop(&self) { + (self.stop)(); + } + + pub fn is_streaming(&self) -> bool { + self.status.get() == StreamStatus::Streaming + } + + /// Number of chunks received so far. + pub fn len(&self) -> usize { + self.chunks.get().len() + } + + pub fn is_empty(&self) -> bool { + self.len() == 0 + } +} + +impl Clone for StreamResult { + fn clone(&self) -> Self { + StreamResult { + chunks: self.chunks.clone(), + status: self.status.clone(), + restart: Arc::clone(&self.restart), + stop: Arc::clone(&self.stop), + } + } +} + +impl std::fmt::Debug for StreamResult { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("StreamResult") + .field("chunks", &self.chunks.get()) + .field("status", &self.status.get()) + .finish() + } +} + +/// The concatenated text of a token stream, handed back by [`use_stream_text`]. +pub struct TextStreamResult { + /// Every chunk so far, joined in arrival order. + pub text: State, + pub status: State, + restart: Arc, + stop: Arc, +} + +impl TextStreamResult { + /// Abort whatever is running, clear the text, reset the status to `Idle`, and + /// reopen the stream. + pub fn restart(&self) { + (self.restart)(); + } + + /// Abort whatever is running and go back to `Idle`, keeping the text so far. + pub fn stop(&self) { + (self.stop)(); + } + + pub fn is_streaming(&self) -> bool { + self.status.get() == StreamStatus::Streaming + } +} + +impl Clone for TextStreamResult { + fn clone(&self) -> Self { + TextStreamResult { + text: self.text.clone(), + status: self.status.clone(), + restart: Arc::clone(&self.restart), + stop: Arc::clone(&self.stop), + } + } +} + +impl std::fmt::Debug for TextStreamResult { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("TextStreamResult") + .field("text", &self.text.get()) + .field("status", &self.status.get()) + .finish() + } +} + +/// The task slot a stream hook keeps across builds. +type TaskSlot = Arc>>>; + +/// Consume a `futures::Stream` into view state, rebuilding as chunks arrive. +/// +/// The counterpart to [`use_download_stream`](crate::hooks::use_download_stream), +/// which streams *out* to the client. This one streams *in* from a server-side +/// source — an LLM SDK call, an SSE relay, a channel feed — and pushes each chunk +/// to the browser over the session's WebSocket. +/// +/// `factory` returns the stream, so a retry or a +/// [`restart`](StreamResult::restart) can reopen it: an exhausted `Stream` cannot +/// be rewound, and this is the same shape `use_download_stream` uses. +/// +/// ```ignore +/// let tokens = use_stream(ctx, || async { open_llm_stream(&prompt).await }, StreamOptions::new()); +/// TextBlock::new(&tokens.chunks.get().join("")) +/// ``` +pub fn use_stream( + ctx: &mut BuildContext, + factory: F, + options: StreamOptions, +) -> StreamResult +where + F: Fn() -> Fut + Send + Sync + 'static, + Fut: Future> + Send + 'static, + S: Stream> + Send + 'static, + T: Send + Sync + Clone + 'static, +{ + let chunks = use_state(ctx, Vec::::new()); + let status = use_state(ctx, StreamStatus::default()); + // A `use_state`, not a `use_ref`: `restart` bumps it, and only a + // rebuild-triggering set makes the next build re-register the effect. + let generation = use_state(ctx, 0u64); + let task: TaskSlot = use_ref(ctx, Arc::new(Mutex::new(None))).get(); + + let on_chunk = { + let chunks = chunks.clone(); + let max_chunks = options.max_chunks; + Arc::new(move |chunk: T| { + chunks.update(move |previous| { + let mut next = previous.clone(); + next.push(chunk); + if let Some(cap) = max_chunks { + while next.len() > cap { + next.remove(0); + } + } + next + }); + }) as Arc + }; + + let clear = { + let chunks = chunks.clone(); + Arc::new(move || chunks.set(Vec::new())) as Arc + }; + + register_stream_effect(ctx, factory, options, &status, &generation, &task, on_chunk); + + StreamResult { + chunks, + status: status.clone(), + restart: restart_fn(&status, &generation, &task, clear), + stop: stop_fn(&status, &task), + } +} + +/// [`use_stream`] for token streams: concatenates the chunks instead of collecting +/// them, which is what an LLM response is rendered from. +/// +/// `max_chunks` caps how many chunks are *appended*; later chunks are still drained +/// from the stream but not shown, so a runaway generator cannot grow the string +/// without bound. +pub fn use_stream_text( + ctx: &mut BuildContext, + factory: F, + options: StreamOptions, +) -> TextStreamResult +where + F: Fn() -> Fut + Send + Sync + 'static, + Fut: Future> + Send + 'static, + S: Stream> + Send + 'static, +{ + let text = use_state(ctx, String::new()); + let status = use_state(ctx, StreamStatus::default()); + let generation = use_state(ctx, 0u64); + let task: TaskSlot = use_ref(ctx, Arc::new(Mutex::new(None))).get(); + + let appended = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let on_chunk = { + let text = text.clone(); + let appended = Arc::clone(&appended); + let max_chunks = options.max_chunks; + Arc::new(move |chunk: String| { + let seen = appended.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + if max_chunks.is_some_and(|cap| seen >= cap) { + return; + } + text.update(move |previous| format!("{previous}{chunk}")); + }) as Arc + }; + + let clear = { + let text = text.clone(); + let appended = Arc::clone(&appended); + Arc::new(move || { + appended.store(0, std::sync::atomic::Ordering::SeqCst); + text.set(String::new()); + }) as Arc + }; + + register_stream_effect(ctx, factory, options, &status, &generation, &task, on_chunk); + + TextStreamResult { + text, + status: status.clone(), + restart: restart_fn(&status, &generation, &task, clear), + stop: stop_fn(&status, &task), + } +} + +/// Register the one effect both hooks use, keyed on `generation` so a `restart` +/// reopens the stream. +/// +/// The effect is the last slot either hook consumes, so the two share a slot layout +/// and swapping one for the other at a call site shifts nothing. +fn register_stream_effect( + ctx: &mut BuildContext, + factory: F, + options: StreamOptions, + status: &State, + generation: &State, + task: &TaskSlot, + on_chunk: Arc, +) where + F: Fn() -> Fut + Send + Sync + 'static, + Fut: Future> + Send + 'static, + S: Stream> + Send + 'static, + T: Send + Sync + Clone + 'static, +{ + let current = generation.get(); + let status = status.clone(); + let spawn_slot = Arc::clone(task); + let cleanup_slot = Arc::clone(task); + + use_effect_with_deps(ctx, &[¤t as &dyn DynEq], move |_| { + // Generation 0 with `auto_start: false` means the view has not asked for + // anything yet: stay Idle and never call the factory. + if current == 0 && !options.auto_start { + return None; + } + + // Belt and braces: the runtime runs the previous effect's cleanup before + // this callback, but a `restart` that raced a rebuild must not leave two + // tasks writing the same state. + abort_previous(&spawn_slot); + let handle = spawn_stream_task(factory, options, status, on_chunk); + *spawn_slot.lock().unwrap() = Some(handle); + + Some(Box::new(move || { + abort_previous(&cleanup_slot); + }) as Box) + }); +} + +/// Drive the stream on its own task, retrying a failure per [`StreamOptions`]. +fn spawn_stream_task( + factory: F, + options: StreamOptions, + status: State, + on_chunk: Arc, +) -> JoinHandle<()> +where + F: Fn() -> Fut + Send + Sync + 'static, + Fut: Future> + Send + 'static, + S: Stream> + Send + 'static, + T: Send + Sync + Clone + 'static, +{ + tokio::spawn(async move { + let mut retries_used = 0u32; + + loop { + status.set(StreamStatus::Streaming); + + // A failure — opening the stream or a chunk part-way through — is + // whatever we end this attempt with. Chunks already delivered stay. + let failure = match factory().await { + Err(error) => Some(error), + Ok(stream) => { + let mut stream = Box::pin(stream); + let mut failure = None; + while let Some(item) = stream.next().await { + match item { + Ok(chunk) => on_chunk(chunk), + Err(error) => { + failure = Some(error); + break; + } + } + } + failure + } + }; + + let Some(error) = failure else { + status.set(StreamStatus::Done); + return; + }; + + if retries_used >= options.max_retries { + status.set(StreamStatus::Error(error.message)); + return; + } + retries_used += 1; + tokio::time::sleep(options.retry_delay).await; + } + }) +} + +/// Abort the task in `slot`, if any. Safe to call when nothing is running. +fn abort_previous(slot: &TaskSlot) { + if let Some(handle) = slot.lock().unwrap().take() { + handle.abort(); + } +} + +/// `restart`: abort now (rather than waiting for the effect's cleanup), clear the +/// accumulated output, and bump the generation so the next build reopens. +/// +/// The status goes back to `Idle` as part of this: leaving a `Done` or `Error` from +/// the previous run in place would make a restart invisible until the first new +/// chunk arrived. +fn restart_fn( + status: &State, + generation: &State, + task: &TaskSlot, + clear: Arc, +) -> Arc { + let status = status.clone(); + let generation = generation.clone(); + let task = Arc::clone(task); + Arc::new(move || { + // Abort first: a task still appending would otherwise race the clear. + abort_previous(&task); + clear(); + status.set(StreamStatus::Idle); + generation.update(|current| current.wrapping_add(1)); + }) +} + +/// `stop`: abort the task and report `Idle`, keeping whatever arrived. +fn stop_fn(status: &State, task: &TaskSlot) -> Arc { + let status = status.clone(); + let task = Arc::clone(task); + Arc::new(move || { + abort_previous(&task); + status.set(StreamStatus::Idle); + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::hooks::hook_store::HookStore; + use crate::views::view::EffectCleanup; + use std::sync::atomic::{AtomicUsize, Ordering}; + + /// One build of `hook`, with its effects run and their cleanups collected. + fn mount( + store: &mut HookStore, + hook: impl FnOnce(&mut BuildContext) -> R, + ) -> (R, Vec) { + let (result, effects) = { + let mut ctx = BuildContext::new(store, None); + let result = hook(&mut ctx); + (result, ctx.drain_effects()) + }; + + let mut cleanups = Vec::new(); + for effect in effects { + if let Some(cleanup) = (effect.callback)() { + cleanups.push(cleanup); + } + } + (result, cleanups) + } + + /// A rebuild of `hook` against the same store, as the runtime does after a + /// `State::set` — including running the previous effect's cleanup first. + fn rebuild( + store: &mut HookStore, + cleanups: Vec, + hook: impl FnOnce(&mut BuildContext) -> R, + ) -> (R, Vec) { + for cleanup in cleanups { + cleanup(); + } + mount(store, hook) + } + + /// Let every ready task run to its next await point, without moving the clock. + /// + /// The paused-clock tests need this instead of [`wait_until`]: a `sleep` on the + /// test's own task keeps the runtime busy, so tokio's auto-advance never fires + /// and only an explicit `tokio::time::advance` moves virtual time. + async fn settle() { + for _ in 0..32 { + tokio::task::yield_now().await; + } + } + + /// Poll until `predicate` holds, or fail. Only for tests on the real clock — + /// see [`settle`] for the paused ones. + async fn wait_until(what: &str, predicate: impl Fn() -> bool) { + for _ in 0..400 { + if predicate() { + return; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + panic!("timed out waiting for {what}"); + } + + async fn wait_for_status(status: &State, expected: StreamStatus) { + let probe = status.clone(); + let target = expected.clone(); + wait_until(&format!("status {expected:?}"), move || { + probe.get() == target + }) + .await; + } + + /// A stream that yields `items` one at a time, sleeping between them so a test + /// can interleave with it. + fn paced_stream( + items: Vec, + delay: Duration, + ) -> impl Stream> + Send { + futures::stream::unfold(items.into_iter(), move |mut items| async move { + tokio::time::sleep(delay).await; + let next = items.next(); + next.map(|item| (Ok(item), items)) + }) + } + + #[tokio::test] + async fn test_chunks_accumulate_in_order_and_status_reaches_done() { + let mut store = HookStore::new(); + + let (result, _cleanups) = mount(&mut store, |ctx| { + use_stream( + ctx, + || async { Ok(futures::stream::iter(vec![Ok(1), Ok(2), Ok(3)])) }, + StreamOptions::new(), + ) + }); + + wait_for_status(&result.status, StreamStatus::Done).await; + assert_eq!(result.chunks.get(), vec![1, 2, 3]); + assert_eq!(result.len(), 3); + assert!(!result.is_empty()); + assert!(!result.is_streaming()); + } + + #[tokio::test] + async fn test_status_is_streaming_while_chunks_are_still_arriving() { + let mut store = HookStore::new(); + + let (result, _cleanups) = mount(&mut store, |ctx| { + use_stream( + ctx, + || async { Ok(paced_stream(vec![1, 2, 3], Duration::from_millis(20))) }, + StreamOptions::new(), + ) + }); + + let chunks = result.chunks.clone(); + wait_until("the first chunk", move || !chunks.get().is_empty()).await; + assert!(result.is_streaming(), "got {:?}", result.status.get()); + + wait_for_status(&result.status, StreamStatus::Done).await; + assert_eq!(result.chunks.get(), vec![1, 2, 3]); + } + + #[tokio::test] + async fn test_factory_error_with_no_retries_becomes_error() { + let mut store = HookStore::new(); + let calls = Arc::new(AtomicUsize::new(0)); + + let (result, _cleanups) = { + let calls = Arc::clone(&calls); + mount(&mut store, move |ctx| { + use_stream( + ctx, + move || { + let calls = Arc::clone(&calls); + async move { + calls.fetch_add(1, Ordering::SeqCst); + Err::>, _>( + QueryError::new("upstream is down"), + ) + } + }, + StreamOptions::new(), + ) + }) + }; + + wait_for_status( + &result.status, + StreamStatus::Error("upstream is down".to_string()), + ) + .await; + assert_eq!(calls.load(Ordering::SeqCst), 1, "max_retries: 0 means once"); + assert!(result.chunks.get().is_empty()); + } + + #[tokio::test] + async fn test_a_chunk_error_is_retried_and_keeps_the_earlier_chunks() { + let mut store = HookStore::new(); + let attempts = Arc::new(AtomicUsize::new(0)); + + let (result, _cleanups) = { + let attempts = Arc::clone(&attempts); + mount(&mut store, move |ctx| { + use_stream( + ctx, + move || { + let attempt = attempts.fetch_add(1, Ordering::SeqCst); + async move { + if attempt == 0 { + // Attempt 1 delivers a chunk, then breaks. + Ok(futures::stream::iter(vec![ + Ok(1), + Err(QueryError::new("connection reset")), + Ok(99), + ])) + } else { + Ok(futures::stream::iter(vec![Ok(2), Ok(3)])) + } + } + }, + StreamOptions::new() + .max_retries(1) + .retry_delay(Duration::from_millis(10)), + ) + }) + }; + + wait_for_status(&result.status, StreamStatus::Done).await; + assert_eq!( + result.chunks.get(), + vec![1, 2, 3], + "the recovered attempt appends to what attempt 1 delivered" + ); + assert_eq!(attempts.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn test_retries_are_exhausted_then_report_the_last_error() { + let mut store = HookStore::new(); + let attempts = Arc::new(AtomicUsize::new(0)); + + let (result, _cleanups) = { + let attempts = Arc::clone(&attempts); + mount(&mut store, move |ctx| { + use_stream( + ctx, + move || { + let attempt = attempts.fetch_add(1, Ordering::SeqCst); + async move { + Err::>, _>( + QueryError::new(format!("attempt {attempt} failed")), + ) + } + }, + StreamOptions::new() + .max_retries(2) + .retry_delay(Duration::from_millis(10)), + ) + }) + }; + + wait_for_status( + &result.status, + StreamStatus::Error("attempt 2 failed".to_string()), + ) + .await; + assert_eq!( + attempts.load(Ordering::SeqCst), + 3, + "one attempt plus two retries" + ); + } + + #[tokio::test(start_paused = true)] + async fn test_a_retry_waits_out_the_retry_delay() { + let mut store = HookStore::new(); + let attempts = Arc::new(AtomicUsize::new(0)); + + let (result, _cleanups) = { + let attempts = Arc::clone(&attempts); + mount(&mut store, move |ctx| { + use_stream( + ctx, + move || { + attempts.fetch_add(1, Ordering::SeqCst); + async move { + Err::>, _>( + QueryError::new("still down"), + ) + } + }, + StreamOptions::new() + .max_retries(1) + .retry_delay(Duration::from_secs(10)), + ) + }) + }; + + settle().await; + assert_eq!(attempts.load(Ordering::SeqCst), 1); + assert_eq!( + result.status.get(), + StreamStatus::Streaming, + "a pending retry is not an Error yet" + ); + + tokio::time::advance(Duration::from_secs(9)).await; + settle().await; + assert_eq!( + attempts.load(Ordering::SeqCst), + 1, + "the retry must wait out retry_delay, not fire immediately" + ); + + tokio::time::advance(Duration::from_secs(2)).await; + settle().await; + assert_eq!(attempts.load(Ordering::SeqCst), 2); + assert_eq!( + result.status.get(), + StreamStatus::Error("still down".to_string()), + "the second failure exhausts the single retry" + ); + } + + #[tokio::test] + async fn test_effect_cleanup_aborts_the_task_and_stops_the_writes() { + let mut store = HookStore::new(); + + let (result, cleanups) = mount(&mut store, |ctx| { + use_stream( + ctx, + || async { + Ok(paced_stream( + (0..100).collect::>(), + Duration::from_millis(10), + )) + }, + StreamOptions::new(), + ) + }); + + let chunks = result.chunks.clone(); + wait_until("the first chunk", move || !chunks.get().is_empty()).await; + + // Unmount. + for cleanup in cleanups { + cleanup(); + } + let at_unmount = result.chunks.get().len(); + + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!( + result.chunks.get().len(), + at_unmount, + "an aborted task must not keep writing state" + ); + assert_ne!(result.status.get(), StreamStatus::Done); + } + + #[tokio::test] + async fn test_restart_clears_the_chunks_and_reopens_the_stream() { + let mut store = HookStore::new(); + let calls = Arc::new(AtomicUsize::new(0)); + + let hook = { + let calls = Arc::clone(&calls); + move |ctx: &mut BuildContext| { + let calls = Arc::clone(&calls); + use_stream( + ctx, + move || { + let calls = Arc::clone(&calls); + async move { + let call = calls.fetch_add(1, Ordering::SeqCst); + Ok(futures::stream::iter(vec![Ok(call as i32)])) + } + }, + StreamOptions::new(), + ) + } + }; + + let (result, cleanups) = mount(&mut store, hook.clone()); + wait_for_status(&result.status, StreamStatus::Done).await; + assert_eq!(result.chunks.get(), vec![0]); + + result.restart(); + assert!( + result.chunks.get().is_empty(), + "restart clears immediately, before the rebuild" + ); + + // The runtime rebuilds because `generation` changed. + let (result, _cleanups) = rebuild(&mut store, cleanups, hook); + wait_for_status(&result.status, StreamStatus::Done).await; + assert_eq!(result.chunks.get(), vec![1], "the factory ran again"); + assert_eq!(calls.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn test_auto_start_false_stays_idle_and_never_calls_the_factory() { + let mut store = HookStore::new(); + let calls = Arc::new(AtomicUsize::new(0)); + + let hook = { + let calls = Arc::clone(&calls); + move |ctx: &mut BuildContext| { + let calls = Arc::clone(&calls); + use_stream( + ctx, + move || { + let calls = Arc::clone(&calls); + async move { + calls.fetch_add(1, Ordering::SeqCst); + Ok(futures::stream::iter(vec![Ok(7)])) + } + }, + StreamOptions::new().auto_start(false), + ) + } + }; + + let (result, cleanups) = mount(&mut store, hook.clone()); + tokio::time::sleep(Duration::from_millis(50)).await; + assert_eq!(result.status.get(), StreamStatus::Idle); + assert_eq!(calls.load(Ordering::SeqCst), 0); + assert!(result.chunks.get().is_empty()); + + // A restart is how an opt-in stream is started. + result.restart(); + let (result, _cleanups) = rebuild(&mut store, cleanups, hook); + wait_for_status(&result.status, StreamStatus::Done).await; + assert_eq!(result.chunks.get(), vec![7]); + assert_eq!(calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn test_stop_aborts_and_reports_idle_while_keeping_the_chunks() { + let mut store = HookStore::new(); + + let (result, _cleanups) = mount(&mut store, |ctx| { + use_stream( + ctx, + || async { + Ok(paced_stream( + (0..100).collect::>(), + Duration::from_millis(10), + )) + }, + StreamOptions::new(), + ) + }); + + let chunks = result.chunks.clone(); + wait_until("the first chunk", move || !chunks.get().is_empty()).await; + + result.stop(); + let at_stop = result.chunks.get(); + assert!(!at_stop.is_empty(), "stop keeps what already arrived"); + assert_eq!(result.status.get(), StreamStatus::Idle); + + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!(result.chunks.get(), at_stop); + assert_eq!(result.status.get(), StreamStatus::Idle); + } + + #[tokio::test] + async fn test_max_chunks_keeps_only_the_most_recent() { + let mut store = HookStore::new(); + + let (result, _cleanups) = mount(&mut store, |ctx| { + use_stream( + ctx, + || async { Ok(futures::stream::iter((1..=10).map(Ok))) }, + StreamOptions::new().max_chunks(3), + ) + }); + + wait_for_status(&result.status, StreamStatus::Done).await; + assert_eq!(result.chunks.get(), vec![8, 9, 10]); + } + + #[tokio::test] + async fn test_use_stream_text_concatenates_its_chunks() { + let mut store = HookStore::new(); + + let (result, _cleanups) = mount(&mut store, |ctx| { + use_stream_text( + ctx, + || async { + Ok(futures::stream::iter(vec![ + Ok("Hello".to_string()), + Ok(", ".to_string()), + Ok("world".to_string()), + ])) + }, + StreamOptions::new(), + ) + }); + + wait_for_status(&result.status, StreamStatus::Done).await; + assert_eq!(result.text.get(), "Hello, world"); + assert!(!result.is_streaming()); + assert!(format!("{result:?}").contains("Hello, world")); + } + + #[tokio::test] + async fn test_use_stream_text_caps_appended_chunks() { + let mut store = HookStore::new(); + + let (result, _cleanups) = mount(&mut store, |ctx| { + use_stream_text( + ctx, + || async { + Ok(futures::stream::iter( + ["a", "b", "c", "d"].map(|s| Ok(s.to_string())), + )) + }, + StreamOptions::new().max_chunks(2), + ) + }); + + wait_for_status(&result.status, StreamStatus::Done).await; + assert_eq!(result.text.get(), "ab"); + } + + #[tokio::test] + async fn test_use_stream_text_reports_a_failure() { + let mut store = HookStore::new(); + + let (result, _cleanups) = mount(&mut store, |ctx| { + use_stream_text( + ctx, + || async { + Ok(futures::stream::iter(vec![ + Ok("partial".to_string()), + Err(QueryError::new("token stream broke")), + ])) + }, + StreamOptions::new(), + ) + }); + + wait_for_status( + &result.status, + StreamStatus::Error("token stream broke".to_string()), + ) + .await; + assert_eq!( + result.text.get(), + "partial", + "what arrived before the error stays" + ); + } + + #[tokio::test] + async fn test_use_stream_text_restart_clears_the_text() { + let mut store = HookStore::new(); + let calls = Arc::new(AtomicUsize::new(0)); + + let hook = { + let calls = Arc::clone(&calls); + move |ctx: &mut BuildContext| { + let calls = Arc::clone(&calls); + use_stream_text( + ctx, + move || { + let calls = Arc::clone(&calls); + async move { + let call = calls.fetch_add(1, Ordering::SeqCst); + Ok(futures::stream::iter(vec![Ok(format!("run-{call}"))])) + } + }, + StreamOptions::new(), + ) + } + }; + + let (result, cleanups) = mount(&mut store, hook.clone()); + wait_for_status(&result.status, StreamStatus::Done).await; + assert_eq!(result.text.get(), "run-0"); + + result.restart(); + assert_eq!(result.text.get(), ""); + + let (result, _cleanups) = rebuild(&mut store, cleanups, hook); + wait_for_status(&result.status, StreamStatus::Done).await; + assert_eq!(result.text.get(), "run-1"); + } + + #[tokio::test] + async fn test_use_stream_text_stop_keeps_the_text_and_reports_idle() { + let mut store = HookStore::new(); + + let (result, _cleanups) = mount(&mut store, |ctx| { + use_stream_text( + ctx, + || async { + Ok(futures::stream::unfold(0usize, |seen| async move { + tokio::time::sleep(Duration::from_millis(10)).await; + Some((Ok(format!("{seen} ")), seen + 1)) + })) + }, + StreamOptions::new(), + ) + }); + + let text = result.text.clone(); + wait_until("the first token", move || !text.get().is_empty()).await; + result.stop(); + + let at_stop = result.text.get(); + tokio::time::sleep(Duration::from_millis(60)).await; + assert_eq!(result.text.get(), at_stop, "the task was aborted"); + assert_eq!(result.status.get(), StreamStatus::Idle); + } + + #[tokio::test] + async fn test_result_is_cloneable_and_shares_its_state() { + let mut store = HookStore::new(); + + let (result, _cleanups) = mount(&mut store, |ctx| { + use_stream( + ctx, + || async { Ok(futures::stream::iter(vec![Ok(42)])) }, + StreamOptions::new(), + ) + }); + let captured = result.clone(); + + wait_for_status(&result.status, StreamStatus::Done).await; + assert_eq!(captured.chunks.get(), vec![42]); + assert_eq!(captured.status.get(), StreamStatus::Done); + } + + #[test] + fn test_default_options_auto_start_with_no_retries() { + let options = StreamOptions::default(); + assert!(options.auto_start); + assert_eq!(options.max_retries, 0); + assert_eq!(options.retry_delay, Duration::from_secs(1)); + assert_eq!(options.max_chunks, None); + + let tuned = StreamOptions::new() + .auto_start(false) + .max_retries(3) + .retry_delay(Duration::from_millis(250)) + .max_chunks(64); + assert!(!tuned.auto_start); + assert_eq!(tuned.max_retries, 3); + assert_eq!(tuned.retry_delay, Duration::from_millis(250)); + assert_eq!(tuned.max_chunks, Some(64)); + } + + #[test] + fn test_stream_status_defaults_to_idle() { + assert_eq!(StreamStatus::default(), StreamStatus::Idle); + } +} diff --git a/rusty/src/lib.rs b/rusty/src/lib.rs index b72694f..b493a4f 100644 --- a/rusty/src/lib.rs +++ b/rusty/src/lib.rs @@ -15,8 +15,8 @@ pub mod prelude { create_context, try_use_service, use_alert, use_callback, use_context, use_download, use_download_bytes, use_download_stream, use_effect, use_effect_with_deps, use_form, use_interval, use_memo, use_mutation, use_query, use_reducer, use_ref, use_service, - use_signal, use_state, use_trigger, use_trigger_unit, DynEq, QueryMutator, QueryResult, - Ref, ShowAlert, State, + use_signal, use_state, use_stream, use_stream_text, use_trigger, use_trigger_unit, DynEq, + QueryMutator, QueryResult, Ref, ShowAlert, State, StreamOptions, StreamStatus, }; pub use crate::server::{RustyServer, DEFAULT_BIND_ADDRESS}; pub use crate::shared::{Align, Color, Density, Icon, Justify, NamedColor, Size}; From 13849c690c7d5404ebf9ce89f7a91418e68d97ed Mon Sep 17 00:00:00 2001 From: rorychatt Date: Mon, 10 Aug 2026 11:16:27 +0200 Subject: [PATCH 4/6] [00140] Add the use_upload and use_upload_to hooks Registers a slot on mount and publishes its URL, so a view renders its picker only once the URL exists. Progress is what the server has received, capped at 99 until the bytes are in hand. use_upload_to hands each file to a sink instead of holding it in view state, and reports a sink error the same way a rejected MIME type is reported. Both hooks use the same slot layout so swapping one for the other does not shift any later hook. --- rusty-macros/src/hook_rules.rs | 2 + rusty/src/hooks/mod.rs | 2 + rusty/src/hooks/use_upload.rs | 609 +++++++++++++++++++++++++++++++++ rusty/src/lib.rs | 8 +- 4 files changed, 619 insertions(+), 2 deletions(-) create mode 100644 rusty/src/hooks/use_upload.rs diff --git a/rusty-macros/src/hook_rules.rs b/rusty-macros/src/hook_rules.rs index b24300f..1aa18ab 100644 --- a/rusty-macros/src/hook_rules.rs +++ b/rusty-macros/src/hook_rules.rs @@ -51,6 +51,8 @@ pub(crate) const SLOT_CONSUMING_HOOKS: &[&str] = &[ "use_stream_text", "use_trigger", "use_trigger_unit", + "use_upload", + "use_upload_to", ]; /// Hooks whose return value triggers a rebuild when mutated. diff --git a/rusty/src/hooks/mod.rs b/rusty/src/hooks/mod.rs index 8b3bfa1..e969a6b 100644 --- a/rusty/src/hooks/mod.rs +++ b/rusty/src/hooks/mod.rs @@ -17,6 +17,7 @@ pub mod use_signal; pub mod use_state; pub mod use_stream; pub mod use_trigger; +pub mod use_upload; pub use deps::{deps_changed, DynEq}; pub use use_alert::{use_alert, AlertCallback, ShowAlert}; @@ -38,3 +39,4 @@ pub use use_stream::{ use_stream, use_stream_text, StreamOptions, StreamResult, StreamStatus, TextStreamResult, }; pub use use_trigger::{use_trigger, use_trigger_unit}; +pub use use_upload::{use_upload, use_upload_to, Upload, UploadStatus}; diff --git a/rusty/src/hooks/use_upload.rs b/rusty/src/hooks/use_upload.rs new file mode 100644 index 0000000..0d9fc52 --- /dev/null +++ b/rusty/src/hooks/use_upload.rs @@ -0,0 +1,609 @@ +use std::future::Future; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; + +use crate::core::query_cache::QueryError; +use crate::hooks::use_effect::use_effect; +use crate::hooks::use_ref::use_ref; +use crate::hooks::use_state::{use_state, State}; +use crate::server::upload::{ + UploadConstraints, UploadEvent, UploadObserver, UploadService, UploadedFile, +}; +use crate::views::view::BuildContext; + +/// Where an upload slot is in its lifecycle. +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub enum UploadStatus { + /// Registered, nothing received yet. + #[default] + Idle, + /// A body is arriving. + Uploading, + /// The file is in hand (and, for [`use_upload_to`], the sink accepted it). + Done, + /// The upload was rejected or broke — the string is the + /// [`UploadError`](crate::server::upload::UploadError)'s message. + Error(String), +} + +/// The upload slot a view holds, handed back by [`use_upload`]. +/// +/// Cheap to clone: every field is `Arc`-backed, so an event handler can capture it. +pub struct Upload { + /// The URL to POST the file to. `None` until the mount effect registers the + /// slot, so a view renders its picker only when this is `Some`. + pub url: State>, + /// Bytes the **server** has received, as a percentage of the request's + /// `Content-Length`, clamped to `99` until the file is complete — so `100` + /// always means the bytes are in hand. The browser's own optimistic progress + /// bar (`useUploadWithProgress.ts`) is a separate number that reaches 100 as + /// soon as the socket is drained. + pub progress: State, + pub status: State, + /// The received file. Always `None` for [`use_upload_to`], which hands the + /// bytes to its sink instead of holding them in view state. + pub file: State>, + cancel: Arc, + reset: Arc, +} + +impl Upload { + /// Ask an in-flight upload to stop. The endpoint notices between chunks and + /// answers `400`; the status becomes `Error("the upload was cancelled")`. + /// + /// A no-op if nothing is in flight — the flag stays raised, so call + /// [`reset`](Self::reset) before offering the slot again. + pub fn cancel(&self) { + (self.cancel)(); + } + + pub fn is_uploading(&self) -> bool { + self.status.get() == UploadStatus::Uploading + } + + /// Back to `Idle` with no file, no progress and the cancellation flag cleared. + /// The URL is untouched: the slot is still registered and reusable. + pub fn reset(&self) { + (self.reset)(); + } +} + +impl Clone for Upload { + fn clone(&self) -> Self { + Upload { + url: self.url.clone(), + progress: self.progress.clone(), + status: self.status.clone(), + file: self.file.clone(), + cancel: Arc::clone(&self.cancel), + reset: Arc::clone(&self.reset), + } + } +} + +impl std::fmt::Debug for Upload { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Upload") + .field("url", &self.url.get()) + .field("progress", &self.progress.get()) + .field("status", &self.status.get()) + .field("file", &self.file.get()) + .finish() + } +} + +/// Persists an uploaded file somewhere other than view state. +type UploadSink = Arc< + dyn Fn(UploadedFile) -> std::pin::Pin> + Send>> + + Send + + Sync, +>; + +/// Register an upload slot and get back its URL plus the progress state. +/// +/// The client half already exists: `uploadFileWithProgress` in the frontend POSTs +/// a `FormData` with a single field named `file` to this URL over XHR, so the +/// endpoint this hook registers matches that shape exactly. +/// +/// The received bytes land in [`Upload::file`]. For anything large enough that +/// holding it in view state is wrong, use [`use_upload_to`]. +/// +/// Requires an [`UploadService`] on the registry, which `AppSessionStore` registers +/// per connection — so upload URLs are not guessable across sessions. +pub fn use_upload(ctx: &mut BuildContext, constraints: UploadConstraints) -> Upload { + use_upload_inner(ctx, constraints, None) +} + +/// [`use_upload`] that streams each completed file into `sink` instead of keeping +/// it in view state. +/// +/// [`Upload::file`] stays `None`. The status reaches `Done` only once the sink's +/// future resolves `Ok`; a sink error becomes `Error` with the `QueryError`'s +/// message, so a failed database write is as visible as a rejected MIME type. +pub fn use_upload_to( + ctx: &mut BuildContext, + constraints: UploadConstraints, + sink: F, +) -> Upload +where + F: Fn(UploadedFile) -> Fut + Send + Sync + 'static, + Fut: Future> + Send + 'static, +{ + let sink: UploadSink = Arc::new(move |file| Box::pin(sink(file))); + use_upload_inner(ctx, constraints, Some(sink)) +} + +/// Slot layout shared by both hooks, so swapping one for the other at a call site +/// does not shift any later hook's slot: `url`, `progress`, `status`, `file`, +/// `cancel flag`, `effect`. +fn use_upload_inner( + ctx: &mut BuildContext, + constraints: UploadConstraints, + sink: Option, +) -> Upload { + let url = use_state(ctx, None::); + let progress = use_state(ctx, 0u8); + let status = use_state(ctx, UploadStatus::default()); + let file = use_state(ctx, None::); + let cancel_flag = use_ref(ctx, Arc::new(AtomicBool::new(false))); + + let service = ctx + .services() + .get::() + .expect("use_upload requires an UploadService on the ServiceRegistry"); + + let observer = observer_for(&progress, &status, &file, sink); + let flag = cancel_flag.get(); + + let effect_url = url.clone(); + let effect_flag = Arc::clone(&flag); + use_effect(ctx, move || { + let (handle, slot_url) = service.add_upload_with_cancel(observer, constraints, effect_flag); + effect_url.set(Some(slot_url)); + + // Dropping the handle unregisters the slot. + Some(Box::new(move || { + drop(handle); + }) as Box) + }); + + let cancel = { + let flag = Arc::clone(&flag); + Arc::new(move || flag.store(true, Ordering::SeqCst)) as Arc + }; + let reset = { + let flag = Arc::clone(&flag); + let progress = progress.clone(); + let status = status.clone(); + let file = file.clone(); + Arc::new(move || { + flag.store(false, Ordering::SeqCst); + progress.set(0); + file.set(None); + status.set(UploadStatus::Idle); + }) as Arc + }; + + Upload { + url, + progress, + status, + file, + cancel, + reset, + } +} + +/// The observer the endpoint calls as a body arrives, writing the hook's states. +/// +/// Runs on the HTTP task, not during a build, so `State::set` here is the same +/// cross-task path `use_query` uses: the `RebuildHandle` inside `State` pushes the +/// rebuild out over the WebSocket. +fn observer_for( + progress: &State, + status: &State, + file: &State>, + sink: Option, +) -> UploadObserver { + let progress = progress.clone(); + let status = status.clone(); + let file = file.clone(); + + Arc::new(move |event| match event { + UploadEvent::Progress { received, total } => { + // An unknown total leaves the percentage where it was: a bar that + // cannot advance is better than one that snaps back to zero. + if let Some(total) = total { + if total > 0 { + progress.set(percent_received(received, total)); + } + } + if status.get() != UploadStatus::Uploading { + status.set(UploadStatus::Uploading); + } + } + UploadEvent::Completed(received) => match sink.clone() { + None => { + file.set(Some(received)); + progress.set(100); + status.set(UploadStatus::Done); + } + Some(sink) => { + let status = status.clone(); + let progress = progress.clone(); + tokio::spawn(async move { + match sink(received).await { + Ok(()) => { + progress.set(100); + status.set(UploadStatus::Done); + } + Err(error) => status.set(UploadStatus::Error(error.message)), + } + }); + } + }, + UploadEvent::Failed(error) => { + status.set(UploadStatus::Error(error.to_string())); + } + }) +} + +/// Received bytes as a percentage, capped at 99 so only `Completed` reports 100. +fn percent_received(received: u64, total: u64) -> u8 { + let percent = received.saturating_mul(100) / total; + percent.min(99) as u8 +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::core::services::ServiceRegistry; + use crate::hooks::hook_store::HookStore; + use crate::server::upload::{UploadError, UploadSlot}; + use crate::views::view::EffectCleanup; + use bytes::Bytes; + use std::sync::Mutex; + use uuid::Uuid; + + fn test_services() -> (Arc, Arc) { + let upload_service = Arc::new(UploadService::new("conn-1")); + let services = Arc::new(ServiceRegistry::new()); + services.register(Arc::clone(&upload_service)); + (services, upload_service) + } + + fn a_file() -> UploadedFile { + UploadedFile { + file_name: "export.csv".to_string(), + mime_type: "text/csv".to_string(), + content: Bytes::from_static(b"id,name\n1,alice"), + } + } + + /// Build once, run the effects, and return the hook's result plus any cleanups. + fn mount( + store: &mut HookStore, + services: &Arc, + constraints: UploadConstraints, + ) -> (Upload, Vec) { + mount_with(store, services, |ctx| use_upload(ctx, constraints)) + } + + fn mount_with( + store: &mut HookStore, + services: &Arc, + build: impl FnOnce(&mut BuildContext) -> Upload, + ) -> (Upload, Vec) { + let (upload, effects) = { + let mut ctx = + BuildContext::with_services(store, None, Uuid::nil(), Arc::clone(services)); + let upload = build(&mut ctx); + (upload, ctx.drain_effects()) + }; + + let mut cleanups = Vec::new(); + for effect in effects { + if let Some(cleanup) = (effect.callback)() { + cleanups.push(cleanup); + } + } + (upload, cleanups) + } + + /// The slot the hook registered, resolved from the URL it published. + fn slot_of(service: &Arc, upload: &Upload) -> UploadSlot { + let url = upload + .url + .get() + .expect("the effect should have set the url"); + let id = Uuid::parse_str(url.rsplit('/').next().unwrap()).unwrap(); + service.slot(id).expect("the slot should be registered") + } + + /// Poll `predicate` until it holds — for the states a spawned sink task writes. + async fn wait_until(predicate: impl Fn() -> bool) { + for _ in 0..200 { + if predicate() { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + panic!("condition never held"); + } + + #[tokio::test] + async fn test_mount_registers_exactly_one_slot_and_publishes_its_url() { + let (services, service) = test_services(); + let mut store = HookStore::new(); + + let (upload, _cleanups) = mount(&mut store, &services, UploadConstraints::new()); + + let url = upload + .url + .get() + .expect("the effect should have set the url"); + assert!( + url.starts_with("/rusty/upload/conn-1/"), + "unexpected url: {url}" + ); + assert_eq!(service.len(), 1); + assert_eq!(upload.status.get(), UploadStatus::Idle); + assert_eq!(upload.progress.get(), 0); + assert!(upload.file.get().is_none()); + assert!(!upload.is_uploading()); + } + + #[tokio::test] + async fn test_constraints_reach_the_registered_slot() { + let (services, service) = test_services(); + let mut store = HookStore::new(); + + let (upload, _cleanups) = mount( + &mut store, + &services, + UploadConstraints::new().accept([".csv"]).max_bytes(64), + ); + + let slot = slot_of(&service, &upload); + assert_eq!(slot.constraints.max_bytes, Some(64)); + assert!(slot + .constraints + .accepts("application/octet-stream", "a.csv")); + } + + #[tokio::test] + async fn test_progress_events_move_the_percentage_and_completion_reaches_100() { + let (services, service) = test_services(); + let mut store = HookStore::new(); + + let (upload, _cleanups) = mount(&mut store, &services, UploadConstraints::new()); + let slot = slot_of(&service, &upload); + + slot.emit(UploadEvent::Progress { + received: 0, + total: Some(15), + }); + assert_eq!(upload.progress.get(), 0); + assert_eq!(upload.status.get(), UploadStatus::Uploading); + assert!(upload.is_uploading()); + + // Even a fully received body reports 99 while it is still a Progress event. + slot.emit(UploadEvent::Progress { + received: 15, + total: Some(15), + }); + assert_eq!(upload.progress.get(), 99); + + slot.emit(UploadEvent::Completed(a_file())); + assert_eq!(upload.progress.get(), 100); + assert_eq!(upload.status.get(), UploadStatus::Done); + assert_eq!( + upload + .file + .get() + .expect("the file should be in hand") + .content, + Bytes::from_static(b"id,name\n1,alice") + ); + } + + #[tokio::test] + async fn test_progress_without_a_total_leaves_the_percentage_alone() { + let (services, service) = test_services(); + let mut store = HookStore::new(); + + let (upload, _cleanups) = mount(&mut store, &services, UploadConstraints::new()); + let slot = slot_of(&service, &upload); + + slot.emit(UploadEvent::Progress { + received: 5, + total: Some(10), + }); + assert_eq!(upload.progress.get(), 50); + + slot.emit(UploadEvent::Progress { + received: 8, + total: None, + }); + assert_eq!( + upload.progress.get(), + 50, + "an unknown total must not reset the bar" + ); + assert_eq!(upload.status.get(), UploadStatus::Uploading); + } + + #[tokio::test] + async fn test_failure_reports_the_reason_and_keeps_the_file_empty() { + let (services, service) = test_services(); + let mut store = HookStore::new(); + + let (upload, _cleanups) = mount(&mut store, &services, UploadConstraints::new()); + let slot = slot_of(&service, &upload); + + slot.emit(UploadEvent::Failed(UploadError::TooLarge { + limit: 10, + actual: 99, + })); + + match upload.status.get() { + UploadStatus::Error(message) => { + assert!(message.contains("99"), "got {message}"); + assert!(message.contains("10"), "got {message}"); + } + other => panic!("expected Error, got {other:?}"), + } + assert!(upload.file.get().is_none()); + } + + #[tokio::test] + async fn test_cancel_raises_the_flag_the_slot_reads() { + let (services, service) = test_services(); + let mut store = HookStore::new(); + + let (upload, _cleanups) = mount(&mut store, &services, UploadConstraints::new()); + let slot = slot_of(&service, &upload); + assert!(!slot.is_cancelled()); + + upload.cancel(); + assert!(slot.is_cancelled()); + + // A slot resolved after the cancel sees it too — the endpoint checks + // between chunks, so it may not have resolved the slot yet. + assert!(slot_of(&service, &upload).is_cancelled()); + } + + #[tokio::test] + async fn test_reset_clears_the_state_and_lowers_the_cancel_flag() { + let (services, service) = test_services(); + let mut store = HookStore::new(); + + let (upload, _cleanups) = mount(&mut store, &services, UploadConstraints::new()); + let slot = slot_of(&service, &upload); + slot.emit(UploadEvent::Completed(a_file())); + upload.cancel(); + + upload.reset(); + + assert_eq!(upload.status.get(), UploadStatus::Idle); + assert_eq!(upload.progress.get(), 0); + assert!(upload.file.get().is_none()); + assert!(!slot.is_cancelled()); + // The slot survives a reset: the URL is still usable. + assert!(upload.url.get().is_some()); + assert_eq!(service.len(), 1); + } + + #[tokio::test] + async fn test_effect_cleanup_unregisters_the_slot() { + let (services, service) = test_services(); + let mut store = HookStore::new(); + + let (upload, cleanups) = mount(&mut store, &services, UploadConstraints::new()); + let url = upload.url.get().unwrap(); + let id = Uuid::parse_str(url.rsplit('/').next().unwrap()).unwrap(); + assert_eq!(service.len(), 1); + + // Unmount. + for cleanup in cleanups { + cleanup(); + } + + assert_eq!(service.len(), 0); + assert!(service.slot(id).is_none()); + } + + #[tokio::test] + async fn test_upload_is_cloneable_and_shares_its_state() { + let (services, service) = test_services(); + let mut store = HookStore::new(); + + let (upload, _cleanups) = mount(&mut store, &services, UploadConstraints::new()); + let captured = upload.clone(); + slot_of(&service, &upload).emit(UploadEvent::Completed(a_file())); + + assert_eq!(captured.status.get(), UploadStatus::Done); + assert_eq!(captured.progress.get(), 100); + assert!(format!("{captured:?}").contains("export.csv")); + } + + #[test] + #[should_panic(expected = "use_upload requires an UploadService")] + fn test_missing_upload_service_panics() { + let mut store = HookStore::new(); + let mut ctx = BuildContext::new(&mut store, None); + let _ = use_upload(&mut ctx, UploadConstraints::new()); + } + + #[test] + #[should_panic(expected = "use_upload requires an UploadService")] + fn test_missing_upload_service_panics_for_the_sink_hook() { + let mut store = HookStore::new(); + let mut ctx = BuildContext::new(&mut store, None); + let _ = use_upload_to(&mut ctx, UploadConstraints::new(), |_| async { Ok(()) }); + } + + #[tokio::test] + async fn test_use_upload_to_hands_the_bytes_to_the_sink_and_holds_nothing() { + let (services, service) = test_services(); + let mut store = HookStore::new(); + let persisted = Arc::new(Mutex::new(Vec::::new())); + + let (upload, _cleanups) = { + let persisted = Arc::clone(&persisted); + mount_with(&mut store, &services, move |ctx| { + use_upload_to(ctx, UploadConstraints::new(), move |file| { + let persisted = Arc::clone(&persisted); + async move { + persisted.lock().unwrap().push(file.file_name.clone()); + Ok(()) + } + }) + }) + }; + + slot_of(&service, &upload).emit(UploadEvent::Completed(a_file())); + + let status = upload.status.clone(); + wait_until(move || status.get() == UploadStatus::Done).await; + assert_eq!(upload.progress.get(), 100); + assert!( + upload.file.get().is_none(), + "use_upload_to must not hold the bytes in view state" + ); + assert_eq!(persisted.lock().unwrap().as_slice(), ["export.csv"]); + } + + #[tokio::test] + async fn test_use_upload_to_reports_a_sink_error_as_error() { + let (services, service) = test_services(); + let mut store = HookStore::new(); + + let (upload, _cleanups) = mount_with(&mut store, &services, |ctx| { + use_upload_to(ctx, UploadConstraints::new(), |_file| async { + Err(QueryError::new("disk is full")) + }) + }); + + slot_of(&service, &upload).emit(UploadEvent::Completed(a_file())); + + let status = upload.status.clone(); + wait_until(move || matches!(status.get(), UploadStatus::Error(_))).await; + assert_eq!( + upload.status.get(), + UploadStatus::Error("disk is full".to_string()) + ); + assert!(upload.file.get().is_none()); + assert_ne!(upload.progress.get(), 100); + } + + #[test] + fn test_percent_received_caps_at_99() { + assert_eq!(percent_received(0, 10), 0); + assert_eq!(percent_received(5, 10), 50); + assert_eq!(percent_received(10, 10), 99); + // The multipart envelope makes received exceed the file's own length. + assert_eq!(percent_received(20, 10), 99); + assert_eq!(percent_received(u64::MAX, 3), 99); + } +} diff --git a/rusty/src/lib.rs b/rusty/src/lib.rs index b493a4f..941b8bf 100644 --- a/rusty/src/lib.rs +++ b/rusty/src/lib.rs @@ -15,8 +15,12 @@ pub mod prelude { create_context, try_use_service, use_alert, use_callback, use_context, use_download, use_download_bytes, use_download_stream, use_effect, use_effect_with_deps, use_form, use_interval, use_memo, use_mutation, use_query, use_reducer, use_ref, use_service, - use_signal, use_state, use_stream, use_stream_text, use_trigger, use_trigger_unit, DynEq, - QueryMutator, QueryResult, Ref, ShowAlert, State, StreamOptions, StreamStatus, + use_signal, use_state, use_stream, use_stream_text, use_trigger, use_trigger_unit, + use_upload, use_upload_to, DynEq, QueryMutator, QueryResult, Ref, ShowAlert, State, + StreamOptions, StreamStatus, UploadStatus, + }; + pub use crate::server::upload::{ + UploadConstraints, UploadError, UploadEvent, UploadedFile, DEFAULT_MAX_UPLOAD_BYTES, }; pub use crate::server::{RustyServer, DEFAULT_BIND_ADDRESS}; pub use crate::shared::{Align, Color, Density, Icon, Justify, NamedColor, Size}; From 590bafc0bdfadc6947b9fa1fb2c092150808e65b Mon Sep 17 00:00:00 2001 From: rorychatt Date: Mon, 10 Aug 2026 11:16:33 +0200 Subject: [PATCH 5/6] [00140] Serve multipart uploads on /rusty/upload/{connection}/{upload} Reads the file field chunk by chunk rather than through Field::bytes(), which is what makes progress reporting, mid-flight cancellation and rejecting an oversize body without buffering it possible. Every failure past a resolved slot reports itself through UploadEvent::Failed first, because the browser only sees the status code. The raised body limit is a layer on this route alone. --- rusty/src/server/ws.rs | 458 ++++++++++++++++++++++++++++++++++++++++- 1 file changed, 454 insertions(+), 4 deletions(-) diff --git a/rusty/src/server/ws.rs b/rusty/src/server/ws.rs index ff04b57..a111065 100644 --- a/rusty/src/server/ws.rs +++ b/rusty/src/server/ws.rs @@ -1,12 +1,13 @@ use axum::{ body::Body, extract::ws::{Message, WebSocket, WebSocketUpgrade}, - extract::{Path, Query, State}, - http::{header, StatusCode}, - response::IntoResponse, - routing::get, + extract::{DefaultBodyLimit, Multipart, Path, Query, State}, + http::{header, HeaderMap, StatusCode}, + response::{IntoResponse, Response}, + routing::{get, post}, Router, }; +use bytes::BytesMut; use futures::{SinkExt, StreamExt}; use serde::{Deserialize, Serialize}; use std::net::SocketAddr; @@ -21,6 +22,10 @@ use crate::views::view::View; use super::download::{DownloadPayload, DownloadService}; use super::session::{AppSession, AppSessionStore}; +use super::upload::{ + accepts, UploadError, UploadEvent, UploadService, UploadSlot, UploadedFile, + DEFAULT_MAX_UPLOAD_BYTES, MULTIPART_ENVELOPE_ALLOWANCE, +}; /// Messages sent from client to server. #[derive(Debug, Serialize, Deserialize)] @@ -92,6 +97,7 @@ pub struct RustyServer { apps: AppRegistry, services: ServiceRegistry, static_dir: Option, + max_upload_bytes: u64, } impl RustyServer { @@ -112,6 +118,7 @@ impl RustyServer { apps, services: ServiceRegistry::new(), static_dir: None, + max_upload_bytes: DEFAULT_MAX_UPLOAD_BYTES, } } @@ -126,6 +133,7 @@ impl RustyServer { apps: AppRegistry::new(), services: ServiceRegistry::new(), static_dir: None, + max_upload_bytes: DEFAULT_MAX_UPLOAD_BYTES, } } @@ -171,8 +179,24 @@ impl RustyServer { self } + /// Cap the request body the upload endpoint will accept. Defaults to + /// [`DEFAULT_MAX_UPLOAD_BYTES`] (32 MiB). + /// + /// This is the outer transport limit, applied as a `DefaultBodyLimit` layer on + /// the upload route alone — the WebSocket and download routes keep axum's own + /// 2 MiB default. A body over the limit is refused by the layer with a bare + /// `413` and no [`UploadError`], so per-file limits belong in + /// [`UploadConstraints::max_bytes`](crate::server::upload::UploadConstraints::max_bytes), + /// which reports a reason the view can render. Keep this at or above the largest + /// `max_bytes` any view asks for. + pub fn with_max_upload_bytes(mut self, max_upload_bytes: u64) -> Self { + self.max_upload_bytes = max_upload_bytes; + self + } + /// Build the axum router with WebSocket support. pub fn router(self) -> Router { + let max_upload_bytes = self.max_upload_bytes; let session_store = AppSessionStore::with_apps(Arc::new(self.apps), Arc::new(self.services)); let state = Arc::new(AppState { session_store }); @@ -185,6 +209,13 @@ impl RustyServer { "/rusty/download/{connection_id}/{download_id}", get(download_handler), ) + .route( + "/rusty/upload/{connection_id}/{upload_id}", + // The raised body limit applies to this route only. + post(upload_handler).layer(DefaultBodyLimit::max( + usize::try_from(max_upload_bytes).unwrap_or(usize::MAX), + )), + ) .with_state(state); if let Some(dir) = self.static_dir { @@ -293,6 +324,179 @@ async fn download_handler( .into_response() } +/// Receive a file for an upload slot registered by a view through `use_upload`. +/// +/// The body shape is fixed by the client half that already exists: +/// `uploadFileWithProgress` POSTs a `multipart/form-data` body with a single field +/// named `file`. Slots are keyed by connection, so a URL only resolves for the +/// session that created it, and anything unresolvable — unknown session, unparseable +/// or unknown upload id, no `UploadService` — is a 404 with no observer to notify. +/// +/// Once a slot *is* resolved, every failure reports itself to the view through +/// `UploadEvent::Failed` before answering, because the browser only sees the status +/// code and cannot render a reason. +async fn upload_handler( + Path((connection_id, upload_id)): Path<(String, String)>, + State(state): State>, + headers: HeaderMap, + // `Multipart` consumes the body, so it must come last. + mut multipart: Multipart, +) -> Response { + let Ok(upload_id) = Uuid::parse_str(&upload_id) else { + return StatusCode::NOT_FOUND.into_response(); + }; + let Some(session_arc) = state.session_store.get_session(&connection_id).await else { + return StatusCode::NOT_FOUND.into_response(); + }; + + // Resolve the service and release the session lock before reading the body: an + // upload can take a while and must not block the session's event loop. + let upload_service = { + let session = session_arc.read().await; + session.services.get::() + }; + let Some(upload_service) = upload_service else { + return StatusCode::NOT_FOUND.into_response(); + }; + // Unlike a download, resolving does not consume the slot — a view can accept a + // second file through the same URL. + let Some(slot) = upload_service.slot(upload_id) else { + return StatusCode::NOT_FOUND.into_response(); + }; + + let total = headers + .get(header::CONTENT_LENGTH) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse::().ok()); + + // Reject a body that cannot possibly fit before reading any of it. See + // MULTIPART_ENVELOPE_ALLOWANCE for why this is not a bare `total > max`. + if let (Some(max_bytes), Some(total)) = (slot.constraints.max_bytes, total) { + if total > max_bytes.saturating_add(MULTIPART_ENVELOPE_ALLOWANCE) { + return reject( + &slot, + UploadError::TooLarge { + limit: max_bytes, + actual: total, + }, + ); + } + } + + loop { + let field = match multipart.next_field().await { + Ok(Some(field)) => field, + // Every field consumed and none of them was the file. + Ok(None) => return reject(&slot, UploadError::NoFile), + Err(error) => return reject(&slot, UploadError::Transport(error.to_string())), + }; + + if field.name() != Some("file") { + continue; + } + + // Copy the metadata out before reading chunks, which borrows the field + // mutably. + let file_name = field.file_name().unwrap_or("upload").to_string(); + let mime_type = field + .content_type() + .unwrap_or("application/octet-stream") + .to_string(); + + if !accepts(&slot.constraints.accept, &mime_type, &file_name) { + return reject( + &slot, + UploadError::RejectedMimeType { + mime_type, + accept: slot.constraints.accept.clone(), + }, + ); + } + + return read_file_field(field, slot, file_name, mime_type, total).await; + } +} + +/// Drain one multipart field into memory, reporting progress and enforcing the +/// size constraints as the bytes arrive. +/// +/// Chunk by chunk rather than `Field::bytes()`: that is what makes progress +/// reporting and mid-flight cancellation possible, and it lets an oversize body be +/// rejected without buffering all of it. +async fn read_file_field( + mut field: axum::extract::multipart::Field<'_>, + slot: UploadSlot, + file_name: String, + mime_type: String, + total: Option, +) -> Response { + let mut content = BytesMut::new(); + let mut received = 0u64; + + loop { + if slot.is_cancelled() { + return reject(&slot, UploadError::Cancelled); + } + + match field.chunk().await { + Ok(Some(chunk)) => { + received += chunk.len() as u64; + if let Some(max_bytes) = slot.constraints.max_bytes { + if received > max_bytes { + return reject( + &slot, + UploadError::TooLarge { + limit: max_bytes, + actual: received, + }, + ); + } + } + content.extend_from_slice(&chunk); + slot.emit(UploadEvent::Progress { received, total }); + } + Ok(None) => break, + Err(error) => return reject(&slot, UploadError::Transport(error.to_string())), + } + } + + if let Some(min_bytes) = slot.constraints.min_bytes { + if received < min_bytes { + return reject( + &slot, + UploadError::TooSmall { + limit: min_bytes, + actual: received, + }, + ); + } + } + + let body = serde_json::json!({ + "fileName": file_name, + "mimeType": mime_type, + "size": content.len(), + }); + slot.emit(UploadEvent::Completed(UploadedFile { + file_name, + mime_type, + content: content.freeze(), + })); + + (StatusCode::OK, axum::Json(body)).into_response() +} + +/// Tell the view why the upload failed, then answer with the matching status. +/// +/// The client reads only the status code, so the observer call is the only way the +/// reason reaches the browser at all — as rendered view state. +fn reject(slot: &UploadSlot, error: UploadError) -> Response { + let status = error.status_code(); + tracing::debug!(%error, "upload rejected"); + slot.emit(UploadEvent::Failed(error)); + status.into_response() +} + /// Build the session's tree, send it as a full `Refresh`, and reset the reconciler /// baseline to what was just sent. /// @@ -931,4 +1135,250 @@ mod tests { assert!(response.contains("200 OK"), "got {response}"); assert!(response.trim_end().ends_with("ok"), "got {response}"); } + + // --- Multipart upload through the real endpoint --- + + use crate::server::upload::UploadConstraints; + + /// Renders its own upload URL, and once a file lands, the name and bytes that + /// reached view state — the only way a test on the far side of the socket can + /// see what the endpoint handed the hook. + struct UploadView { + constraints: UploadConstraints, + } + + impl crate::views::view::View for UploadView { + fn build(&self, ctx: &mut crate::views::view::BuildContext) -> crate::views::view::Element { + let upload = crate::hooks::use_upload(ctx, self.constraints.clone()); + let outcome = match upload.file.get() { + Some(file) => format!( + "got:{}:{}:{}", + file.file_name, + file.mime_type, + String::from_utf8_lossy(&file.content) + ), + None => format!("status:{:?}", upload.status.get()), + }; + + crate::widgets::Layout::vertical() + .children(vec![ + TextBlock::new(&upload.url.get().unwrap_or_default()).into(), + TextBlock::new(&outcome).into(), + ]) + .into() + } + } + + fn upload_router(constraints: UploadConstraints) -> Router { + RustyServer::empty(0) + .with_app("uploader", "Uploader", move || UploadView { + constraints: constraints.clone(), + }) + .router() + } + + /// Read messages until one carries the slot URL the mount effect published. + /// + /// It arrives in an `update`, not the initial `refresh`: the effect runs after + /// the first build, so the first tree still has `url == None`. + async fn next_upload_url(client: &mut Client) -> String { + for _ in 0..5 { + let text = next_message(client).await; + if let Some(start) = text.find("/rusty/upload/") { + let rest = &text[start..]; + let end = rest + .find(|c: char| !c.is_ascii_alphanumeric() && !"/-_".contains(c)) + .unwrap_or(rest.len()); + return rest[..end].to_string(); + } + } + panic!("the upload url never reached the client"); + } + + /// Read messages until one contains `needle`, which is how a test observes the + /// view state the endpoint's observer wrote. + async fn wait_for_text(client: &mut Client, needle: &str) -> String { + for _ in 0..10 { + let text = next_message(client).await; + if text.contains(needle) { + return text; + } + } + panic!("the client never received {needle}"); + } + + /// POST a single-field `multipart/form-data` body and return the raw response. + /// + /// Hand-rolled over `TcpStream` rather than adding an HTTP client + /// dev-dependency, the same way `test_health_endpoint_still_responds` does it. + /// The body shape mirrors what the browser's `FormData` produces. + async fn post_multipart( + addr: SocketAddr, + path: &str, + field_name: &str, + file_name: &str, + content_type: &str, + content: &[u8], + ) -> String { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + const BOUNDARY: &str = "rustyTestBoundary"; + + let mut body = Vec::new(); + body.extend_from_slice(format!("--{BOUNDARY}\r\n").as_bytes()); + body.extend_from_slice( + format!( + "Content-Disposition: form-data; name=\"{field_name}\"; filename=\"{file_name}\"\r\n" + ) + .as_bytes(), + ); + body.extend_from_slice(format!("Content-Type: {content_type}\r\n\r\n").as_bytes()); + body.extend_from_slice(content); + body.extend_from_slice(format!("\r\n--{BOUNDARY}--\r\n").as_bytes()); + + let mut request = format!( + "POST {path} HTTP/1.1\r\nHost: localhost\r\nContent-Type: multipart/form-data; \ + boundary={BOUNDARY}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ) + .into_bytes(); + request.extend_from_slice(&body); + + let mut stream = TcpStream::connect(addr).await.unwrap(); + stream.write_all(&request).await.unwrap(); + + let mut response = Vec::new(); + stream.read_to_end(&mut response).await.unwrap(); + String::from_utf8_lossy(&response).to_string() + } + + #[tokio::test] + async fn test_upload_accepts_a_file_and_the_bytes_reach_view_state() { + let addr = serve_on_loopback(upload_router(UploadConstraints::new())).await; + let mut client = connect(addr, "").await; + let url = next_upload_url(&mut client).await; + + let response = post_multipart( + addr, + &url, + "file", + "notes.csv", + "text/csv", + b"id,name\n1,alice", + ) + .await; + + assert!(response.contains("200 OK"), "got {response}"); + assert!( + response.contains("\"fileName\":\"notes.csv\""), + "got {response}" + ); + assert!(response.contains("\"size\":15"), "got {response}"); + + // The bytes made it all the way into the view's own state, not just the + // endpoint's response. + let tree = wait_for_text(&mut client, "got:").await; + assert!( + tree.contains("got:notes.csv:text/csv:id,name"), + "got {tree}" + ); + } + + #[tokio::test] + async fn test_upload_over_max_bytes_is_rejected_while_it_streams() { + // Small enough that Content-Length stays inside the envelope allowance, so + // the rejection has to come from the chunk loop. + let addr = serve_on_loopback(upload_router(UploadConstraints::new().max_bytes(8))).await; + let mut client = connect(addr, "").await; + let url = next_upload_url(&mut client).await; + + let response = + post_multipart(addr, &url, "file", "big.bin", "text/plain", &[b'x'; 64]).await; + + assert!(response.contains("413"), "got {response}"); + let tree = wait_for_text(&mut client, "status:Error(").await; + assert!(tree.contains("over the 8 byte limit"), "got {tree}"); + } + + #[tokio::test] + async fn test_upload_far_over_max_bytes_is_rejected_from_content_length() { + let addr = serve_on_loopback(upload_router(UploadConstraints::new().max_bytes(8))).await; + let mut client = connect(addr, "").await; + let url = next_upload_url(&mut client).await; + + // Past the envelope allowance, so it never gets read at all. + let response = post_multipart( + addr, + &url, + "file", + "huge.bin", + "text/plain", + &[b'x'; 20 * 1024], + ) + .await; + + assert!(response.contains("413"), "got {response}"); + let tree = wait_for_text(&mut client, "status:Error(").await; + assert!(tree.contains("byte limit"), "got {tree}"); + } + + #[tokio::test] + async fn test_upload_of_a_disallowed_mime_type_is_rejected() { + let router = upload_router(UploadConstraints::new().accept(["text/csv"])); + let addr = serve_on_loopback(router).await; + let mut client = connect(addr, "").await; + let url = next_upload_url(&mut client).await; + + let response = post_multipart(addr, &url, "file", "cat.png", "image/png", b"\x89PNG").await; + + assert!(response.contains("415"), "got {response}"); + let tree = wait_for_text(&mut client, "status:Error(").await; + assert!( + tree.contains("image/png is not one of text/csv"), + "got {tree}" + ); + } + + #[tokio::test] + async fn test_upload_to_an_unknown_upload_id_is_not_found() { + let addr = serve_on_loopback(upload_router(UploadConstraints::new())).await; + let mut client = connect(addr, "").await; + let url = next_upload_url(&mut client).await; + + // Same session, a slot that was never registered. + let (prefix, _) = url.rsplit_once('/').unwrap(); + let path = format!("{prefix}/{}", Uuid::new_v4()); + let response = post_multipart(addr, &path, "file", "a.csv", "text/csv", b"hi").await; + + assert!(response.contains("404"), "got {response}"); + } + + #[tokio::test] + async fn test_upload_to_another_connections_url_is_not_found() { + let addr = serve_on_loopback(upload_router(UploadConstraints::new())).await; + let mut client = connect(addr, "").await; + let url = next_upload_url(&mut client).await; + + // The slot id is real; only the connection is wrong. Guessing an id must not + // be enough to upload into someone else's session. + let upload_id = url.rsplit('/').next().unwrap(); + let path = format!("/rusty/upload/{}/{upload_id}", Uuid::new_v4()); + let response = post_multipart(addr, &path, "file", "a.csv", "text/csv", b"hi").await; + + assert!(response.contains("404"), "got {response}"); + } + + #[tokio::test] + async fn test_upload_without_a_file_field_is_a_bad_request() { + let addr = serve_on_loopback(upload_router(UploadConstraints::new())).await; + let mut client = connect(addr, "").await; + let url = next_upload_url(&mut client).await; + + let response = + post_multipart(addr, &url, "notTheFile", "a.csv", "text/csv", b"id,name").await; + + assert!(response.contains("400"), "got {response}"); + let tree = wait_for_text(&mut client, "status:Error(").await; + assert!(tree.contains("no file field"), "got {tree}"); + } } From 23f8639a23a8512a45b3d027964fa1c543514f3c Mon Sep 17 00:00:00 2001 From: rorychatt Date: Mon, 10 Aug 2026 11:16:33 +0200 Subject: [PATCH 6/6] [00140] Document the use_stream and use_upload hooks --- rusty-docs/docs/02_concepts/03_hooks.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/rusty-docs/docs/02_concepts/03_hooks.md b/rusty-docs/docs/02_concepts/03_hooks.md index a7293fb..e4ba2b6 100644 --- a/rusty-docs/docs/02_concepts/03_hooks.md +++ b/rusty-docs/docs/02_concepts/03_hooks.md @@ -21,6 +21,8 @@ Hooks let you add state and side effects to views. They must be called in the sa | `use_signal` | Send and receive messages between views, per session or server-wide | | `use_download` | Register a download and get the URL to serve it from | | `use_download_stream` | Serve a large download as a chunked stream instead of buffering it | +| `use_stream` / `use_stream_text` | Consume an async stream into view state chunk by chunk, with retries | +| `use_upload` / `use_upload_to` | Accept a file from the browser with progress, into view state or a sink | | `use_alert` | Show a modal alert and get the user's answer in a callback | | `use_trigger` | Render an element on demand, carrying a value to its factory |