diff --git a/crates/polymodel-renderer-worker/src/worker.rs b/crates/polymodel-renderer-worker/src/worker.rs index 01ffe99..2533aa2 100644 --- a/crates/polymodel-renderer-worker/src/worker.rs +++ b/crates/polymodel-renderer-worker/src/worker.rs @@ -4,7 +4,7 @@ //! `requestAnimationFrame` frame loop, the `OffscreenCanvas` GL context, mesh loading, //! and the postcard message protocol bridge to the Dioxus main thread. -use std::cell::RefCell; +use std::cell::{Cell, RefCell}; use std::rc::Rc; use std::sync::Arc; @@ -1182,7 +1182,12 @@ async fn load_mesh_async( let expected_digest = expected_digest.clone(); async move { tracing::debug!(%url, verify_primary, "renderer worker: fetching mesh resource"); - let bytes = fetch_bytes(&url, None).await.map_err(|e| { + let expected_bytes = if verify_primary { + expected_size.and_then(|size| u64::try_from(size).ok()) + } else { + None + }; + let bytes = fetch_bytes(&url, None, expected_bytes).await.map_err(|e| { polymodel_mesh::contract::MeshLoadError::Decode(format!("fetch {url}: {e}")) })?; tracing::debug!( @@ -1403,7 +1408,7 @@ async fn fetch_descriptor_bytes( expected_sha256: &[u8], abort: Option<&AbortController>, ) -> Result, String> { - let bytes = fetch_bytes(route, abort).await?; + let bytes = fetch_bytes(route, abort, Some(expected_length)).await?; let expected_size = i64::try_from(expected_length) .map_err(|_| "compound descriptor byte length is too large".to_string())?; polymodel_renderer_protocol::verify_part_file(&bytes, expected_size, Some(expected_sha256)) @@ -1518,7 +1523,8 @@ async fn render_preview_request( drop(reservation); return Err("preview load cancelled".into()); } - let model = load_mesh_async_with_abort(primary_url, format, abort.as_ref()).await?; + let model = + load_mesh_async_with_abort(primary_url, format, fetch_bytes, abort.as_ref()).await?; if abort .as_ref() .is_some_and(|controller| controller.signal().aborted()) @@ -1543,15 +1549,32 @@ async fn render_preview_request( async fn load_mesh_async_with_abort( primary_url: &str, format: MeshFormat, + fetch_limit: u64, abort: Option<&AbortController>, ) -> Result { + let remaining = Rc::new(Cell::new(fetch_limit)); let model = load_mesh_with_companions(primary_url, format, |url| { let route = url.clone(); let abort = abort.cloned(); + let remaining = remaining.clone(); async move { - let bytes = fetch_bytes(&route, abort.as_ref()).await.map_err(|e| { - polymodel_mesh::contract::MeshLoadError::Decode(format!("fetch {route}: {e}")) + let limit = remaining.get(); + if limit == 0 { + return Err(polymodel_mesh::contract::MeshLoadError::Decode( + "preview fetch budget exhausted".into(), + )); + } + let bytes = fetch_bytes(&route, abort.as_ref(), Some(limit)) + .await + .map_err(|e| { + polymodel_mesh::contract::MeshLoadError::Decode(format!("fetch {route}: {e}")) + })?; + let fetched = u64::try_from(bytes.len()).map_err(|_| { + polymodel_mesh::contract::MeshLoadError::Decode( + "preview fetched byte count overflowed".into(), + ) })?; + remaining.set(limit.saturating_sub(fetched)); Ok(bytes) } }) @@ -1651,7 +1674,12 @@ async fn offscreen_canvas_png_bytes(canvas: &web_sys::OffscreenCanvas) -> Result Ok(js_sys::Uint8Array::new(&buffer).to_vec()) } -async fn fetch_bytes(url: &str, abort: Option<&AbortController>) -> Result, String> { +async fn fetch_bytes( + url: &str, + abort: Option<&AbortController>, + expected_bytes: Option, +) -> Result, String> { + let limit = expected_bytes.unwrap_or(MAX_UNKNOWN_FETCH_BYTES); let opts = web_sys::RequestInit::new(); opts.set_method("GET"); opts.set_mode(web_sys::RequestMode::SameOrigin); @@ -1661,61 +1689,141 @@ async fn fetch_bytes(url: &str, abort: Option<&AbortController>) -> Result() .map_err(|_| "fetch: not a Promise".to_string())?; - - let response = wasm_bindgen_futures::JsFuture::from(fetch_promise) + let response = JsFuture::from(fetch_promise) .await .map_err(|e| format!("fetch: {:?}", e))? .dyn_into::() .map_err(|_| "fetch: not a Response".to_string())?; + if !response.ok() { let status = response.status(); - let status_text = response.status_text(); - let bounded_status = status_text.chars().take(96).collect::(); - let body = response - .text() - .ok() - .map(wasm_bindgen_futures::JsFuture::from) - .map(|future| async move { future.await.ok().and_then(|value| value.as_string()) }); - let bounded_body = match body { - Some(future) => future - .await - .unwrap_or_default() - .chars() - .take(512) - .collect::(), - None => String::new(), - }; + let status_text = response.status_text().chars().take(96).collect::(); + let bounded_body = bounded_error_body(&response).await; let detail = if bounded_body.is_empty() { - format!("fetch returned HTTP {status} {bounded_status}") + format!("fetch returned HTTP {status} {status_text}") } else { - format!("fetch returned HTTP {status} {bounded_status}: {bounded_body}") + format!("fetch returned HTTP {status} {status_text}: {bounded_body}") }; return Err(detail); } - let array_buffer_promise = response - .array_buffer() - .map_err(|e| format!("array_buffer: {:?}", e))?; - let array_buffer = wasm_bindgen_futures::JsFuture::from(array_buffer_promise) - .await - .map_err(|e| format!("array_buffer: {:?}", e))?; + let content_length = response + .headers() + .get("content-length") + .ok() + .flatten() + .and_then(|value| value.parse::().ok()); + if let Some(content_length) = content_length + && content_length > limit + { + return Err(format!( + "response exceeds fetch bound: {content_length} bytes > {limit}" + )); + } - let uint8 = js_sys::Uint8Array::new(&array_buffer); - Ok(uint8.to_vec()) + let body = js_sys::Reflect::get(&response, &"body".into()) + .map_err(|e| format!("response body: {e:?}"))?; + if body.is_null() || body.is_undefined() { + return Err("response has no readable body".into()); + } + let reader = js_sys::Reflect::get(&body, &"getReader".into()) + .map_err(|e| format!("response reader: {e:?}"))? + .dyn_into::() + .map_err(|_| "response body has no reader".to_string())? + .call0(&body) + .map_err(|e| format!("response reader: {e:?}"))?; + let read = js_sys::Reflect::get(&reader, &"read".into()) + .map_err(|e| format!("response reader read: {e:?}"))? + .dyn_into::() + .map_err(|_| "response reader has no read method".to_string())?; + let mut bytes = Vec::with_capacity( + usize::try_from(content_length.unwrap_or(0).min(limit)).unwrap_or_default(), + ); + loop { + let result = read + .call0(&reader) + .map_err(|e| format!("response reader read: {e:?}"))? + .dyn_into::() + .map_err(|_| "response reader read did not return a Promise".to_string())?; + let result = JsFuture::from(result) + .await + .map_err(|e| format!("response body: {e:?}"))?; + if js_sys::Reflect::get(&result, &"done".into()) + .ok() + .and_then(|done| done.as_bool()) + .unwrap_or(false) + { + break; + } + let value = js_sys::Reflect::get(&result, &"value".into()) + .map_err(|e| format!("response body chunk: {e:?}"))?; + let chunk = js_sys::Uint8Array::new(&value); + let next_len = bytes + .len() + .checked_add(chunk.length() as usize) + .ok_or_else(|| "response exceeds fetch bound".to_string())?; + if next_len > usize::try_from(limit).unwrap_or(usize::MAX) { + return Err(format!( + "response exceeds fetch bound: more than {limit} bytes" + )); + } + bytes.extend_from_slice(&chunk.to_vec()); + } + Ok(bytes) +} + +async fn bounded_error_body(response: &web_sys::Response) -> String { + const MAX_ERROR_BODY_BYTES: u64 = 2048; + let body = js_sys::Reflect::get(response, &"body".into()).ok(); + let Some(body) = body.filter(|body| !body.is_null() && !body.is_undefined()) else { + return String::new(); + }; + let Ok(reader) = js_sys::Reflect::get(&body, &"getReader".into()) + .and_then(|function| function.dyn_into::()) + .and_then(|function| function.call0(&body)) + else { + return String::new(); + }; + let Ok(read) = js_sys::Reflect::get(&reader, &"read".into()) + .and_then(|function| function.dyn_into::()) + else { + return String::new(); + }; + let mut bytes = Vec::new(); + while bytes.len() < MAX_ERROR_BODY_BYTES as usize { + let Ok(promise) = read + .call0(&reader) + .and_then(|value| value.dyn_into::()) + else { + break; + }; + let Ok(result) = JsFuture::from(promise).await else { + break; + }; + if js_sys::Reflect::get(&result, &"done".into()) + .ok() + .and_then(|done| done.as_bool()) + .unwrap_or(false) + { + break; + } + let Ok(value) = js_sys::Reflect::get(&result, &"value".into()) else { + break; + }; + let chunk = js_sys::Uint8Array::new(&value); + let remaining = MAX_ERROR_BODY_BYTES as usize - bytes.len(); + let chunk = chunk.to_vec(); + bytes.extend_from_slice(&chunk[..chunk.len().min(remaining)]); + } + String::from_utf8_lossy(&bytes).chars().take(512).collect() } // --------------------------------------------------------------------------- diff --git a/migrations/012_pm80_ldraw_resource_identity.sql b/migrations/012_pm80_ldraw_resource_identity.sql new file mode 100644 index 0000000..558cf45 --- /dev/null +++ b/migrations/012_pm80_ldraw_resource_identity.sql @@ -0,0 +1,48 @@ +-- PM-80 G28: project/path is membership metadata, not source identity. +-- Rebuild the table created by migration 009 without its obsolete compound +-- uniqueness constraint. resource_uri remains the sole source key; all +-- project/path visibility is represented by ldraw_resource_memberships. +ALTER TABLE ldraw_resources RENAME TO ldraw_resources_pm80_old; +DROP INDEX idx_ldraw_resources_owner; +DROP INDEX idx_ldraw_resources_project_path; +DROP INDEX idx_ldraw_resources_source; + +CREATE TABLE ldraw_resources ( + resource_uri TEXT PRIMARY KEY, + owner_did TEXT NOT NULL, + project_uri TEXT NOT NULL, + record_cid TEXT NOT NULL, + rkey TEXT NOT NULL, + root TEXT NOT NULL, + canonical_path TEXT NOT NULL, + mime_type TEXT NOT NULL, + byte_length INTEGER NOT NULL CHECK (byte_length >= 0), + sha256 BLOB NOT NULL CHECK (length(sha256) = 32), + ordered_blob_cids TEXT NOT NULL, + provenance TEXT NOT NULL, + licence TEXT NOT NULL, + notices_json TEXT NOT NULL, + snapshot_id TEXT, + legacy INTEGER NOT NULL DEFAULT 0 CHECK (legacy IN (0, 1)), + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL +); + +INSERT INTO ldraw_resources ( + resource_uri, owner_did, project_uri, record_cid, rkey, root, + canonical_path, mime_type, byte_length, sha256, ordered_blob_cids, + provenance, licence, notices_json, snapshot_id, legacy, created_at, updated_at +) +SELECT + resource_uri, owner_did, project_uri, record_cid, rkey, root, + canonical_path, mime_type, byte_length, sha256, ordered_blob_cids, + provenance, licence, notices_json, snapshot_id, legacy, created_at, updated_at +FROM ldraw_resources_pm80_old; + +DROP TABLE ldraw_resources_pm80_old; + +CREATE INDEX idx_ldraw_resources_owner ON ldraw_resources(owner_did); +CREATE INDEX idx_ldraw_resources_project_path + ON ldraw_resources(project_uri, root, canonical_path); +CREATE INDEX idx_ldraw_resources_source + ON ldraw_resources(resource_uri, sha256, byte_length); diff --git a/src/indexing/projection.rs b/src/indexing/projection.rs index 041caa1..14fb4ea 100644 --- a/src/indexing/projection.rs +++ b/src/indexing/projection.rs @@ -2096,6 +2096,71 @@ mod tests { assert_eq!(resolved.source.resource_uri.as_ref(), part_uri); assert_eq!(resolved.target.blob_cids, vec![cid.to_owned()]); + // Two distinct Parts may expose the same project/root/path. The source + // rows must both commit, and the resolver must report an ambiguity rather + // than poisoning the transaction or silently choosing one source. + let duplicate_part_uri = format!("at://{did}/space.polymodel.library.part/duplicate"); + project_event( + &db, + &ProjectionInput::Record(event( + 4, + "space.polymodel.library.part", + "duplicate", + "create", + Some(part_record("main.ldr")), + Some("bafyreduplicate"), + )), + ) + .await + .unwrap(); + project_event( + &db, + &ProjectionInput::Record(event( + 5, + "space.polymodel.library.part", + "duplicate", + "update", + Some(part_record("main.ldr")), + Some("bafyreduplicateupdated"), + )), + ) + .await + .unwrap(); + project_event( + &db, + &ProjectionInput::Record(event( + 6, + "space.polymodel.library.model", + "model", + "update", + Some( + to_data(&json!({ + "name": "Main model", + "createdAt": "2024-01-01T00:00:00.000Z", + "parts": [ + { "uri": part_uri, "cid": "bafyreipart" }, + { "uri": duplicate_part_uri, "cid": "bafyreduplicateupdated" } + ] + })) + .unwrap(), + ), + Some("bafyremodelupdated"), + )), + ) + .await + .unwrap(); + let duplicate_count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM ldraw_resources WHERE provenance = 'ordinary-part'", + ) + .fetch_one(&db) + .await + .unwrap(); + assert_eq!(duplicate_count, 2); + assert!(matches!( + resolver.resolve(&project, &path).await, + Err(crate::ldraw::ResolveError::Collision { .. }) + )); + // A second Thing can share the same Part source. The source row remains // singular while each project gets its own visible membership/path. let second_did = "did:plc:other"; @@ -2129,6 +2194,16 @@ mod tests { .fetch_one(&db) .await .unwrap(), + 4 + ); + assert_eq!( + sqlx::query_scalar::<_, i64>( + "SELECT COUNT(*) FROM ldraw_resource_memberships WHERE provenance = 'ordinary-part' AND project_uri = ?", + ) + .bind(&thing_uri) + .fetch_one(&db) + .await + .unwrap(), 2 ); let second_project = AtUri::new_owned(&second_thing_uri).unwrap(); @@ -2157,7 +2232,7 @@ mod tests { .await .unwrap(); let mut update = create_part.clone(); - update.seq = 5; + update.seq = 7; update.action = "update".to_owned(); update.cid = Some("bafyrepartupdated".to_owned()); update.record = Some(part_record("updated.ldr")); @@ -2171,9 +2246,9 @@ mod tests { .fetch_one(&db) .await .unwrap(), - 1 + 2 ); - assert!(resolver.resolve(&project, &path).await.is_err()); + assert!(resolver.resolve(&project, &path).await.is_ok()); let updated_path = CanonicalPath::parse("models/updated.ldr").unwrap(); assert!(resolver.resolve(&project, &updated_path).await.is_ok()); assert!( @@ -2184,7 +2259,9 @@ mod tests { ); // A stale replay cannot restore the old source identity or path. - project_event(&db, &ProjectionInput::Record(create_part.clone())) + let mut stale_replay = create_part.clone(); + stale_replay.seq = 3; + project_event(&db, &ProjectionInput::Record(stale_replay)) .await .unwrap(); assert!(resolver.resolve(&project, &updated_path).await.is_ok()); @@ -2194,7 +2271,7 @@ mod tests { project_event( &db, &ProjectionInput::Record(event( - 6, + 8, "space.polymodel.library.part", "part", "update", @@ -2218,18 +2295,23 @@ mod tests { .fetch_one(&db) .await .unwrap(), - 0 + 2 ); project_event( &db, - &ProjectionInput::Record(delete_event(7, did, "space.polymodel.library.part", "part")), + &ProjectionInput::Record(delete_event(9, did, "space.polymodel.library.part", "part")), ) .await .unwrap(); project_event( &db, - &ProjectionInput::Record(delete_event(8, did, "space.polymodel.library.part", "part")), + &ProjectionInput::Record(delete_event( + 10, + did, + "space.polymodel.library.part", + "duplicate", + )), ) .await .unwrap(); diff --git a/src/viewer.rs b/src/viewer.rs index f18a75d..f4e681c 100644 --- a/src/viewer.rs +++ b/src/viewer.rs @@ -2602,24 +2602,6 @@ mod tests { assert_eq!(context_acquisitions_after_event(1, &other), 1); } - #[test] - fn dispose_command_sent_on_drop() { - // The WorkerBridgeState::drop sends Dispose + terminates the worker. - // This test verifies the command type exists and serializes in a session envelope. - let bytes = polymodel_renderer_protocol::serialize_command_envelope( - &polymodel_renderer_protocol::CommandEnvelope { - version: polymodel_renderer_protocol::PROTOCOL_VERSION, - session_id: 7, - payload: polymodel_renderer_protocol::CommandPayload::Command( - RendererCommand::Dispose, - ), - }, - ) - .expect("serialize"); - let cmd = deserialize_command_at(&bytes); - assert!(matches!(cmd, RendererCommand::Dispose)); - } - fn deserialize_command_at(bytes: &[u8]) -> RendererCommand { match polymodel_renderer_protocol::deserialize_command_envelope(bytes) .expect("deserialize")