diff --git a/Cargo.toml b/Cargo.toml index 64465a6..b5a6c33 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -145,7 +145,7 @@ js-sys = "0.3" tracing-wasm = "0.2" wasm-bindgen = "=0.2.121" wasm-bindgen-futures = "0.4" -web-sys = { version = "0.3", features = ["Window", "Document", "Element", "Event", "EventTarget", "HtmlCanvasElement", "WebGl2RenderingContext", "Performance", "MouseEvent", "WheelEvent", "PointerEvent", "DomRect", "BroadcastChannel", "console", "Response", "File", "FileList", "Blob", "HtmlAnchorElement", "HtmlInputElement", "HtmlDetailsElement", "Worker", "WorkerOptions", "WorkerType", "OffscreenCanvas", "MessageEvent", "ErrorEvent"] } +web-sys = { version = "0.3", features = ["Window", "Document", "Element", "Event", "EventTarget", "HtmlCanvasElement", "WebGl2RenderingContext", "Performance", "MouseEvent", "WheelEvent", "PointerEvent", "DomRect", "BroadcastChannel", "console", "Response", "ReadableStream", "ReadableStreamDefaultReader", "File", "FileList", "Blob", "HtmlAnchorElement", "HtmlInputElement", "HtmlDetailsElement", "Worker", "WorkerOptions", "WorkerType", "OffscreenCanvas", "MessageEvent", "ErrorEvent"] } [dev-dependencies] anyhow = "1" diff --git a/src/appview/drafts.rs b/src/appview/drafts.rs index b2c5594..e43875d 100644 --- a/src/appview/drafts.rs +++ b/src/appview/drafts.rs @@ -482,7 +482,8 @@ mod tests { crate::appview::test_support::seed_identity(&state.pool, DID_B, "bob.com").await; crate::appview::test_support::seed_profile(&state.pool, DID_A, "Alice").await; crate::appview::test_support::seed_profile(&state.pool, DID_B, "Bob").await; - let (pds, server) = crate::appview::test_support::loopback_pds_for_publish(false).await; + let (pds, create_records, server) = + crate::appview::test_support::loopback_pds_for_publish(false).await; state.resolver = crate::appview::test_support::loopback_resolver(&pds); let key_a = crate::appview::test_support::seed_oauth_session_at( &state, @@ -714,6 +715,60 @@ mod tests { ) .unwrap(); assert_eq!(published.models.len(), 1); + let requests = create_records.lock().await; + assert_eq!(requests.len(), 3); + assert_eq!( + requests + .iter() + .map(|request| request["collection"].as_str().unwrap()) + .collect::>(), + [ + "space.polymodel.library.part", + "space.polymodel.library.model", + "space.polymodel.library.thing", + ] + ); + for request in requests.iter() { + assert_eq!(request["repo"], DID_A); + } + assert_eq!(requests[0]["rkey"], serde_json::Value::Null); + assert_eq!( + requests[0]["record"]["$type"], + "space.polymodel.library.part" + ); + assert_eq!(requests[0]["record"]["name"], "part.stl"); + assert_eq!(requests[0]["record"]["file"]["mimeType"], "model/stl"); + assert_eq!(requests[0]["record"]["file"]["size"], staged.file.size); + assert_eq!(requests[1]["rkey"], serde_json::Value::Null); + assert_eq!( + requests[1]["record"]["$type"], + "space.polymodel.library.model" + ); + assert_eq!(requests[1]["record"]["name"], "Model"); + assert_eq!( + requests[1]["record"]["parts"][0]["uri"], + format!("at://{DID_A}/space.polymodel.library.part/record-0") + ); + assert_eq!( + requests[1]["record"]["parts"][0]["cid"], + "bafkreih3ryqpylsmh4siyygdtplff46bgrzjro4xpofu2widxbifkyqgam" + ); + assert_eq!(requests[2]["rkey"], serde_json::Value::Null); + assert_eq!( + requests[2]["record"]["$type"], + "space.polymodel.library.thing" + ); + assert_eq!(requests[2]["record"]["name"], "After"); + assert_eq!(requests[2]["record"]["license"], "CC-BY-4.0"); + assert_eq!( + requests[2]["record"]["models"][0]["uri"], + format!("at://{DID_A}/space.polymodel.library.model/record-1") + ); + assert_eq!( + requests[2]["record"]["models"][0]["cid"], + "bafkreih3ryqpylsmh4siyygdtplff46bgrzjro4xpofu2widxbifkyqgam" + ); + drop(requests); let deleted_after_publish = app .oneshot( axum::http::Request::builder() @@ -899,7 +954,8 @@ mod tests { let mut state = crate::appview::test_support::state().await; crate::appview::test_support::seed_identity(&state.pool, DID_A, "alice.com").await; crate::appview::test_support::seed_profile(&state.pool, DID_A, "Alice").await; - let (pds, server) = crate::appview::test_support::loopback_pds_for_publish(true).await; + let (pds, _create_records, server) = + crate::appview::test_support::loopback_pds_for_publish(true).await; state.resolver = crate::appview::test_support::loopback_resolver(&pds); let key = crate::appview::test_support::seed_oauth_session_at( &state, diff --git a/src/appview/test_support.rs b/src/appview/test_support.rs index ca3f472..badf775 100644 --- a/src/appview/test_support.rs +++ b/src/appview/test_support.rs @@ -285,9 +285,11 @@ pub(crate) async fn loopback_pds_with_upload( (endpoint, task) } +pub(crate) type RecordedCreateRecords = std::sync::Arc>>; + pub(crate) async fn loopback_pds_for_publish( fail_create: bool, -) -> (String, tokio::task::JoinHandle<()>) { +) -> (String, RecordedCreateRecords, tokio::task::JoinHandle<()>) { use axum::response::IntoResponse; use axum::routing::any; use std::sync::Arc; @@ -299,6 +301,7 @@ pub(crate) async fn loopback_pds_for_publish( let endpoint = format!("http://{}", listener.local_addr().unwrap()); let did_doc_endpoint = endpoint.clone(); let sequence = Arc::new(AtomicU64::new(0)); + let create_records: RecordedCreateRecords = Arc::new(tokio::sync::Mutex::new(Vec::new())); let app = axum::Router::new() .route( "/xrpc/com.atproto.identity.resolveDid", @@ -352,9 +355,14 @@ pub(crate) async fn loopback_pds_for_publish( "/xrpc/com.atproto.repo.createRecord", any({ let sequence = sequence.clone(); + let create_records = create_records.clone(); move |body: Body| { let sequence = sequence.clone(); + let create_records = create_records.clone(); async move { + let body = axum::body::to_bytes(body, 1 << 20).await.unwrap(); + let request: serde_json::Value = serde_json::from_slice(&body).unwrap(); + create_records.lock().await.push(request.clone()); if fail_create { return ( StatusCode::SERVICE_UNAVAILABLE, @@ -365,8 +373,6 @@ pub(crate) async fn loopback_pds_for_publish( ) .into_response(); } - let body = axum::body::to_bytes(body, 1 << 20).await.unwrap(); - let request: serde_json::Value = serde_json::from_slice(&body).unwrap(); let collection = request["collection"].as_str().unwrap(); let ordinal = sequence.fetch_add(1, Ordering::Relaxed); let rkey = request["rkey"] @@ -385,7 +391,7 @@ pub(crate) async fn loopback_pds_for_publish( let task = tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); - (endpoint, task) + (endpoint, create_records, task) } pub(crate) async fn loopback_pds_with_blobs( diff --git a/src/appview/writes.rs b/src/appview/writes.rs index fdfc333..687a6a2 100644 --- a/src/appview/writes.rs +++ b/src/appview/writes.rs @@ -1948,7 +1948,11 @@ mod tests { "{mime}" ); assert_eq!(output.file.chunks.len(), 1, "{mime}"); - assert_eq!(output.file.chunks[0].size, bytes.len() as i64, "{mime}"); + let output_chunk = &output.file.chunks[0]; + assert_eq!(output_chunk.offset, 0, "{mime}"); + assert_eq!(output_chunk.size, bytes.len() as i64, "{mime}"); + assert_eq!(output_chunk.blob.blob().mime_type.as_str(), mime, "{mime}"); + assert_eq!(output_chunk.blob.blob().size, bytes.len(), "{mime}"); let upload_id = output.upload_id.to_string(); let row = sqlx::query_as::<_, (String, String, i64, Vec, String)>( "SELECT owner_did, filename, size, sha256, file_json FROM upload_staging WHERE owner_did = ? AND upload_id = ?", @@ -1964,11 +1968,19 @@ mod tests { assert_eq!(row.3, digest, "{mime}"); let stored: File = serde_json::from_str(&row.4).unwrap(); assert_eq!(stored.mime_type.as_str(), mime, "{mime}"); + assert_eq!(stored.size, bytes.len() as i64, "{mime}"); + assert_eq!(stored.digest.as_deref(), Some(digest.as_slice()), "{mime}"); + assert_eq!(stored.chunks.len(), 1, "{mime}"); + let stored_chunk = &stored.chunks[0]; + assert_eq!(stored_chunk.offset, output_chunk.offset, "{mime}"); + assert_eq!(stored_chunk.size, output_chunk.size, "{mime}"); assert_eq!( - stored.chunks[0].blob.blob().r#ref.as_str(), - output.file.chunks[0].blob.blob().r#ref.as_str(), + stored_chunk.blob.blob().r#ref.as_str(), + output_chunk.blob.blob().r#ref.as_str(), "{mime}" ); + assert_eq!(stored_chunk.blob.blob().mime_type.as_str(), mime, "{mime}"); + assert_eq!(stored_chunk.blob.blob().size, bytes.len(), "{mime}"); assert_eq!( load_upload_file(&state, &did(DID_A), &upload_id) .await diff --git a/src/publish.rs b/src/publish.rs index 6699fe6..3006ce3 100644 --- a/src/publish.rs +++ b/src/publish.rs @@ -73,6 +73,13 @@ enum UploadStatus { Error(String), } +#[derive(Clone, Copy, PartialEq, Eq)] +enum PreviewOutcome { + Complete, + Error, + Stale, +} + /// Whether the server-side draft is in sync with the in-memory composition. #[derive(Clone, Copy, PartialEq)] enum SaveState { @@ -184,8 +191,11 @@ struct Flow { /// Current preview operation token per stable model. Independent models do /// not cancel each other; stale completions must match their own run token. preview_generations: Signal>, - /// Monotonic token source for preview operations in this flow instance. - preview_operation_counter: Signal, + /// Current owner of the shared file-upload status. A stale operation may + /// retire or replace the status only while its token remains current. + upload_operation: Signal>, + /// Monotonic token source for asynchronous operations in this flow instance. + operation_counter: Signal, } #[component] @@ -230,7 +240,8 @@ fn PublishFlow(resume: String) -> Element { // The mounted name input element, focused once after the first upload. let name_element: Signal>> = use_signal(|| None); let preview_generations = use_signal(HashMap::new); - let preview_operation_counter = use_signal(|| 0_u64); + let upload_operation = use_signal(|| None); + let operation_counter = use_signal(|| 0_u64); let flow = Flow { client, @@ -253,7 +264,8 @@ fn PublishFlow(resume: String) -> Element { save_in_flight, name_element, preview_generations, - preview_operation_counter, + upload_operation, + operation_counter, }; // Resume a draft once on mount when `?draft=` is present; always load @@ -1068,7 +1080,7 @@ impl Flow { } drop(guard); let flow = premove_w.clone(); - spawn(async move { flow.recompute_model_preview(model_id).await; }); + spawn(async move { flow.recompute_model_preview(model_id, None).await; }); premove_w.clone().mark_dirty_and_save(); }, "Remove" @@ -2034,10 +2046,34 @@ impl Flow { } impl Flow { + fn next_operation_token(&self) -> u64 { + let mut counter = self.operation_counter; + let next = counter.read().saturating_add(1); + counter.set(next); + next + } + + fn begin_upload(&self, label: String) -> u64 { + let token = self.next_operation_token(); + let mut operation = self.upload_operation; + operation.set(Some(token)); + let mut upload = self.upload; + upload.set(UploadStatus::Uploading(label)); + token + } + + fn set_upload_if_current(&self, token: u64, status: UploadStatus) -> bool { + if self.upload_operation.read().as_ref() != Some(&token) { + return false; + } + let mut upload = self.upload; + upload.set(status); + true + } + /// Stage every dropped/picked file through one batch: ensure the default /// model + auto name, stage each file (stop at first failure), then save once. async fn handle_files_upload(self, files: Vec) { - let mut upload = self.upload; let first_name = files.first().map(|f| f.name()).unwrap_or_default(); { let mut composition = self.composition; @@ -2045,11 +2081,11 @@ impl Flow { ensure_default_model_for_upload(&mut g, &first_name); } let count = files.len(); - upload.set(UploadStatus::Uploading(if count == 1 { + let upload_token = self.begin_upload(if count == 1 { first_name.clone() } else { format!("{count} files") - })); + }); for file in files { let filename = file.name(); match file.read_bytes().await { @@ -2062,9 +2098,12 @@ impl Flow { ) { Ok(value) => value, Err(error) => { - upload.set(UploadStatus::Error(format!( - "Upload failed for {filename}: {error}" - ))); + self.set_upload_if_current( + upload_token, + UploadStatus::Error(format!( + "Upload failed for {filename}: {error}" + )), + ); let flow = self.clone(); flow.save_draft().await; return; @@ -2082,9 +2121,12 @@ impl Flow { format.clone(), &staged.file, ) else { - upload.set(UploadStatus::Error(format!( - "Upload failed for {filename}: staged metadata is incomplete" - ))); + self.set_upload_if_current( + upload_token, + UploadStatus::Error(format!( + "Upload failed for {filename}: staged metadata is incomplete" + )), + ); return; }; let part = DraftPartInput { @@ -2109,8 +2151,16 @@ impl Flow { let model_id = g.models.first().and_then(|model| model.model_id.clone()); drop(g); - if let Some(model_id) = model_id { - self.clone().recompute_model_preview(model_id).await; + if let Some(model_id) = model_id + && self + .clone() + .recompute_model_preview(model_id, Some(upload_token)) + .await + == PreviewOutcome::Error + { + let flow = self.clone(); + flow.save_draft().await; + return; } // Focus the name field once, on the first staged // file, so the user lands on the next likely edit. @@ -2123,9 +2173,12 @@ impl Flow { } } Err(error) => { - upload.set(UploadStatus::Error(format!( - "Upload failed for {filename}: {error}" - ))); + self.set_upload_if_current( + upload_token, + UploadStatus::Error(format!( + "Upload failed for {filename}: {error}" + )), + ); let flow = self.clone(); flow.save_draft().await; return; @@ -2133,14 +2186,17 @@ impl Flow { } } Err(_) => { - upload.set(UploadStatus::Error(format!("Could not read {filename}."))); + self.set_upload_if_current( + upload_token, + UploadStatus::Error(format!("Could not read {filename}.")), + ); let flow = self.clone(); flow.save_draft().await; return; } } } - upload.set(UploadStatus::Idle); + self.set_upload_if_current(upload_token, UploadStatus::Idle); let flow = self.clone(); flow.save_draft().await; } @@ -2154,26 +2210,23 @@ impl Flow { let Some(model) = guard.models.get_mut(model_index) else { return; }; - model.model_id.get_or_insert_with(|| { - let mut counter = self.preview_operation_counter; - let next = counter.read().saturating_add(1); - counter.set(next); - format!("pending-model:{next}") - }); + model + .model_id + .get_or_insert_with(|| format!("pending-model:{}", self.next_operation_token())); model .model_id .clone() .expect("model identity was installed") }; let filename = file.name(); - let mut upload = self.upload; - upload.set(UploadStatus::Uploading(filename.clone())); + let upload_token = self.begin_upload(filename.clone()); let bytes = match file.read_bytes().await { Ok(bytes) => bytes.to_vec(), Err(_) => { - upload.set(UploadStatus::Error( - "Could not read the selected file.".into(), - )); + self.set_upload_if_current( + upload_token, + UploadStatus::Error("Could not read the selected file.".into()), + ); return; } }; @@ -2181,7 +2234,10 @@ impl Flow { match classify_geometry_upload(file.content_type().as_deref(), &filename, &bytes) { Ok(value) => value, Err(error) => { - upload.set(UploadStatus::Error(format!("Upload failed: {error}"))); + self.set_upload_if_current( + upload_token, + UploadStatus::Error(format!("Upload failed: {error}")), + ); return; } }; @@ -2195,9 +2251,10 @@ impl Flow { format.clone(), &staged.file, ) else { - upload.set(UploadStatus::Error( - "Upload failed: staged metadata is incomplete".into(), - )); + self.set_upload_if_current( + upload_token, + UploadStatus::Error("Upload failed: staged metadata is incomplete".into()), + ); return; }; let part = DraftPartInput { @@ -2216,6 +2273,7 @@ impl Flow { .iter_mut() .find(|model| model.model_id.as_deref() == Some(model_id.as_str())) else { + self.set_upload_if_current(upload_token, UploadStatus::Idle); return; }; install_staged_resource(model, resource); @@ -2223,20 +2281,33 @@ impl Flow { model.parts.push(part); true }; - if installed { - self.clone().recompute_model_preview(model_id).await; + let outcome = if installed { + self.clone() + .recompute_model_preview(model_id, Some(upload_token)) + .await + } else { + PreviewOutcome::Stale + }; + if outcome != PreviewOutcome::Error { + self.set_upload_if_current(upload_token, UploadStatus::Idle); } - upload.set(UploadStatus::Idle); let flow = self.clone(); flow.save_draft().await; } Err(error) => { - upload.set(UploadStatus::Error(format!("Upload failed: {error}"))); + self.set_upload_if_current( + upload_token, + UploadStatus::Error(format!("Upload failed: {error}")), + ); } } } - async fn recompute_model_preview(mut self, model_id: String) { + async fn recompute_model_preview( + mut self, + model_id: String, + upload_token: Option, + ) -> PreviewOutcome { let draft_scope = self .draft_id .read() @@ -2249,17 +2320,17 @@ impl Flow { .iter() .find(|model| model.model_id.as_deref() == Some(model_id.as_str())) else { - return; + return PreviewOutcome::Stale; }; let Some(primary_id) = model.primary_resource_id.clone() else { - return; + return PreviewOutcome::Complete; }; let Some(primary) = model .staged_resources .iter() .find(|resource| resource.resource_id == primary_id) else { - return; + return PreviewOutcome::Complete; }; ( primary_id, @@ -2271,12 +2342,10 @@ impl Flow { preview_identity(&draft_scope, &model_id, &primary_id, &staged) else { self.clear_model_preview(&model_id, None); - return; + return PreviewOutcome::Complete; }; let operation_id = { - let mut counter = self.preview_operation_counter; - let next = counter.read().saturating_add(1); - counter.set(next); + let next = self.next_operation_token(); format!("{model_id}:{next}") }; self.preview_generations @@ -2287,12 +2356,12 @@ impl Flow { Err(error) => { tracing::debug!(%error, %model_id, "generated preview acquisition incomplete"); self.clear_model_preview(&model_id, Some(&operation_id)); - return; + return PreviewOutcome::Complete; } }; if !self.preview_is_current(&model_id, &operation_id) { self.retire_preview_operation(&model_id, &operation_id); - return; + return PreviewOutcome::Stale; } let png = if format_registry::format_for_extension(&primary_name) == Some(format_registry::FormatId::Ldraw) @@ -2300,7 +2369,7 @@ impl Flow { let DraftBundlePlan::Complete(bundle) = plan_draft_ldraw_bundle(&primary_id, &acquired) else { self.clear_model_preview(&model_id, Some(&operation_id)); - return; + return PreviewOutcome::Complete; }; render_preview_bundle(bundle).await } else { @@ -2309,53 +2378,74 @@ impl Flow { .find(|resource| resource.metadata.resource_id == primary_id) else { self.clear_model_preview(&model_id, Some(&operation_id)); - return; + return PreviewOutcome::Complete; }; render_preview(&primary_name, primary.bytes.clone()).await }; let Ok(png_bytes) = png else { self.clear_model_preview(&model_id, Some(&operation_id)); - return; + return PreviewOutcome::Complete; }; if !self.preview_is_current(&model_id, &operation_id) { self.retire_preview_operation(&model_id, &operation_id); - return; + return PreviewOutcome::Stale; } - let image = match upload_generated_preview( + if let Some(upload_token) = upload_token { + self.set_upload_if_current( + upload_token, + UploadStatus::Uploading(format!("preview of {primary_name}")), + ); + } + let completion = upload_generated_preview( self.client.clone(), primary_name.clone(), png_bytes, - self.upload, self.image_data_uris, - ) + ); + let mut composition = self.composition; + let current_preview = self.clone(); + let committed = match commit_preview_after(completion, |image| { + if !current_preview.preview_is_current(&model_id, &operation_id) { + return false; + } + let mut guard = composition.write(); + commit_generated_preview( + &mut guard, + &draft_scope, + &model_id, + &primary_id, + &composition_identity, + &operation_id, + image, + ) + }) .await { - Ok(image) => image, + Ok(committed) => committed, Err(error) => { + let current = self.preview_is_current(&model_id, &operation_id); self.clear_model_preview(&model_id, Some(&operation_id)); - let mut upload = self.upload; - upload.set(UploadStatus::Error(format!( - "Preview upload failed for {primary_name}: {error}" - ))); - return; + if current + && upload_token.is_some_and(|token| { + self.set_upload_if_current( + token, + UploadStatus::Error(format!( + "Preview upload failed for {primary_name}: {error}" + )), + ) + }) + { + return PreviewOutcome::Error; + } + return PreviewOutcome::Stale; } }; - if !self.preview_is_current(&model_id, &operation_id) { - self.retire_preview_operation(&model_id, &operation_id); - return; - } - let mut composition = self.composition; - let mut guard = composition.write(); - commit_generated_preview( - &mut guard, - &draft_scope, - &model_id, - &primary_id, - &composition_identity, - &operation_id, - image, - ); self.retire_preview_operation(&model_id, &operation_id); + if committed { + PreviewOutcome::Complete + } else { + PreviewOutcome::Stale + } } fn retire_preview_operation(&self, model_id: &str, operation_id: &str) -> bool { @@ -2667,6 +2757,13 @@ fn stable_model_id(primary_resource_id: &str) -> String { fn normalize_loaded_draft(composition: &mut DraftThingInput) { let mut used_ids = std::collections::HashSet::new(); + for model in &mut composition.models { + if let Some(model_id) = model.model_id.as_ref() + && !used_ids.insert(model_id.clone()) + { + model.model_id = None; + } + } for (index, model) in composition.models.iter_mut().enumerate() { let primary = model .primary_resource_id @@ -2696,15 +2793,16 @@ fn normalize_loaded_draft(composition: &mut DraftThingInput) { }); if let Some(primary) = primary { model.primary_resource_id = Some(primary.clone()); - let mut model_id = model - .model_id - .clone() - .unwrap_or_else(|| stable_model_id(&primary)); - if used_ids.contains(&model_id) { - model_id = format!("{}-{index}", stable_model_id(&primary)); + if model.model_id.is_none() { + let base = stable_model_id(&primary); + let mut candidate = base.clone(); + let mut suffix = index; + while !used_ids.insert(candidate.clone()) { + candidate = format!("{base}-{suffix}"); + suffix = suffix.saturating_add(1); + } + model.model_id = Some(candidate); } - used_ids.insert(model_id.clone()); - model.model_id = Some(model_id); } if let Some(generated) = model.generated_preview.as_mut() && generated.operation_id.is_none() @@ -2868,6 +2966,14 @@ fn clear_generated_preview( } } +async fn commit_preview_after(completion: F, commit: C) -> Result +where + F: std::future::Future>, + C: FnOnce(Image) -> bool, +{ + completion.await.map(commit) +} + fn commit_generated_preview( composition: &mut DraftThingInput, draft_scope: &str, @@ -2975,11 +3081,9 @@ async fn upload_generated_preview( client: PolymodelClient, filename: String, png_bytes: Vec, - mut upload: Signal, mut image_data_uris: Signal>, ) -> Result { let alt = format!("Rendered preview of {filename}"); - upload.set(UploadStatus::Uploading(format!("preview of {filename}"))); let image = draft_client::upload_image(&client, &alt, "image/png", png_bytes.clone()).await?; let cid = image.image.blob().r#ref.as_str().to_string(); image_data_uris @@ -3641,6 +3745,84 @@ mod tests { ); } + #[test] + fn legacy_draft_hydration_reserves_explicit_ids_and_loops_until_unique() { + let base = stable_model_id("resource-a"); + let mut composition = DraftThingInput { + models: vec![ + DraftModelInput { + primary_resource_id: Some("resource-a".into()), + staged_resources: vec![DraftStagedResource { + resource_id: "resource-a".into(), + upload_id: "upload-a".into(), + source_name: "a.stl".into(), + mime_type: "model/stl".into(), + digest: vec![1; 32], + byte_length: 1, + format: Some("stl".into()), + }], + ..Default::default() + }, + DraftModelInput { + model_id: Some(base.clone()), + primary_resource_id: Some("resource-b".into()), + staged_resources: vec![DraftStagedResource { + resource_id: "resource-b".into(), + upload_id: "upload-b".into(), + source_name: "b.stl".into(), + mime_type: "model/stl".into(), + digest: vec![2; 32], + byte_length: 1, + format: Some("stl".into()), + }], + ..Default::default() + }, + DraftModelInput { + model_id: Some(format!("{base}-0")), + primary_resource_id: Some("resource-c".into()), + staged_resources: vec![DraftStagedResource { + resource_id: "resource-c".into(), + upload_id: "upload-c".into(), + source_name: "c.stl".into(), + mime_type: "model/stl".into(), + digest: vec![3; 32], + byte_length: 1, + format: Some("stl".into()), + }], + ..Default::default() + }, + DraftModelInput { + model_id: Some(base.clone()), + primary_resource_id: Some("resource-d".into()), + staged_resources: vec![DraftStagedResource { + resource_id: "resource-d".into(), + upload_id: "upload-d".into(), + source_name: "d.stl".into(), + mime_type: "model/stl".into(), + digest: vec![4; 32], + byte_length: 1, + format: Some("stl".into()), + }], + ..Default::default() + }, + ], + ..Default::default() + }; + + normalize_loaded_draft(&mut composition); + + let ids = composition + .models + .iter() + .map(|model| model.model_id.as_deref().unwrap()) + .collect::>(); + assert_eq!(ids[0], format!("{base}-1")); + assert_eq!(ids[1], base); + assert_eq!(ids[2], format!("{base}-0")); + assert_ne!(ids[3], ids[1]); + assert_eq!(ids.iter().copied().collect::>().len(), ids.len()); + } + #[test] fn removed_model_cannot_receive_stale_preview_after_vector_compaction() { let generated = test_image("stale generated"); @@ -3691,6 +3873,74 @@ mod tests { assert!(composition.previews.is_none()); } + #[tokio::test] + async fn in_flight_removed_model_cannot_receive_preview_after_vector_compaction() { + let generated = test_image("stale async generated"); + let resource = DraftStagedResource { + resource_id: "resource-a".into(), + upload_id: "upload-a".into(), + source_name: "a.stl".into(), + mime_type: "model/stl".into(), + digest: vec![7; 32], + byte_length: 12, + format: Some("stl".into()), + }; + let composition = std::sync::Arc::new(std::sync::Mutex::new(DraftThingInput { + models: vec![ + DraftModelInput { + model_id: Some("model-a".into()), + primary_resource_id: Some("resource-a".into()), + staged_resources: vec![resource.clone()], + parts: vec![DraftPartInput { + resource_id: Some("resource-a".into()), + ..Default::default() + }], + ..Default::default() + }, + DraftModelInput { + model_id: Some("model-b".into()), + primary_resource_id: Some("resource-b".into()), + ..Default::default() + }, + ], + ..Default::default() + })); + let identity = preview_identity("draft", "model-a", "resource-a", &[resource]).unwrap(); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let commit_composition = composition.clone(); + let completion = async move { + started_tx.send(()).unwrap(); + release_rx.await.unwrap(); + Ok::<_, ()>(generated) + }; + let commit = commit_preview_after(completion, move |image| { + commit_generated_preview( + &mut commit_composition.lock().unwrap(), + "draft", + "model-a", + "resource-a", + &identity, + "model-a:1", + image, + ) + }); + tokio::pin!(commit); + assert!(futures::poll!(&mut commit).is_pending()); + started_rx.await.unwrap(); + composition.lock().unwrap().models.remove(0); + release_tx.send(()).unwrap(); + + assert!(!commit.await.unwrap()); + let composition = composition.lock().unwrap(); + assert_eq!(composition.models[0].model_id.as_deref(), Some("model-b")); + assert!(composition.models[0].generated_preview.is_none()); + assert!(composition.models[0].previews.is_none()); + assert!(composition.models[0].parts.is_empty()); + assert!(composition.previews.is_none()); + assert!(composition.cover.is_none()); + } + #[test] fn generated_preview_cleanup_preserves_user_cover_and_normalizes_empty_lists() { let generated = test_image("generated"); diff --git a/src/viewer.rs b/src/viewer.rs index 540f901..e9ee0d7 100644 --- a/src/viewer.rs +++ b/src/viewer.rs @@ -689,14 +689,37 @@ async fn classify_verified_source(source: &ViewerMeshSource) -> Result() + .map_err(|_| "part file body has no readable stream".to_string())?; + let mut bytes = Vec::with_capacity(expected_size as usize); + loop { + let result = JsFuture::from(reader.read()) + .await + .map_err(|error| format!("part file body failed: {error:?}"))?; + let done = js_sys::Reflect::get(&result, &wasm_bindgen::JsValue::from_str("done")) + .map_err(|error| format!("part file stream state failed: {error:?}"))? + .as_bool() + .unwrap_or(false); + if done { + break; + } + let value = js_sys::Reflect::get(&result, &wasm_bindgen::JsValue::from_str("value")) + .map_err(|error| format!("part file stream chunk failed: {error:?}"))?; + let chunk = js_sys::Uint8Array::new(&value); + let chunk_len = chunk.length() as usize; + if bytes.len().saturating_add(chunk_len) > MAX_EVIDENCE_BYTES { + let _ = reader.cancel(); + return Err("verified file exceeds the bounded format evidence limit".into()); + } + let offset = bytes.len(); + bytes.resize(offset + chunk_len, 0); + chunk.copy_to(&mut bytes[offset..]); + } verify_part_file(&bytes, expected_size, source.expected_digest.as_deref()) .map_err(|error| format!("part file verification failed: {error}"))?; viewer_format_from_evidence(&bytes)