diff --git a/seaweed-volume/src/images.rs b/seaweed-volume/src/images.rs index 9ad7ca71c8..2125f433da 100644 --- a/seaweed-volume/src/images.rs +++ b/seaweed-volume/src/images.rs @@ -143,12 +143,12 @@ fn flip_horizontal_if(img: DynamicImage, do_flip: bool) -> DynamicImage { DynamicImage::ImageRgba8(out) } -/// Returns true if the given MIME type or file path extension indicates a JPEG file. -pub fn is_jpeg(mime_type: &str, path: &str) -> bool { +/// Returns true if the given MIME type or file name extension indicates a JPEG file. +pub fn is_jpeg(mime_type: &str, filename: &str) -> bool { if mime_type == "image/jpeg" { return true; } - let lower = path.to_lowercase(); + let lower = filename.to_lowercase(); lower.ends_with(".jpg") || lower.ends_with(".jpeg") } diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 91c2d2a7ca..e4c150c5ea 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -695,6 +695,27 @@ async fn batch_delete_file_ids( Ok(()) } +/// `query` with the fields named in `overrides` replaced by their values. +fn override_query_fields(query: &str, overrides: &[(&str, String)]) -> Result { + let mut fields = Vec::new(); + for field in query.split('&').filter(|field| !field.is_empty()) { + let decoded: Vec<(String, String)> = serde_urlencoded::from_str(field) + .map_err(|e| format!("parse replication query: {e}"))?; + if !decoded + .iter() + .any(|(name, _)| overrides.iter().any(|(o, _)| o == name)) + { + fields.push(field); + } + } + let encoded = serde_urlencoded::to_string(overrides) + .map_err(|e| format!("encode replication query: {e}"))?; + if !encoded.is_empty() { + fields.push(&encoded); + } + Ok(fields.join("&")) +} + /// Helper to synchronously replicate a request to peer volume servers. async fn do_replicated_request( state: &VolumeServerState, @@ -756,16 +777,7 @@ async fn do_replicated_request( return Ok(()); } - let mut query_fields = Vec::new(); - for field in query.split('&').filter(|field| !field.is_empty()) { - let decoded: Vec<(String, String)> = serde_urlencoded::from_str(field) - .map_err(|e| format!("parse replication query: {e}"))?; - if !decoded.iter().any(|(name, _)| name == "type") { - query_fields.push(field); - } - } - query_fields.push("type=replicate"); - let new_query = query_fields.join("&"); + let new_query = override_query_fields(query, &[("type", "replicate".to_string())])?; let mut futures = Vec::new(); for loc in remote_locations { @@ -793,6 +805,12 @@ async fn do_replicated_request( if let Some(auth) = headers.get(axum::http::header::AUTHORIZATION) { req_builder = req_builder.header(axum::http::header::AUTHORIZATION, auth); } + for (name, value) in headers + .iter() + .filter(|(name, _)| name.as_str().starts_with("seaweed-")) + { + req_builder = req_builder.header(name, value); + } if let Some(ref b) = body { req_builder = req_builder.body(b.clone()); @@ -2855,8 +2873,12 @@ pub async fn post_handler( while let Ok(Some(field)) = multipart.next_field().await { let field_name = field.name().map(|s| s.to_string()); let fname = field.file_name().map(clean_windows_path_base); - let fct = field.content_type().map(|m| m.to_string()); let field_headers = field.headers().clone(); + let fct = field_headers + .get(header::CONTENT_TYPE) + .and_then(|v| v.to_str().ok()) + .filter(|s| !s.is_empty()) + .map(|s| s.to_string()); let fce = field_headers .get(header::CONTENT_ENCODING) .and_then(|v| v.to_str().ok()) @@ -3022,15 +3044,18 @@ pub async fn post_handler( let mime_type = if let Some(ref pct) = parsed_content_type { pct.clone() } else { - let multipart_fallback = - if should_parse_multipart && !filename.is_empty() && !is_chunk_manifest { - mime_guess::from_path(&filename) - .first() - .map(|m| m.to_string()) - .unwrap_or_default() - } else { - String::new() - }; + let multipart_fallback = if should_parse_multipart + && !filename.is_empty() + && !is_chunk_manifest + && !is_replicate + { + mime_guess::from_path(&filename) + .first() + .map(|m| m.to_string()) + .unwrap_or_default() + } else { + String::new() + }; headers .get(header::CONTENT_TYPE) .and_then(|v| v.to_str().ok()) @@ -3093,15 +3118,19 @@ pub async fn post_handler( .collect(); // Fix JPEG orientation from EXIF data before storing (matches Go behavior). - let body_data = if state.fix_jpg_orientation && crate::images::is_jpeg(&mime_type, &path) { - crate::images::fix_jpg_orientation(&body_data_raw) - } else { - body_data_raw - }; + let (body_data, body_rewritten) = + if state.fix_jpg_orientation && crate::images::is_jpeg(&mime_type, &filename) { + let fixed = crate::images::fix_jpg_orientation(&body_data_raw); + let rewritten = fixed != body_data_raw; + (fixed, rewritten) + } else { + (body_data_raw, false) + }; // Auto-compress compressible file types (matches Go's IsCompressableFileType). // Only compress if not already gzipped and compression saves >10%. - let (final_data, final_is_gzipped) = if !is_gzipped && !is_chunk_manifest { + // A replica keeps the compression state the primary stored. + let (final_data, final_is_gzipped) = if !is_gzipped && !is_chunk_manifest && !is_replicate { let ext = filename .rfind('.') .map(|p| &filename[p..]) @@ -3194,23 +3223,55 @@ pub async fn post_handler( }; // The fan-out runs concurrently with the local needle write, like Go's - // ReplicatedWrite: it reads the already-buffered request body, so the two - // sides share no mutable state. The ack still waits for both. + // ReplicatedWrite: replicas get the stored bytes with their compression + // state, so they keep it instead of re-deriving it. The ack waits for both. let replication = if needs_replication { let state_clone = state.clone(); let path_clone = path.clone(); let query_clone = query.clone(); - let headers_clone = headers.clone(); - let body_clone = body.clone(); + let mut overrides = vec![("ts", n.last_modified.to_string())]; + if let Some(t) = ttl.filter(|t| !t.is_empty()) { + overrides.push(("ttl", t.to_string())); + } + if is_chunk_manifest { + overrides.push(("cm", "true".to_string())); + } + if fsync { + overrides.push(("fsync", "true".to_string())); + } + let mut replica_headers = HeaderMap::new(); + if let Some(auth) = headers.get(header::AUTHORIZATION) { + replica_headers.insert(header::AUTHORIZATION, auth.clone()); + } + if n.has_pairs() { + for (key, value) in &pair_map { + if let (Ok(name), Ok(value)) = ( + header::HeaderName::try_from(format!("seaweed-{key}")), + header::HeaderValue::from_str(value), + ) { + replica_headers.insert(name, value); + } + } + } + let (content_type, replica_body) = replica_upload_body( + &filename, + &mime_type, + n.is_compressed().then_some("gzip"), + original_content_md5.as_deref().filter(|_| !body_rewritten), + &n.data, + ); + if let Ok(value) = header::HeaderValue::from_str(&content_type) { + replica_headers.insert(header::CONTENT_TYPE, value); + } Some(tokio::spawn(async move { do_replicated_request( &state_clone, vid.0, Method::POST, &path_clone, - &query_clone, - &headers_clone, - Some(body_clone), + &override_query_fields(&query_clone, &overrides)?, + &replica_headers, + Some(replica_body), ) .await })) @@ -4782,6 +4843,41 @@ fn is_compressible_file_type(ext: &str, mtype: &str) -> bool { false } +/// One-part multipart upload of a stored needle, the form Go's upload_content +/// sends: the bytes as stored, labeled with their content encoding. +fn replica_upload_body( + filename: &str, + mime: &str, + content_encoding: Option<&str>, + content_md5: Option<&str>, + data: &[u8], +) -> (String, bytes::Bytes) { + let boundary = uuid::Uuid::new_v4().simple().to_string(); + let mut head = format!( + "--{boundary}\r\nContent-Disposition: form-data; name=\"file\"; filename=\"{}\"\r\n", + filename.replace('\\', "\\\\").replace('"', "\\\"") + ); + if !mime.is_empty() { + head.push_str(&format!("Content-Type: {mime}\r\n")); + } + if let Some(encoding) = content_encoding { + head.push_str(&format!("Content-Encoding: {encoding}\r\n")); + } + if let Some(md5) = content_md5 { + head.push_str(&format!("Content-MD5: {md5}\r\n")); + } + head.push_str("\r\n"); + let tail = format!("\r\n--{boundary}--\r\n"); + let mut body = Vec::with_capacity(head.len() + data.len() + tail.len()); + body.extend_from_slice(head.as_bytes()); + body.extend_from_slice(data); + body.extend_from_slice(tail.as_bytes()); + ( + format!("multipart/form-data; boundary={boundary}"), + body.into(), + ) +} + /// Try to gzip data. Returns None on error. fn try_gzip_data(data: &[u8]) -> Option> { use flate2::Compression; @@ -8228,12 +8324,15 @@ mod tests { ); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); *state.current_master_url.write().await = listener.local_addr().unwrap().to_string(); - let master_server = tokio::spawn(async move { axum::serve(listener, master).await.unwrap() }); + let master_server = + tokio::spawn(async move { axum::serve(listener, master).await.unwrap() }); let id = 0x6e7a_0c05u64; let fid = format!("1,{id:x}{TEST_COOKIE:08x}"); let token = gen_jwt(&key, 60, &fid).unwrap(); - let query = "%6Awt=expired&tag=a%2Fb+%2B&tag=x%20y&bare&%74ype=client&type=old"; - let expected_query = "%6Awt=expired&tag=a%2Fb+%2B&tag=x%20y&bare&type=replicate"; + let query = + "%6Awt=expired&tag=a%2Fb+%2B&tag=x%20y&bare&ts=1700000000&%74ype=client&type=old"; + let expected_query = + "%6Awt=expired&tag=a%2Fb+%2B&tag=x%20y&bare&ts=1700000000&type=replicate"; let payload = b"authenticated replica bytes"; for method in [Method::POST, Method::DELETE] { let response = tokio::time::timeout( @@ -8282,6 +8381,264 @@ mod tests { peer_server.abort(); } + #[tokio::test] + async fn http_replica_stores_the_primary_needle() { + use crate::storage::super_block::ReplicaPlacement; + use axum::{Json, Router, routing::get}; + use tower::ServiceExt; + + let tmp = tempfile::TempDir::new().unwrap(); + let mut state = volume_test_state(&tmp); + Arc::get_mut(&mut state).unwrap().self_url = "127.0.0.1:8080".to_string(); + let volume = state + .store + .read() + .unwrap() + .find_volume(VolumeId(1)) + .unwrap() + .1; + volume + .set_replica_placement(ReplicaPlacement::from_string("001").unwrap()) + .unwrap(); + let peer_tmp = tempfile::TempDir::new().unwrap(); + let peer = volume_test_state(&peer_tmp); + let peer_volume = peer + .store + .read() + .unwrap() + .find_volume(VolumeId(1)) + .unwrap() + .1; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let peer_addr = listener.local_addr().unwrap(); + let peer_server = tokio::spawn(async move { + axum::serve( + listener, + super::super::volume_server::build_admin_router(peer), + ) + .await + .unwrap(); + }); + let master = Router::new().route( + "/dir/lookup", + get(move || async move { + Json(serde_json::json!({"locations": [ + {"url": "127.0.0.1:8080"}, + {"url": peer_addr.to_string()} + ]})) + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + *state.current_master_url.write().await = listener.local_addr().unwrap().to_string(); + let master_server = + tokio::spawn(async move { axum::serve(listener, master).await.unwrap() }); + + let payload = "compressible replica text ".repeat(64); + let boundary = "replica-boundary"; + let multipart = format!( + "--{boundary}\r\nContent-Disposition: form-data; name=\"ttl\"\r\n\r\n3m\r\n\ + --{boundary}\r\nContent-Disposition: form-data; name=\"file\"; filename=\"a \\\"b\\\".txt\"\r\n\ + Content-Type: text/plain\r\n\r\n{payload}\r\n--{boundary}--\r\n" + ); + let cases = [ + ( + 0x6e7a_0c06u64, + Request::builder() + .header(header::CONTENT_TYPE, "text/plain") + .header( + header::CONTENT_DISPOSITION, + "attachment; filename=\"notes.txt\"", + ) + .header("Content-MD5", compute_md5_base64(payload.as_bytes())) + .header("Seaweed-Color", "blue"), + Body::from(payload.clone()), + ), + ( + 0x6e7a_0c07u64, + Request::builder().header( + header::CONTENT_TYPE, + format!("multipart/form-data; boundary={boundary}"), + ), + Body::from(multipart), + ), + ( + 0x6e7a_0c0au64, + Request::builder().header( + header::CONTENT_DISPOSITION, + "attachment; filename=\"untyped.txt\"", + ), + Body::from(payload.clone()), + ), + ( + 0x6e7a_0c0bu64, + Request::builder().header( + header::CONTENT_DISPOSITION, + format!("attachment; filename=\"{}.txt\"", "n".repeat(300)), + ), + Body::from(payload.clone()), + ), + ( + 0x6e7a_0c0eu64, + Request::builder() + .header(header::CONTENT_TYPE, "Text/Plain; charset=UTF-8") + .header( + header::CONTENT_DISPOSITION, + "attachment; filename=\"upper.txt\"", + ), + Body::from(payload.clone()), + ), + ]; + for (id, request, body) in cases { + let response = tokio::time::timeout( + std::time::Duration::from_secs(5), + super::super::volume_server::build_admin_router(state.clone()).oneshot( + request + .method(Method::POST) + .uri(format!("/1,{id:x}{TEST_COOKIE:08x}?ts=1700000000")) + .body(body) + .unwrap(), + ), + ) + .await + .unwrap() + .unwrap(); + assert_eq!(response.status(), StatusCode::CREATED); + let read = |v: &Arc| { + let mut stored = Needle { + id: NeedleId(id), + cookie: Cookie(TEST_COOKIE), + ..Needle::default() + }; + v.read_needle(&mut stored).unwrap(); + stored + }; + let (primary, replica) = (read(&volume), read(&peer_volume)); + assert!(primary.is_compressed()); + assert_eq!( + maybe_decompress_gzip(&primary.data).unwrap(), + payload.as_bytes() + ); + assert_eq!(replica.data, primary.data); + assert_eq!(replica.flags, primary.flags); + assert_eq!(replica.name, primary.name); + assert_eq!(replica.mime, primary.mime); + assert_eq!(replica.pairs, primary.pairs); + assert_eq!(replica.ttl, primary.ttl); + assert_eq!(replica.last_modified, 1700000000); + } + master_server.abort(); + peer_server.abort(); + } + + #[tokio::test] + async fn http_replica_keeps_uncompressed_bytes() { + use tower::ServiceExt; + let tmp = tempfile::TempDir::new().unwrap(); + let state = volume_test_state(&tmp); + let volume = state + .store + .read() + .unwrap() + .find_volume(VolumeId(1)) + .unwrap() + .1; + let payload = "compressible replica text ".repeat(64); + for (id, query, compressed) in [ + (0x6e7a_0c08u64, "?type=replicate", false), + (0x6e7a_0c09u64, "", true), + ] { + let response = super::super::volume_server::build_admin_router(state.clone()) + .oneshot( + Request::builder() + .method(Method::POST) + .uri(format!("/1,{id:x}{TEST_COOKIE:08x}{query}")) + .header(header::CONTENT_TYPE, "text/plain") + .body(Body::from(payload.clone())) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::CREATED); + let mut stored = Needle { + id: NeedleId(id), + cookie: Cookie(TEST_COOKIE), + ..Needle::default() + }; + volume.read_needle(&mut stored).unwrap(); + assert_eq!(stored.is_compressed(), compressed, "query {query:?}"); + } + } + + #[tokio::test] + async fn http_orientation_fix_follows_the_filename() { + use tower::ServiceExt; + let tmp = tempfile::TempDir::new().unwrap(); + let mut state = volume_test_state(&tmp); + Arc::get_mut(&mut state).unwrap().fix_jpg_orientation = true; + let volume = state + .store + .read() + .unwrap() + .find_volume(VolumeId(1)) + .unwrap() + .1; + let mut encoded = std::io::Cursor::new(Vec::new()); + image::DynamicImage::ImageRgba8(image::RgbaImage::new(3, 2)) + .write_to(&mut encoded, image::ImageFormat::Jpeg) + .unwrap(); + let exif: &[u8] = + b"Exif\0\0MM\0\x2a\0\0\0\x08\0\x01\x01\x12\0\x03\0\0\0\x01\0\x03\0\0\0\0\0\0"; + let mut tagged = vec![0xFF, 0xD8, 0xFF, 0xE1, 0, exif.len() as u8 + 2]; + tagged.extend_from_slice(exif); + tagged.extend_from_slice(&encoded.into_inner()[2..]); + let upload = |id: u64, suffix: &'static str, filename: &'static str, body: Vec| { + let (state, volume) = (state.clone(), volume.clone()); + async move { + let response = super::super::volume_server::build_admin_router(state) + .oneshot( + Request::builder() + .method(Method::POST) + .uri(format!("/1,{id:x}{TEST_COOKIE:08x}{suffix}")) + .header(header::CONTENT_TYPE, "application/octet-stream") + .header( + header::CONTENT_DISPOSITION, + format!("attachment; filename=\"{filename}\""), + ) + .body(Body::from(body)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!( + response.status(), + StatusCode::CREATED, + "{suffix} {filename}" + ); + let mut stored = Needle { + id: NeedleId(id), + cookie: Cookie(TEST_COOKIE), + ..Needle::default() + }; + volume.read_needle(&mut stored).unwrap(); + stored.data + } + }; + let rotated = upload(0x6e7a_0c10, "", "photo.jpg", tagged.clone()).await; + assert_ne!(rotated, tagged); + assert_eq!( + upload(0x6e7a_0c11, ".jpg", "photo.png", tagged.clone()).await, + tagged + ); + assert_eq!( + upload(0x6e7a_0c12, "?type=replicate", "photo.jpg", tagged.clone()).await, + rotated + ); + assert_eq!( + upload(0x6e7a_0c13, "?type=replicate", "photo.jpg", rotated.clone()).await, + rotated + ); + } + #[tokio::test] async fn http_multipart_replica_refusal_is_not_marked() { use tower::ServiceExt; diff --git a/weed/operation/needle_parse_test.go b/weed/operation/needle_parse_test.go index fb1d530a78..6251ca771d 100644 --- a/weed/operation/needle_parse_test.go +++ b/weed/operation/needle_parse_test.go @@ -146,6 +146,27 @@ func TestUploadKeepsMarginalGzipUncompressed(t *testing.T) { assert.Equal(t, data, stored.Data) } +func TestReplicationKeepsTheSourceMime(t *testing.T) { + for _, mimeType := range []string{"", "application/octet-stream", "Text/Plain; charset=UTF-8"} { + mockClient := &MockClient{} + var stored *needle.Needle + mockClient.needleHandling = func(n *needle.Needle, originalSize int, err error) { + assert.Equal(t, nil, err, "upload: %v", err) + stored = n + } + newUploader(mockClient).UploadData(context.Background(), []byte(textContent), &UploadOption{ + UploadUrl: "http://localhost:8080/389,0f084d17353afda0?type=replicate", + Filename: "report.html", + MimeType: mimeType, + IsReplication: true, + }) + if stored == nil { + t.Fatal("no needle reached the volume server") + } + assert.Equal(t, mimeType, string(stored.Mime)) + } +} + var textContent = `Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: diff --git a/weed/operation/upload_content.go b/weed/operation/upload_content.go index 4d829e33e7..71992dfe78 100644 --- a/weed/operation/upload_content.go +++ b/weed/operation/upload_content.go @@ -474,6 +474,7 @@ func (uploader *Uploader) doUploadData(ctx context.Context, data []byte, option Filename: option.Filename, Cipher: false, IsInputCompressed: contentIsGzipped, + IsReplication: option.IsReplication, MimeType: option.MimeType, PairMap: option.PairMap, Jwt: option.Jwt, @@ -510,7 +511,7 @@ func (uploader *Uploader) upload_content(ctx context.Context, fillBufferFunction // properly handling non-ASCII characters and special characters h.Set("Content-Disposition", mime.FormatMediaType("form-data", map[string]string{"name": "file", "filename": option.Filename})) h.Set("Idempotency-Key", option.UploadUrl) - if option.MimeType == "" { + if option.MimeType == "" && !option.IsReplication { option.MimeType = mime.TypeByExtension(strings.ToLower(filepath.Ext(option.Filename))) } if option.MimeType != "" { diff --git a/weed/storage/needle/needle_parse_upload.go b/weed/storage/needle/needle_parse_upload.go index 3a668de9b6..f0b07e21db 100644 --- a/weed/storage/needle/needle_parse_upload.go +++ b/weed/storage/needle/needle_parse_upload.go @@ -298,7 +298,9 @@ func parseUpload(r *http.Request, sizeLimit int64, pu *ParsedUpload) (e error) { pu.IsChunkedFile, _ = strconv.ParseBool(r.FormValue("cm")) - if !pu.IsChunkedFile { + if r.FormValue("type") == "replicate" { + pu.MimeType = contentType + } else if !pu.IsChunkedFile { dotIndex := strings.LastIndex(pu.FileName, ".") ext, mtype := "", ""