fix(volume): return conflict and explicit proof for read-only upload refusals (#12236)

* fix(volume): report read-only HTTP refusals in Rust

* fix(volume): mirror HTTP write refusal proof in Go

* test(volume): cover HTTP read-only refusal parity

* test(volume): cover local refusal after remote append

* fix(volume): honor decoded replica controls before retry proof

* fix(operation): reassign proven HTTP conflict refusals

* fix(operation): stop same-target retries on HTTP conflict

* fix(volume): replace replica marker when forwarding requests

* fix(java): gate conflict reassignment on refusal proof

* fix(volume): preserve authentication query encoding in replication

* fix(operation): stop unproven conflicts in forever retries

other/java/client/src/main/java/seaweedfs/client/SeaweedWrite.java

@@ -2,6 +2,7 @@ package seaweedfs.client;
 
 import com.google.common.base.Strings;
 import com.google.protobuf.ByteString;
+import org.apache.http.Header;
 import org.apache.http.HttpEntity;
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpPost;
@@ -35,6 +36,7 @@ public class SeaweedWrite {
             final String path) throws IOException {
 
         IOException lastException = null;
+        boolean reassigned = false;
         for (long waitTime = 1000L; waitTime < 10 * 1000; waitTime += waitTime / 2) {
             try {
                 FilerProto.FileChunk.Builder chunkBuilder = writeChunk(
@@ -47,6 +49,16 @@ public class SeaweedWrite {
             } catch (IOException ioe) {
                 LOG.debug("writeData:{}", ioe);
                 lastException = ioe;
+                if (ioe instanceof UploadException) {
+                    UploadException error = (UploadException) ioe;
+                    if (error.status == 409) {
+                        if (!error.refusedBeforeAppend || reassigned) {
+                            throw error;
+                        }
+                        reassigned = true;
+                        continue;
+                    }
+                }
             }
             try {
                 Thread.sleep(waitTime);
@@ -195,17 +207,24 @@ public class SeaweedWrite {
                 .addBinaryBody("upload", inputStream)
                 .build());
 
-        CloseableHttpResponse response = SeaweedUtil.getClosableHttpClient().execute(post);
-
-        try {
-            if (response.getStatusLine().getStatusCode() / 100 != 2) {
+        try (CloseableHttpResponse response = SeaweedUtil.getClosableHttpClient().execute(post)) {
+            int status = response.getStatusLine().getStatusCode();
+            if (status / 100 != 2) {
+                Header outcome = response.getFirstHeader("X-Seaweed-Write-Outcome");
+                boolean refusedBeforeAppend = status == 409 && outcome != null
+                        && "refused-before-append".equals(outcome.getValue());
+                String message = response.getStatusLine().getReasonPhrase();
+                IOException cause = null;
                 HttpEntity entity = response.getEntity();
-                if (entity != null && entity.getContentType() != null
-                        && entity.getContentType().getValue().equals("application/json")) {
-                    throw new IOException(EntityUtils.toString(entity, "UTF-8"));
-                } else {
-                    throw new IOException(response.getStatusLine().getReasonPhrase());
+                try {
+                    if (entity != null && entity.getContentType() != null
+                            && entity.getContentType().getValue().equals("application/json")) {
+                        message = EntityUtils.toString(entity, "UTF-8");
+                    }
+                } catch (IOException e) {
+                    cause = e;
                 }
+                throw new UploadException(status, refusedBeforeAppend, message, cause);
             }
 
             String etag = response.getLastHeader("ETag").getValue();
@@ -218,12 +237,22 @@ public class SeaweedWrite {
 
             return etag;
         } finally {
-            response.close();
             post.releaseConnection();
         }
 
     }
 
+    private static final class UploadException extends IOException {
+        private final int status;
+        private final boolean refusedBeforeAppend;
+
+        private UploadException(int status, boolean refusedBeforeAppend, String message, IOException cause) {
+            super(message, cause);
+            this.status = status;
+            this.refusedBeforeAppend = refusedBeforeAppend;
+        }
+    }
+
     private static byte[] genCipherKey() {
         byte[] b = new byte[32];
         random.nextBytes(b);

other/java/client/src/test/java/seaweedfs/client/SeaweedWriteIntegrationTest.java

@@ -0,0 +1,207 @@
+package seaweedfs.client;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import io.grpc.Server;
+import io.grpc.ServerBuilder;
+import io.grpc.stub.StreamObserver;
+import org.junit.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.*;
+
+public class SeaweedWriteIntegrationTest {
+    private static final byte[] BYTES = "prefix-upload\u0000payload-suffix".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] PAYLOAD = Arrays.copyOfRange(BYTES, 7, BYTES.length - 7);
+    private static final String PROOF = "refused-before-append";
+
+    @Test(timeout = 30000)
+    public void testProvenRefusalReassignsOnce() throws Exception {
+        for (boolean cipher : new boolean[]{false, true}) {
+            try (UploadServer server = new UploadServer(cipher, 1, 409, PROOF, false)) {
+                FilerProto.Entry.Builder entry = server.write();
+                server.assertAttempts(2);
+                assertEquals(1, entry.getChunksCount());
+                FilerProto.FileChunk chunk = entry.getChunks(0);
+                assertEquals("7,200000000", chunk.getFileId());
+                assertEquals(23, chunk.getOffset());
+                assertEquals(PAYLOAD.length, chunk.getSize());
+                assertEquals("uploaded", chunk.getETag());
+                byte[] uploaded = server.payloads.get(1);
+                if (cipher) {
+                    uploaded = SeaweedCipher.decrypt(uploaded, chunk.getCipherKey().toByteArray());
+                } else {
+                    assertArrayEquals(PAYLOAD, server.payloads.get(0));
+                }
+                assertArrayEquals(PAYLOAD, uploaded);
+            }
+        }
+    }
+
+    @Test(timeout = 30000)
+    public void testUnprovenConflictDoesNotRetry() throws Exception {
+        for (String outcome : new String[]{null, "unknown", "REFUSED-BEFORE-APPEND"}) {
+            try (UploadServer server = new UploadServer(false, 1, 409, outcome, false)) {
+                server.assertWriteFails(1);
+            }
+        }
+    }
+
+    @Test(timeout = 30000)
+    public void testRepeatedProvenRefusalDoesNotRetryAgain() throws Exception {
+        try (UploadServer server = new UploadServer(false, 2, 409, PROOF, false)) {
+            server.assertWriteFails(2);
+        }
+    }
+
+    @Test(timeout = 30000)
+    public void testTruncatedConflictRetainsRetryProof() throws Exception {
+        try (UploadServer server = new UploadServer(false, 1, 409, null, true)) {
+            assertNotNull(server.assertWriteFails(1).getCause());
+        }
+        try (UploadServer server = new UploadServer(false, 1, 409, PROOF, true)) {
+            assertEquals(1, server.write().getChunksCount());
+            server.assertAttempts(2);
+        }
+    }
+
+    @Test(timeout = 30000)
+    public void testOtherUploadErrorsKeepExistingRetry() throws Exception {
+        try (UploadServer server = new UploadServer(false, 1, 500, PROOF, false)) {
+            assertEquals(1, server.write().getChunksCount());
+            server.assertAttempts(2);
+        }
+    }
+
+    private static final class UploadServer implements AutoCloseable {
+        private final AtomicInteger assignments = new AtomicInteger();
+        private final AtomicInteger uploads = new AtomicInteger();
+        private final List<byte[]> payloads = Collections.synchronizedList(new ArrayList<>());
+        private final List<String> paths = Collections.synchronizedList(new ArrayList<>());
+        private final List<String> auth = Collections.synchronizedList(new ArrayList<>());
+        private final List<String> digests = Collections.synchronizedList(new ArrayList<>());
+        private final HttpServer volume;
+        private final Server filer;
+        private final FilerClient client;
+
+        private UploadServer(boolean cipher, int refusals, int status, String outcome, boolean truncated)
+                throws IOException {
+            volume = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+            volume.createContext("/", exchange -> upload(exchange, refusals, status, outcome, truncated));
+            volume.start();
+            filer = ServerBuilder.forPort(0).addService(new SeaweedFilerGrpc.SeaweedFilerImplBase() {
+                @Override
+                public void getFilerConfiguration(FilerProto.GetFilerConfigurationRequest request,
+                        StreamObserver<FilerProto.GetFilerConfigurationResponse> observer) {
+                    observer.onNext(FilerProto.GetFilerConfigurationResponse.newBuilder().setCipher(cipher).build());
+                    observer.onCompleted();
+                }
+
+                @Override
+                public void assignVolume(FilerProto.AssignVolumeRequest request,
+                        StreamObserver<FilerProto.AssignVolumeResponse> observer) {
+                    int attempt = assignments.incrementAndGet();
+                    observer.onNext(FilerProto.AssignVolumeResponse.newBuilder()
+                            .setFileId("7," + attempt + "00000000")
+                            .setAuth("token-" + attempt)
+                            .setLocation(FilerProto.Location.newBuilder()
+                                    .setUrl("127.0.0.1:" + volume.getAddress().getPort()))
+                            .build());
+                    observer.onCompleted();
+                }
+            }).build().start();
+            client = new FilerClient("127.0.0.1", filer.getPort());
+        }
+
+        private void upload(HttpExchange exchange, int refusals, int status, String outcome, boolean truncated)
+                throws IOException {
+            try {
+                ByteArrayOutputStream body = new ByteArrayOutputStream();
+                byte[] buffer = new byte[1024];
+                int count;
+                while ((count = exchange.getRequestBody().read(buffer)) != -1) {
+                    body.write(buffer, 0, count);
+                }
+                String multipart = new String(body.toByteArray(), StandardCharsets.ISO_8859_1);
+                String boundary = exchange.getRequestHeaders().getFirst("Content-Type").split("boundary=")[1];
+                boundary = boundary.replace("\"", "");
+                int start = multipart.indexOf("\r\n\r\n") + 4;
+                int end = multipart.lastIndexOf("\r\n--" + boundary);
+                payloads.add(Arrays.copyOfRange(body.toByteArray(), start, end));
+                paths.add(exchange.getRequestURI().getPath());
+                auth.add(exchange.getRequestHeaders().getFirst("Authorization"));
+                digests.add(exchange.getRequestHeaders().getFirst("Content-MD5"));
+                int attempt = uploads.incrementAndGet();
+                byte[] response = "{\"error\":\"read only; refused-before-append\"}"
+                        .getBytes(StandardCharsets.UTF_8);
+                if (attempt <= refusals) {
+                    exchange.getResponseHeaders().add("Content-Type", "application/json");
+                    if (outcome != null) {
+                        exchange.getResponseHeaders().add("X-Seaweed-Write-Outcome", outcome);
+                    }
+                    exchange.sendResponseHeaders(status, response.length + (truncated ? 20 : 0));
+                } else {
+                    exchange.getResponseHeaders().add("ETag", "\"uploaded\"");
+                    exchange.sendResponseHeaders(201, response.length);
+                }
+                exchange.getResponseBody().write(response);
+            } finally {
+                exchange.close();
+            }
+        }
+
+        private FilerProto.Entry.Builder write() throws IOException {
+            FilerProto.Entry.Builder entry = FilerProto.Entry.newBuilder();
+            SeaweedWrite.writeData(entry, "", "", client, 23, BYTES, 7, PAYLOAD.length, "/upload");
+            return entry;
+        }
+
+        private IOException assertWriteFails(int attempts) throws Exception {
+            FilerProto.Entry.Builder entry = FilerProto.Entry.newBuilder();
+            IOException error = null;
+            try {
+                SeaweedWrite.writeData(entry, "", "", client, 23, BYTES, 7, PAYLOAD.length, "/upload");
+                fail("expected upload conflict");
+            } catch (IOException expected) {
+                assertNotNull(expected.getMessage());
+                error = expected;
+            }
+            assertEquals(0, entry.getChunksCount());
+            assertAttempts(attempts);
+            return error;
+        }
+
+        private void assertAttempts(int attempts) throws Exception {
+            assertEquals(attempts, assignments.get());
+            assertEquals(attempts, uploads.get());
+            for (int i = 0; i < attempts; i++) {
+                assertEquals("/7," + (i + 1) + "00000000", paths.get(i));
+                assertEquals("BEARER token-" + (i + 1), auth.get(i));
+                assertEquals(Base64.getEncoder().encodeToString(MessageDigest.getInstance("MD5")
+                        .digest(payloads.get(i))), digests.get(i));
+            }
+        }
+
+        @Override
+        public void close() throws InterruptedException {
+            try {
+                client.shutdown();
+            } finally {
+                filer.shutdownNow().awaitTermination(5, TimeUnit.SECONDS);
+                volume.stop(0);
+            }
+        }
+    }
+}

seaweed-volume/src/server/handlers.rs

@@ -751,11 +751,16 @@ async fn do_replicated_request(
         return Ok(());
     }
 
-    let new_query = if query.is_empty() {
-        String::from("type=replicate")
-    } else {
-        format!("{}&type=replicate", query)
-    };
+    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 mut futures = Vec::new();
     for loc in remote_locations {
@@ -2684,7 +2689,10 @@ pub async fn post_handler(
     }
 
     // Upload throttling: check inflight bytes against limit
-    let is_replicate = query.split('&').any(|p| p == "type=replicate");
+    let is_replicate = query_fields
+        .iter()
+        .find(|(name, _)| name == "type")
+        .is_some_and(|(_, value)| value == "replicate");
     let content_length = headers
         .get(header::CONTENT_LENGTH)
         .and_then(|v| v.to_str().ok())
@@ -2896,6 +2904,8 @@ pub async fn post_handler(
             .or_else(|| multipart_form_fields.get(name).cloned())
     };
 
+    let is_replicate = form_value("type").as_deref() == Some("replicate");
+
     // Check for chunk manifest flag.
     // Go uses r.FormValue("cm"), which falls back to multipart fields when present.
     let is_chunk_manifest = matches!(
@@ -3193,7 +3203,12 @@ pub async fn post_handler(
         None
     };
 
-    let write_result = write_upload_needle(&state, vid, n, fsync).await;
+    let (write_result, refused_before_append) =
+        match write_upload_needle(&state, vid, n, fsync).await {
+            Ok(Ok(written)) => (Ok(written), false),
+            Ok(Err(refused)) => (Err(refused), true),
+            Err(error) => (Err(error), false),
+        };
 
     // No leader-known gate: when the lookup cannot run, the write must fail
     // rather than be acknowledged on fewer copies than required. The local
@@ -3255,11 +3270,21 @@ pub async fn post_handler(
             metrics::HANDLER_COUNTER
                 .with_label_values(&[metrics::ERROR_WRITE_TO_LOCAL_DISK])
                 .inc();
-            json_error_with_query(
-                StatusCode::INTERNAL_SERVER_ERROR,
-                format!("{}", e),
-                Some(&query),
-            )
+            let read_only = matches!(e, crate::storage::volume::VolumeError::ReadOnly(_));
+            let status = if read_only {
+                StatusCode::CONFLICT
+            } else {
+                StatusCode::INTERNAL_SERVER_ERROR
+            };
+            let mut response = json_error_with_query(status, format!("{}", e), Some(&query));
+            // Replication or a failed append can leave another copy written.
+            if read_only && refused_before_append && !fsync && !needs_replication && !is_replicate {
+                response.headers_mut().insert(
+                    "x-seaweed-write-outcome",
+                    header::HeaderValue::from_static("refused-before-append"),
+                );
+            }
+            response
         }
     };
 
@@ -3267,26 +3292,34 @@ pub async fn post_handler(
     resp
 }
 
-/// Append an uploaded needle the way Go's `writeNeedle2` does: a durable
-/// write goes to the volume's batching worker, which group-commits the
-/// batch's .dat/.idx syncs, while anything else — a non-durable write, or a
-/// durable one landing while the server is stopping and the workers are
-/// winding down — takes the synchronous path and fsyncs inline.
+/// Durable uploads use the batching worker unless the server is stopping.
+/// Only an inner error proves refusal before append.
 async fn write_upload_needle(
     state: &Arc<VolumeServerState>,
     vid: VolumeId,
     mut n: Needle,
     fsync: bool,
-) -> Result<(u64, Size, bool), crate::storage::volume::VolumeError> {
+) -> Result<
+    Result<(u64, Size, bool), crate::storage::volume::VolumeError>,
+    crate::storage::volume::VolumeError,
+> {
     let queued = fsync && !*state.is_stopping.read().unwrap();
     if queued && let Some(wq) = state.write_queue.get() {
-        return wq.submit(vid, n, fsync).await;
+        return wq.submit(vid, n, fsync).await.map(Ok);
     }
     let vol = {
         let store = state.store.read().unwrap();
         store.writable_volume(vid)
     };
-    vol.and_then(|vol| vol.write_needle(&mut n, true, fsync))
+    let vol = match vol {
+        Ok(vol) => vol,
+        Err(refused) => return Ok(Err(refused)),
+    };
+    if fsync {
+        vol.write_needle(&mut n, true, fsync).map(Ok)
+    } else {
+        vol.write_needle_unsynced_or_refused(&mut n)
+    }
 }
 
 // ============================================================================
@@ -7902,6 +7935,355 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn http_sealed_volume_reports_refused_before_append_without_changing_data() {
+        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;
+        volume.set_read_only().unwrap();
+        let before = volume.dat_file_size().unwrap();
+        let response = super::super::volume_server::build_admin_router(state)
+            .oneshot(
+                Request::builder()
+                    .method(Method::POST)
+                    .uri(format!("/1,{:x}{TEST_COOKIE:08x}", 0x6e7a_0c01u64))
+                    .body(Body::from("kv-http-bytes"))
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(response.status(), StatusCode::CONFLICT);
+        assert_eq!(
+            response
+                .headers()
+                .get("x-seaweed-write-outcome")
+                .map(|value| value.to_str().unwrap()),
+            Some("refused-before-append")
+        );
+        assert_eq!(volume.dat_file_size().unwrap(), before);
+    }
+
+    #[tokio::test]
+    async fn http_local_refusal_after_replica_append_is_not_marked() {
+        for local_read_only in [true, false] {
+            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();
+            if local_read_only {
+                volume.set_read_only().unwrap();
+            }
+            let before = volume.dat_file_size().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;
+            if !local_read_only {
+                peer_volume.set_read_only().unwrap();
+            }
+            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();
+            });
+
+            // The master's writable view can lag behind the local seal.
+            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 id = 0x6e7a_0c03u64;
+            let payload = b"replicated bytes";
+            let response = tokio::time::timeout(
+                std::time::Duration::from_secs(5),
+                super::super::volume_server::build_admin_router(state).oneshot(
+                    Request::builder()
+                        .method(Method::POST)
+                        .uri(format!("/1,{id:x}{TEST_COOKIE:08x}"))
+                        .body(Body::from(payload.as_slice()))
+                        .unwrap(),
+                ),
+            )
+            .await
+            .unwrap()
+            .unwrap();
+            assert_eq!(
+                response.status(),
+                if local_read_only {
+                    StatusCode::CONFLICT
+                } else {
+                    StatusCode::INTERNAL_SERVER_ERROR
+                }
+            );
+            assert!(response.headers().get("x-seaweed-write-outcome").is_none());
+            let (refused, written) = if local_read_only {
+                (&volume, &peer_volume)
+            } else {
+                (&peer_volume, &volume)
+            };
+            assert_eq!(refused.dat_file_size().unwrap(), before);
+            let mut stored = Needle {
+                id: NeedleId(id),
+                cookie: Cookie(TEST_COOKIE),
+                ..Needle::default()
+            };
+            written.read_needle(&mut stored).unwrap();
+            assert_eq!(stored.data, payload);
+            master_server.abort();
+            peer_server.abort();
+        }
+    }
+
+    #[tokio::test]
+    async fn http_replication_preserves_query_authentication() {
+        use crate::security::{SigningKey, gen_jwt};
+        use crate::storage::super_block::ReplicaPlacement;
+        use axum::{Json, Router, middleware, 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 key = SigningKey(b"replication-secret".to_vec());
+        state.guard.write().unwrap().signing_key = key.clone();
+        peer.guard.write().unwrap().signing_key = key.clone();
+
+        let queries = Arc::new(std::sync::Mutex::new(Vec::new()));
+        let received = queries.clone();
+        let app = super::super::volume_server::build_admin_router(peer).layer(middleware::from_fn(
+            move |request: Request<Body>, next: middleware::Next| {
+                received
+                    .lock()
+                    .unwrap()
+                    .push(request.uri().query().unwrap().to_string());
+                async move { next.run(request).await }
+            },
+        ));
+        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, app).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 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 payload = b"authenticated replica bytes";
+        for method in [Method::POST, Method::DELETE] {
+            let response = tokio::time::timeout(
+                std::time::Duration::from_secs(5),
+                super::super::volume_server::build_admin_router(state.clone()).oneshot(
+                    Request::builder()
+                        .method(method.clone())
+                        .uri(format!("/{fid}?{query}"))
+                        .header(header::AUTHORIZATION, format!("Bearer {token}"))
+                        .body(Body::from(if method == Method::POST {
+                            payload.as_slice()
+                        } else {
+                            &[]
+                        }))
+                        .unwrap(),
+                ),
+            )
+            .await
+            .unwrap()
+            .unwrap();
+            assert_eq!(
+                response.status(),
+                if method == Method::POST {
+                    StatusCode::CREATED
+                } else {
+                    StatusCode::ACCEPTED
+                }
+            );
+            for stored_volume in [&volume, &peer_volume] {
+                let mut stored = Needle {
+                    id: NeedleId(id),
+                    cookie: Cookie(TEST_COOKIE),
+                    ..Needle::default()
+                };
+                let result = stored_volume.read_needle(&mut stored);
+                if method == Method::POST {
+                    result.unwrap();
+                    assert_eq!(stored.data, payload);
+                } else {
+                    assert!(result.is_err());
+                }
+            }
+        }
+        assert_eq!(*queries.lock().unwrap(), [expected_query, expected_query]);
+        master_server.abort();
+        peer_server.abort();
+    }
+
+    #[tokio::test]
+    async fn http_multipart_replica_refusal_is_not_marked() {
+        use tower::ServiceExt;
+        for field_first in [true, false] {
+            let tmp = tempfile::TempDir::new().unwrap();
+            let state = volume_test_state(&tmp);
+            let volume = state
+                .store
+                .read()
+                .unwrap()
+                .find_volume(VolumeId(1))
+                .unwrap()
+                .1;
+            volume.set_read_only().unwrap();
+            let before = volume.dat_file_size().unwrap();
+            let field =
+                "--boundary\r\nContent-Disposition: form-data; name=\"type\"\r\n\r\nreplicate\r\n";
+            let file = "--boundary\r\nContent-Disposition: form-data; name=\"file\"; filename=\"data.bin\"\r\nContent-Type: application/octet-stream\r\n\r\nreplica bytes\r\n";
+            let body = if field_first {
+                format!("{field}{file}--boundary--\r\n")
+            } else {
+                format!("{file}{field}--boundary--\r\n")
+            };
+            let response = super::super::volume_server::build_admin_router(state)
+                .oneshot(
+                    Request::builder()
+                        .method(Method::POST)
+                        .uri(format!("/1,{:x}{TEST_COOKIE:08x}", 0x6e7a_0c04u64))
+                        .header(
+                            header::CONTENT_TYPE,
+                            "multipart/form-data; boundary=boundary",
+                        )
+                        .body(Body::from(body))
+                        .unwrap(),
+                )
+                .await
+                .unwrap();
+            assert_eq!(response.status(), StatusCode::CONFLICT);
+            assert!(response.headers().get("x-seaweed-write-outcome").is_none());
+            assert_eq!(volume.dat_file_size().unwrap(), before);
+        }
+    }
+
+    #[tokio::test]
+    async fn http_write_errors_without_no_append_proof_are_not_marked() {
+        use tower::ServiceExt;
+        for query in [
+            "",
+            "?fsync=true",
+            "?type=replicate",
+            "?type=%72eplicate",
+            "?%74ype=replicate",
+        ] {
+            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 id = 0x6e7a_0c02u64;
+            if query.is_empty() {
+                // This error is injected after the append. Its error variant
+                // cannot prove a safe retry, even if rollback later succeeds.
+                volume.fail_append_for_test(&[NeedleId(id)]);
+            } else {
+                volume.set_read_only().unwrap();
+            }
+            let response = super::super::volume_server::build_admin_router(state)
+                .oneshot(
+                    Request::builder()
+                        .method(Method::POST)
+                        .uri(format!("/1,{id:x}{TEST_COOKIE:08x}{query}"))
+                        .body(Body::from("kv-http-bytes"))
+                        .unwrap(),
+                )
+                .await
+                .unwrap();
+            assert_eq!(
+                response.status(),
+                if query.is_empty() {
+                    StatusCode::INTERNAL_SERVER_ERROR
+                } else {
+                    StatusCode::CONFLICT
+                },
+                "{query}"
+            );
+            assert!(
+                response.headers().get("x-seaweed-write-outcome").is_none(),
+                "{query}"
+            );
+        }
+    }
+
     /// Issue #801: an fsync upload takes the per-volume worker so its sync is
     /// shared across the batch (Go's `writeNeedle2` split); a non-fsync write,
     /// or an fsync write landing while the server is stopping, takes the
@@ -7936,17 +8318,20 @@ mod tests {
 
         write_upload_needle(&state, VolumeId(1), needle(0x6e7a_0b01), true)
             .await
+            .unwrap()
             .unwrap();
         assert_eq!(queue.submits_for_test(), 1);
 
         write_upload_needle(&state, VolumeId(1), needle(0x6e7a_0b02), false)
             .await
+            .unwrap()
             .unwrap();
         assert_eq!(queue.submits_for_test(), 1);
 
         *state.is_stopping.write().unwrap() = true;
         write_upload_needle(&state, VolumeId(1), needle(0x6e7a_0b03), true)
             .await
+            .unwrap()
             .unwrap();
         assert_eq!(queue.submits_for_test(), 1);
     }

seaweed-volume/src/storage/volume.rs

@@ -2631,6 +2631,20 @@ impl Volume {
         results.pop().unwrap_or(Err(VolumeError::NotFound))
     }
 
+    /// The inner error proves that no append was attempted.
+    pub fn write_needle_unsynced_or_refused(
+        &self,
+        n: &mut Needle,
+    ) -> Result<Result<(u64, Size, bool), VolumeError>, VolumeError> {
+        let _sync_guard = self.data_file_sync_lock.lock().unwrap();
+        let _guard = self.data_file_access_control.write_lock();
+        let mut io = self.io.write().unwrap();
+        if let Err(refused) = self.check_writable() {
+            return Ok(Err(refused));
+        }
+        self.do_write_request(&mut io, n, true, false).map(Ok)
+    }
+
     /// Write a batch of needles the way Go's processBatch does: one .dat
     /// sync and one .idx sync per run of distinct needle ids that holds a
     /// durable write, instead of two per durable needle. Nothing in such a
@@ -6440,6 +6454,49 @@ mod tests {
         (format!("http://{}", addr), shutdown_tx)
     }
 
+    #[test]
+    fn a_sealed_volume_refuses_before_append_and_a_write_error_does_not() {
+        let tmp = TempDir::new().unwrap();
+        let v = make_test_volume(tmp.path().to_str().unwrap());
+        let needle = |id: u64, cookie: u32| Needle {
+            id: NeedleId(id),
+            cookie: Cookie(cookie),
+            data: b"uploaded bytes".to_vec(),
+            data_size: 14,
+            ..Needle::default()
+        };
+
+        let mut n = needle(1, 7);
+        assert!(matches!(
+            v.write_needle_unsynced_or_refused(&mut n),
+            Ok(Ok(_))
+        ));
+
+        let mut wrong = needle(1, 8);
+        assert!(v.write_needle_unsynced_or_refused(&mut wrong).is_err());
+
+        v.fail_append_for_test(&[NeedleId(3)]);
+        let mut torn = needle(3, 10);
+        assert!(matches!(
+            v.write_needle_unsynced_or_refused(&mut torn),
+            Err(VolumeError::Io(_))
+        ));
+        v.fail_append_for_test(&[]);
+
+        v.set_read_only().unwrap();
+        let before = v.dat_file_size().unwrap();
+        let mut late = needle(2, 9);
+        assert!(matches!(
+            v.write_needle_unsynced_or_refused(&mut late),
+            Ok(Err(VolumeError::ReadOnly(_)))
+        ));
+        assert_eq!(
+            v.dat_file_size().unwrap(),
+            before,
+            "a refused write appended bytes"
+        );
+    }
+
     fn make_test_volume(dir: &str) -> Volume {
         Volume::new(
             dir,

test/volume_server/http/replication_lifecycle_test.go

@@ -43,7 +43,6 @@ func TestReplicatedUploadSucceedsImmediatelyAfterAllocate(t *testing.T) {
 	}
 
 	client := framework.NewHTTPClient()
-	fid := framework.NewFileID(volumeID, 881001, 0x0B0C0D0E)
 	payload := []byte("replicated-upload-after-allocate")
 
 	// The master only learns about replica locations through volume-server
@@ -57,21 +56,24 @@ func TestReplicatedUploadSucceedsImmediatelyAfterAllocate(t *testing.T) {
 		t.Fatalf("master did not report 2 replica locations for volume %d within deadline", volumeID)
 	}
 
-	uploadResp := framework.UploadBytes(t, client, clusterHarness.VolumeAdminURL(0), fid, payload)
-	_ = framework.ReadAllAndClose(t, uploadResp)
-	if uploadResp.StatusCode != http.StatusCreated {
-		t.Fatalf("replicated upload expected 201, got %d", uploadResp.StatusCode)
-	}
+	for i, query := range []string{"", "?type=client", "?type=", "?%74ype=client", "?type=client&type=old", "?type=client&fsync=%74rue"} {
+		fid := framework.NewFileID(volumeID, 881001+uint64(i), 0x0B0C0D0E)
+		uploadResp := framework.UploadBytes(t, client, clusterHarness.VolumeAdminURL(0), fid+query, payload)
+		_ = framework.ReadAllAndClose(t, uploadResp)
+		if uploadResp.StatusCode != http.StatusCreated {
+			t.Fatalf("replicated upload expected 201, got %d", uploadResp.StatusCode)
+		}
 
-	replicaReadURL := clusterHarness.VolumeAdminURL(1) + "/" + fid
-	var replicaBody []byte
-	if !waitForHTTPStatus(t, client, replicaReadURL, http.StatusOK, 10*time.Second, func(resp *http.Response) {
-		replicaBody = framework.ReadAllAndClose(t, resp)
-	}) {
-		t.Fatalf("replica did not become readable within deadline")
-	}
-	if string(replicaBody) != string(payload) {
-		t.Fatalf("replica body mismatch: got %q want %q", string(replicaBody), string(payload))
+		replicaReadURL := clusterHarness.VolumeAdminURL(1) + "/" + fid
+		var replicaBody []byte
+		if !waitForHTTPStatus(t, client, replicaReadURL, http.StatusOK, 10*time.Second, func(resp *http.Response) {
+			replicaBody = framework.ReadAllAndClose(t, resp)
+		}) {
+			t.Fatalf("replica did not become readable within deadline")
+		}
+		if string(replicaBody) != string(payload) {
+			t.Fatalf("replica body mismatch: got %q want %q", string(replicaBody), string(payload))
+		}
 	}
 }
 

test/volume_server/http/write_error_variants_test.go

@@ -1,12 +1,18 @@
 package volume_server_http_test
 
 import (
+	"bytes"
+	"context"
+	"mime/multipart"
 	"net/http"
+	"os"
+	"path/filepath"
 	"strings"
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/test/volume_server/framework"
 	"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 )
 
 func TestWriteInvalidVidAndFidReturnBadRequest(t *testing.T) {
@@ -72,3 +78,99 @@ func TestWriteMalformedMultipartAndMD5Mismatch(t *testing.T) {
 		t.Fatalf("content-md5 mismatch response should mention Content-MD5, got %q", string(md5MismatchBody))
 	}
 }
+
+func TestWriteReadOnlyReportsConflictAndSafeRefusal(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+	cluster := framework.StartVolumeCluster(t, matrix.P1())
+	conn, volume := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress())
+	defer conn.Close()
+	const volumeID = uint32(99)
+	framework.AllocateVolume(t, volume, volumeID, "")
+	client := framework.NewHTTPClient()
+	fid := framework.NewFileID(volumeID, 773001, 0x1A2B3C4D)
+	payload := []byte("existing bytes must survive a refused overwrite")
+	response := framework.DoRequest(t, client, newUploadRequest(t, cluster.VolumeAdminURL()+"/"+fid, payload))
+	body := framework.ReadAllAndClose(t, response)
+	if response.StatusCode != http.StatusCreated {
+		t.Fatalf("initial upload: %d %s", response.StatusCode, body)
+	}
+	if _, err := volume.VolumeMarkReadonly(context.Background(), &volume_server_pb.VolumeMarkReadonlyRequest{VolumeId: volumeID}); err != nil {
+		t.Fatal(err)
+	}
+	dat := filepath.Join(cluster.BaseDir(), "volume", "99.dat")
+	before, err := os.Stat(dat)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, query := range []string{"", "?fsync=true", "?fsync=%74rue", "?type=replicate", "?type=%72eplicate", "?%74ype=replicate", "multipart:first", "multipart:last"} {
+		t.Run(query, func(t *testing.T) {
+			request := newUploadRequest(t, cluster.VolumeAdminURL()+"/"+fid+query, []byte("refused replacement"))
+			if strings.HasPrefix(query, "multipart:") {
+				var multipartBody bytes.Buffer
+				writer := multipart.NewWriter(&multipartBody)
+				if query == "multipart:first" {
+					if err := writer.WriteField("type", "replicate"); err != nil {
+						t.Fatal(err)
+					}
+				}
+				file, err := writer.CreateFormFile("file", "data.bin")
+				if err != nil {
+					t.Fatal(err)
+				}
+				if _, err := file.Write([]byte("refused replacement")); err != nil {
+					t.Fatal(err)
+				}
+				if query == "multipart:last" {
+					if err := writer.WriteField("type", "replicate"); err != nil {
+						t.Fatal(err)
+					}
+				}
+				if err := writer.Close(); err != nil {
+					t.Fatal(err)
+				}
+				request, err = http.NewRequest(http.MethodPost, cluster.VolumeAdminURL()+"/"+fid, &multipartBody)
+				if err != nil {
+					t.Fatal(err)
+				}
+				request.Header.Set("Content-Type", writer.FormDataContentType())
+			}
+			response := framework.DoRequest(t, client, request)
+			body := framework.ReadAllAndClose(t, response)
+			if response.StatusCode != http.StatusConflict {
+				t.Fatalf("read-only upload: %d %s", response.StatusCode, body)
+			}
+			want := ""
+			if query == "" {
+				want = "refused-before-append"
+			}
+			if got := response.Header.Get("X-Seaweed-Write-Outcome"); got != want {
+				t.Fatalf("write outcome: %q, want %q", got, want)
+			}
+		})
+	}
+	after, err := os.Stat(dat)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if after.Size() != before.Size() {
+		t.Fatalf("refused write grew .dat from %d to %d", before.Size(), after.Size())
+	}
+	response = framework.DoRequest(t, client, mustNewRequest(t, http.MethodGet, cluster.VolumeAdminURL()+"/"+fid))
+	body = framework.ReadAllAndClose(t, response)
+	if response.StatusCode != http.StatusOK || !bytes.Equal(body, payload) {
+		t.Fatalf("existing data changed: %d %q", response.StatusCode, body)
+	}
+	if _, err := volume.VolumeMarkWritable(context.Background(), &volume_server_pb.VolumeMarkWritableRequest{VolumeId: volumeID}); err != nil {
+		t.Fatal(err)
+	}
+	response = framework.DoRequest(t, client, newUploadRequest(t, cluster.VolumeAdminURL()+"/"+fid, []byte("writable again")))
+	body = framework.ReadAllAndClose(t, response)
+	if response.StatusCode != http.StatusCreated {
+		t.Fatalf("unsealed upload: %d %s", response.StatusCode, body)
+	}
+	if got := response.Header.Get("X-Seaweed-Write-Outcome"); got != "" {
+		t.Fatalf("successful write marked refused: %q", got)
+	}
+}

weed/operation/upload_content.go

@@ -129,9 +129,10 @@ var (
 // the retry gate can decide on the status code instead of the message text.
 // StatusCode 0 means the request never got a response — a transport failure.
 type uploadStatusError struct {
-	StatusCode int
-	stalled    bool // the stall watchdog cancelled this attempt's request
-	err        error
+	StatusCode          int
+	refusedBeforeAppend bool
+	stalled             bool // the stall watchdog cancelled this attempt's request
+	err                 error
 }
 
 func (e *uploadStatusError) Error() string { return e.err.Error() }
@@ -140,19 +141,15 @@ func (e *uploadStatusError) Unwrap() error { return e.err }
 // ShouldReassignUpload reports whether an upload error means the client should
 // ask for a fresh volume assignment and retry on another volume.
 //
-// On the write path a volume server only 5xxs on a ReplicatedWrite failure
-// (local disk, replica peer down, or under-replication) — all of which a
-// different volume dodges — so any 5xx is reassignable. A status of 0 means the
-// assigned target never answered (down/unreachable), also reassignable. A 4xx
-// is a genuine client error and is surfaced. Errors without a status come from
-// the AssignVolume RPC or request setup; retry only transient transport ones.
+// Transport and 5xx failures retain their existing retry policy. A conflict
+// requires explicit proof that the upload was refused before append.
 func ShouldReassignUpload(err error) bool {
 	if err == nil {
 		return false
 	}
 	var se *uploadStatusError
 	if errors.As(err, &se) {
-		return se.StatusCode == 0 || se.StatusCode >= 500
+		return se.StatusCode == 0 || se.StatusCode >= 500 || (se.StatusCode == http.StatusConflict && se.refusedBeforeAppend)
 	}
 	return strings.Contains(err.Error(), "transport")
 }
@@ -243,6 +240,10 @@ func (uploader *Uploader) uploadWithRetryData(assignFn func() (fileId string, ho
 	if uploadOption.RetryForever {
 		util.RetryUntil("uploadWithRetryForever", doUploadFunc, func(err error) (shouldContinue bool) {
 			glog.V(0).Infof("upload content: %v", err)
+			var statusErr *uploadStatusError
+			if errors.As(err, &statusErr) && statusErr.StatusCode == http.StatusConflict {
+				return ShouldReassignUpload(err)
+			}
 			return true
 		})
 	} else {
@@ -365,8 +366,13 @@ func (uploader *Uploader) retriedUploadData(ctx context.Context, data []byte, op
 			return nil, ctx.Err()
 		}
 		var statusErr *uploadStatusError
-		if errors.As(err, &statusErr) && statusErr.stalled {
-			option.stallWatchdogDisarmed = true
+		if errors.As(err, &statusErr) {
+			if statusErr.StatusCode == http.StatusConflict {
+				return nil, err
+			}
+			if statusErr.stalled {
+				option.stallWatchdogDisarmed = true
+			}
 		}
 		glog.WarningfCtx(ctx, "uploading %d to %s: %v", i, option.UploadUrl, err)
 	}
@@ -664,22 +670,26 @@ func (uploader *Uploader) upload_content(ctx context.Context, fillBufferFunction
 		respStallTimer := time.AfterFunc(uploadStallMinWait, stallFired)
 		defer respStallTimer.Stop()
 	}
+	refusedBeforeAppend := resp.StatusCode == http.StatusConflict && resp.Header.Get("X-Seaweed-Write-Outcome") == "refused-before-append"
 	resp_body, ra_err := io.ReadAll(resp.Body)
 	uploadDone.Store(true)
 	if ra_err != nil {
 		stats.UploadErrorCounter.WithLabelValues(strconv.Itoa(resp.StatusCode)).Inc()
-		return nil, &uploadStatusError{StatusCode: resp.StatusCode, stalled: reqCtx.Err() != nil && ctx.Err() == nil, err: fmt.Errorf("read response body %v: %w", option.UploadUrl, ra_err)}
+		return nil, &uploadStatusError{StatusCode: resp.StatusCode, refusedBeforeAppend: refusedBeforeAppend, stalled: reqCtx.Err() != nil && ctx.Err() == nil, err: fmt.Errorf("read response body %v: %w", option.UploadUrl, ra_err)}
 	}
 
 	unmarshal_err := json.Unmarshal(resp_body, &ret)
 	if unmarshal_err != nil {
 		stats.UploadErrorCounter.WithLabelValues(strconv.Itoa(resp.StatusCode)).Inc()
 		glog.ErrorfCtx(ctx, "unmarshal %s: %v", option.UploadUrl, string(resp_body))
-		return nil, &uploadStatusError{StatusCode: resp.StatusCode, err: fmt.Errorf("unmarshal %v: %w", option.UploadUrl, unmarshal_err)}
+		return nil, &uploadStatusError{StatusCode: resp.StatusCode, refusedBeforeAppend: refusedBeforeAppend, err: fmt.Errorf("unmarshal %v: %w", option.UploadUrl, unmarshal_err)}
+	}
+	if resp.StatusCode == http.StatusConflict && ret.Error == "" {
+		ret.Error = http.StatusText(resp.StatusCode)
 	}
 	if ret.Error != "" {
 		stats.UploadErrorCounter.WithLabelValues(strconv.Itoa(resp.StatusCode)).Inc()
-		return nil, &uploadStatusError{StatusCode: resp.StatusCode, err: fmt.Errorf("unmarshalled error %v: %v", option.UploadUrl, ret.Error)}
+		return nil, &uploadStatusError{StatusCode: resp.StatusCode, refusedBeforeAppend: refusedBeforeAppend, err: fmt.Errorf("unmarshalled error %v: %v", option.UploadUrl, ret.Error)}
 	}
 	if bps := uploadObservedBps(bodySize, movedIn); bps > 0 {
 		uploader.fastUploadHosts.Store(req.URL.Host, fastHostMark{at: time.Now().UnixNano(), bps: bps})

weed/operation/upload_content_test.go

@@ -27,6 +27,7 @@ import (
 type scriptedHTTPResponse struct {
 	status int
 	body   string
+	header http.Header
 }
 
 type scriptedHTTPClient struct {
@@ -51,7 +52,7 @@ func (c *scriptedHTTPClient) Do(req *http.Request) (*http.Response, error) {
 
 	return &http.Response{
 		StatusCode: plan.status,
-		Header:     make(http.Header),
+		Header:     plan.header.Clone(),
 		Body:       io.NopCloser(strings.NewReader(plan.body)),
 	}, nil
 }
@@ -66,6 +67,8 @@ func TestShouldReassignUpload(t *testing.T) {
 		{name: "no response (transport)", err: &uploadStatusError{StatusCode: 0, err: fmt.Errorf("dial tcp: no such host")}, want: true},
 		{name: "500 replica write failed", err: &uploadStatusError{StatusCode: 500, err: fmt.Errorf("failed to write to replicas")}, want: true},
 		{name: "503 service unavailable", err: &uploadStatusError{StatusCode: 503, err: fmt.Errorf("busy")}, want: true},
+		{name: "proven conflict", err: &uploadStatusError{StatusCode: http.StatusConflict, refusedBeforeAppend: true, err: errors.New("read only")}, want: true},
+		{name: "unmarked conflict", err: &uploadStatusError{StatusCode: http.StatusConflict, err: errors.New("read only")}, want: false},
 		{name: "400 bad request", err: &uploadStatusError{StatusCode: 400, err: fmt.Errorf("bad needle")}, want: false},
 		{name: "401 unauthorized", err: &uploadStatusError{StatusCode: 401, err: fmt.Errorf("wrong jwt")}, want: false},
 		{name: "assign rpc transport", err: fmt.Errorf("assign volume: transport is closing"), want: true},
@@ -81,6 +84,52 @@ func TestShouldReassignUpload(t *testing.T) {
 	}
 }
 
+func TestUploadWithRetryDataReassignsOnProvenConflict(t *testing.T) {
+	for _, retryForever := range []bool{false, true} {
+		for _, outcome := range []string{"refused-before-append", "", "unknown"} {
+			t.Run(fmt.Sprintf("retryForever=%t/outcome=%s", retryForever, outcome), func(t *testing.T) {
+				httpClient := &scriptedHTTPClient{responses: map[string][]scriptedHTTPResponse{
+					"http://volume-a/1,first":  {{status: http.StatusConflict, body: `{}`, header: http.Header{"X-Seaweed-Write-Outcome": {outcome}}}},
+					"http://volume-b/2,second": {{status: http.StatusCreated, body: `{"name":"test.bin","size":3}`}},
+				}}
+				uploader := newUploader(httpClient)
+				assignments := 0
+				fid, _, err := uploader.uploadWithRetryData(func() (string, string, security.EncodedJwt, bool, error) {
+					assignments++
+					if assignments == 1 {
+						return "1,first", "volume-a", "", false, nil
+					}
+					return "2,second", "volume-b", "", false, nil
+				}, &UploadOption{Filename: "test.bin", RetryForever: retryForever}, []byte("abc"))
+				if outcome == "refused-before-append" {
+					if err != nil || fid != "2,second" || assignments != 2 || len(httpClient.calls) != 2 {
+						t.Fatalf("reassignment: fid=%s, assignments=%d, error=%v", fid, assignments, err)
+					}
+				} else if err == nil || fid != "1,first" || assignments != 1 || len(httpClient.calls) != 1 {
+					t.Fatalf("unproven conflict retried or acknowledged: assignments=%d, error=%v", assignments, err)
+				}
+			})
+		}
+	}
+}
+
+func TestUploadWithRetryDataRetryForeverPreservesAssignmentRetries(t *testing.T) {
+	httpClient := &scriptedHTTPClient{responses: map[string][]scriptedHTTPResponse{
+		"http://volume/1,first": {{status: http.StatusCreated, body: `{"name":"test.bin","size":3}`}},
+	}}
+	assignments := 0
+	fileID, _, err := newUploader(httpClient).uploadWithRetryData(func() (string, string, security.EncodedJwt, bool, error) {
+		assignments++
+		if assignments < 3 {
+			return "", "", "", false, errors.New("no writable volumes")
+		}
+		return "1,first", "volume", "", false, nil
+	}, &UploadOption{Filename: "test.bin", RetryForever: true}, []byte("abc"))
+	if err != nil || fileID != "1,first" || assignments != 3 || len(httpClient.calls) != 1 {
+		t.Fatalf("assignment retries: fid=%s, assignments=%d, uploads=%d, error=%v", fileID, assignments, len(httpClient.calls), err)
+	}
+}
+
 func TestUploadWithRetryDataReassignsOnVolumeSizeExceeded(t *testing.T) {
 	httpClient := &scriptedHTTPClient{
 		responses: map[string][]scriptedHTTPResponse{

weed/server/volume_server_handlers_write.go

@@ -55,7 +55,14 @@ func (vs *VolumeServer) PostHandler(w http.ResponseWriter, r *http.Request) {
 	isUnchanged, writeError := topology.ReplicatedWrite(context.WithoutCancel(ctx), vs.GetMaster, vs.grpcDialOption, vs.store, volumeId, reqNeedle, r, contentMd5)
 	if writeError != nil {
 		stats.VolumeServerFileWriteFailures.Inc()
-		writeJsonError(w, r, http.StatusInternalServerError, writeError)
+		status := http.StatusInternalServerError
+		if errors.Is(writeError, storage.ErrorReadOnly) {
+			status = http.StatusConflict
+		}
+		if errors.Is(writeError, topology.ErrorWriteRefused) {
+			w.Header().Set("X-Seaweed-Write-Outcome", "refused-before-append")
+		}
+		writeJsonError(w, r, status, writeError)
 		return
 	}
 

weed/storage/needle/needle_parse_upload.go

@@ -94,7 +94,7 @@ func ParseUpload(r *http.Request, sizeLimit int64, bytesBuffer *bytes.Buffer) (p
 		} else if n, err := util.GunzipStream(io.Discard, bytes.NewReader(pu.Data)); err == nil {
 			pu.OriginalDataSize = int(n)
 		}
-	} else if r.URL.Query().Get("type") != "replicate" {
+	} else if r.FormValue("type") != "replicate" {
 		// replica writes must keep the source needle's compression state, not re-derive it
 		ext := filepath.Base(pu.FileName)
 		mimeType := pu.MimeType
@@ -148,6 +148,9 @@ func parseUpload(r *http.Request, sizeLimit int64, pu *ParsedUpload) (e error) {
 	var dataSize int64
 
 	if r.Method == http.MethodPost && (contentType == "" || strings.Contains(contentType, "form-data")) {
+		if r.Form == nil {
+			_ = r.ParseForm()
+		}
 		form, fe := r.MultipartReader()
 
 		if fe != nil {
@@ -156,6 +159,19 @@ func parseUpload(r *http.Request, sizeLimit int64, pu *ParsedUpload) (e error) {
 			return
 		}
 
+		saveControlField := func(name string, data []byte) {
+			if name != "type" && name != "fsync" {
+				return
+			}
+			if len(r.Form[name]) == 0 {
+				// Longer values cannot match either control marker.
+				if len(data) > len("replicate")+1 {
+					data = data[:len("replicate")+1]
+				}
+				r.Form[name] = []string{string(data)}
+			}
+		}
+
 		// first multi-part item
 		part, fe := form.NextPart()
 		if fe != nil {
@@ -180,11 +196,15 @@ func parseUpload(r *http.Request, sizeLimit int64, pu *ParsedUpload) (e error) {
 		}
 		pu.Data = pu.bytesBuffer.Bytes()
 
+		if part.FileName() == "" {
+			saveControlField(part.FormName(), pu.Data)
+		}
+
 		contentType = part.Header.Get("Content-Type")
 		pu.ContentMd5 = part.Header.Get("Content-MD5")
 
-		// if the filename is empty string, do a search on the other multi-part items
-		for pu.FileName == "" {
+		// Find the first file and retain control fields before or after it.
+		for {
 			part2, fe := form.NextPart()
 			if fe != nil {
 				break // no more or on error, just safely break
@@ -193,7 +213,7 @@ func parseUpload(r *http.Request, sizeLimit int64, pu *ParsedUpload) (e error) {
 			fName := part2.FileName()
 
 			// found the first <file type> multi-part has filename
-			if fName != "" {
+			if pu.FileName == "" && fName != "" {
 				pu.bytesBuffer.Reset()
 				dataSize2, fe2 := pu.bytesBuffer.ReadFrom(io.LimitReader(part2, sizeLimit+1))
 				if fe2 != nil {
@@ -212,7 +232,14 @@ func parseUpload(r *http.Request, sizeLimit int64, pu *ParsedUpload) (e error) {
 				contentType = part2.Header.Get("Content-Type")
 				pu.ContentMd5 = part2.Header.Get("Content-MD5")
 				part = part2
-				break
+				continue
+			}
+			if fName == "" && (part2.FormName() == "type" || part2.FormName() == "fsync") {
+				value, err := io.ReadAll(io.LimitReader(part2, int64(len("replicate")+1)))
+				if err != nil {
+					return err
+				}
+				saveControlField(part2.FormName(), value)
 			}
 		}
 

weed/storage/needle/needle_parse_upload_stream_test.go

@@ -167,3 +167,53 @@ func buildGzipReq(t *testing.T, body []byte, md5OnPart, md5OnReq string) *http.R
 	}
 	return req
 }
+
+func TestParseUploadRetainsReplicaControls(t *testing.T) {
+	for _, fieldFirst := range []bool{true, false} {
+		for _, query := range []string{"", "?type=client&fsync=false", "?type=&fsync="} {
+			var body bytes.Buffer
+			writer := multipart.NewWriter(&body)
+			fields := func() {
+				if err := writer.WriteField("type", "replicate"); err != nil {
+					t.Fatal(err)
+				}
+				if err := writer.WriteField("fsync", "true"); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if fieldFirst {
+				fields()
+			}
+			file, err := writer.CreateFormFile("file", "data.bin")
+			if err != nil {
+				t.Fatal(err)
+			}
+			if _, err := file.Write([]byte("payload")); err != nil {
+				t.Fatal(err)
+			}
+			if !fieldFirst {
+				fields()
+			}
+			if err := writer.Close(); err != nil {
+				t.Fatal(err)
+			}
+			request := httptest.NewRequest(http.MethodPost, "/1,12345678"+query, &body)
+			request.Header.Set("Content-Type", writer.FormDataContentType())
+			upload, err := ParseUpload(request, 1024, &bytes.Buffer{})
+			if err != nil {
+				t.Fatal(err)
+			}
+			if string(upload.Data) != "payload" {
+				t.Fatalf("file data = %q", upload.Data)
+			}
+			for name, want := range map[string]string{"type": "replicate", "fsync": "true"} {
+				if request.URL.Query().Has(name) {
+					want = request.URL.Query().Get(name)
+				}
+				if got := request.FormValue(name); got != want {
+					t.Fatalf("field-first=%v query=%q %s=%q, want %q", fieldFirst, query, name, got, want)
+				}
+			}
+		}
+	}
+}

weed/storage/store.go

@@ -954,7 +954,7 @@ func (s *Store) Close() {
 func (s *Store) WriteVolumeNeedle(i needle.VolumeId, n *needle.Needle, checkCookie bool, fsync bool) (isUnchanged bool, err error) {
 	if v := s.findVolume(i); v != nil {
 		if v.IsReadOnly() {
-			err = fmt.Errorf("volume %d is read only", i)
+			err = fmt.Errorf("volume %d is %w", i, ErrorReadOnly)
 			return
 		}
 		_, _, isUnchanged, err = v.writeNeedle2(n, checkCookie, fsync, s.isStopping.Load())

weed/storage/volume_write.go

@@ -17,6 +17,7 @@ import (
 var ErrorNotFound = errors.New("not found")
 var ErrorDeleted = errors.New("already deleted")
 var ErrorSizeMismatch = errors.New("size mismatch")
+var ErrorReadOnly = errors.New("read only")
 
 // isFileUnchanged checks whether this needle to write is same as last one.
 // It requires serialized access in the same volume.
@@ -182,7 +183,7 @@ func (v *Volume) syncWrite(n *needle.Needle, checkCookie bool, fsync bool) (offs
 		return 0, 0, false, err
 	}
 	if v.IsReadOnly() {
-		return 0, 0, false, fmt.Errorf("volume %d is read only", v.Id)
+		return 0, 0, false, fmt.Errorf("volume %d is %w", v.Id, ErrorReadOnly)
 	}
 
 	// A caller can still hold the volume after it was closed or destroyed, which
@@ -314,7 +315,7 @@ func (v *Volume) appendWriteRequest(n *needle.Needle, checkCookie bool, fsync bo
 	// Re-check under the write lock: the outer checks run before the needle
 	// is assembled, and VolumeMarkReadonly can land in between.
 	if v.IsReadOnly() {
-		return 0, 0, false, fmt.Errorf("volume %d is read only", v.Id)
+		return 0, 0, false, fmt.Errorf("volume %d is %w", v.Id, ErrorReadOnly)
 	}
 	if v.isFileUnchanged(n) {
 		size = Size(n.DataSize)
@@ -564,7 +565,7 @@ func (v *Volume) processBatchRun(run []*needle.AsyncRequest) {
 			if unavailableErr := v.UnavailableError(); unavailableErr != nil {
 				run[i].UpdateResult(0, 0, false, unavailableErr)
 			} else if v.IsReadOnly() {
-				run[i].UpdateResult(0, 0, false, fmt.Errorf("volume %d is read only", v.Id))
+				run[i].UpdateResult(0, 0, false, fmt.Errorf("volume %d is %w", v.Id, ErrorReadOnly))
 			} else {
 				offset, size, isUnchanged, err := v.appendWriteRequest(run[i].N, true, true)
 				if err == nil {

weed/topology/store_replicate.go

@@ -25,6 +25,8 @@ import (
 	"google.golang.org/grpc"
 )
 
+var ErrorWriteRefused = errors.New("write refused before append")
+
 // ReplicatedWrite writes a needle to the local volume and fans it out to all
 // remote replica locations. When type=replicate is set, the request is itself
 // a forwarded replication and no further remote lookups are performed.
@@ -198,6 +200,9 @@ func ReplicatedWrite(ctx context.Context, masterFn operation.GetMasterFn, grpcDi
 	wg.Wait()
 
 	if localErr != nil {
+		if errors.Is(localErr, storage.ErrorReadOnly) && replicaCount == 0 && !fsync && r.FormValue("type") != "replicate" {
+			localErr = fmt.Errorf("%w: %w", ErrorWriteRefused, localErr)
+		}
 		return false, localErr
 	}
 	if remoteErr != nil {

weed/topology/store_replicate_test.go

@@ -361,6 +361,7 @@ func TestReplicatedWriteFailsWhenEitherLegFails(t *testing.T) {
 	}{
 		{name: "local write fails", volumeId: 8, readonlyLocal: true, replicaStatus: http.StatusCreated},
 		{name: "replica write fails", volumeId: 9, replicaStatus: http.StatusInternalServerError},
+		{name: "replica read only", volumeId: 10, replicaStatus: http.StatusConflict},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			replica := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -396,9 +397,13 @@ func TestReplicatedWriteFailsWhenEitherLegFails(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			if _, err := ReplicatedWrite(context.Background(), masterFn, dialOption, store, tc.volumeId, n, r, ""); err == nil {
+			_, err := ReplicatedWrite(context.Background(), masterFn, dialOption, store, tc.volumeId, n, r, "")
+			if err == nil {
 				t.Fatal("ReplicatedWrite succeeded despite a failed leg")
 			}
+			if errors.Is(err, ErrorWriteRefused) {
+				t.Fatalf("replicated write incorrectly proves no append: %v", err)
+			}
 		})
 	}
 }