volume: replicas keep the primary's compression and metadata (#12350)

* volume(rust): replicate the stored needle instead of the request body

The primary forwarded the original request body and headers, so a
replica re-derived compression from the raw bytes. Send what Go's
ReplicatedWrite sends: a one-part multipart upload of the stored bytes,
gzip-marked when compressed, with the name, mime, pairs and Content-MD5,
and carry ts, ttl, cm and fsync in the query so form-field values reach
the replica too.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* volume(rust): keep the primary's compression on replica writes

A type=replicate upload now stores the bytes as received, like Go's
ParseUpload, so a replica no longer compresses data the primary stored
uncompressed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* volume(rust): replicas keep the primary's name and mime

Forward the primary's filename and mime even past the 255-byte needle
fields, so a replica applies the same limit and sets the same flags,
and skip the filename MIME guess on type=replicate, so an untyped
upload stays untyped on every copy.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* volume(rust): keep the multipart part Content-Type as sent

Parsing the part type as a mime value lowercased it and dropped values
it could not parse, so a replica stored a different mime than the
primary. Read the header verbatim, as Go's part.Header.Get does.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* volume: replicas keep the primary's mime as forwarded

A Go replica guessed the mime from the file extension, so an upload the
primary stored as application/octet-stream, or with no mime at all, got
a different type on the replica. On type=replicate keep the part's
Content-Type as sent, and have the replication sender leave an empty
mime empty instead of filling it in from the extension. This matches
what the Rust replica does.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* volume(rust): pick the JPEG orientation fix by file name, as Go does

Go checks the mime and the upload's file name, Rust checked the mime and
the URL path, so the same upload could be rotated on one server and not
on the other. With the name and mime forwarded as stored, a replica now
reaches the primary's decision, and a rotated image has no EXIF tag left
to act on, so the fix stays a no-op there.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

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")
 }
 

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<String, String> {
+    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<Vec<u8>> {
     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<crate::storage::volume::Volume>| {
+                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<u8>| {
+            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;

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:

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 != "" {

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 := "", ""