diff --git a/other/java/client/src/main/java/seaweedfs/client/SeaweedWrite.java b/other/java/client/src/main/java/seaweedfs/client/SeaweedWrite.java index c568007214..d961047b59 100644 --- a/other/java/client/src/main/java/seaweedfs/client/SeaweedWrite.java +++ b/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); diff --git a/other/java/client/src/test/java/seaweedfs/client/SeaweedWriteIntegrationTest.java b/other/java/client/src/test/java/seaweedfs/client/SeaweedWriteIntegrationTest.java new file mode 100644 index 0000000000..4b36f3fdf7 --- /dev/null +++ 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 payloads = Collections.synchronizedList(new ArrayList<>()); + private final List paths = Collections.synchronizedList(new ArrayList<>()); + private final List auth = Collections.synchronizedList(new ArrayList<>()); + private final List 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 observer) { + observer.onNext(FilerProto.GetFilerConfigurationResponse.newBuilder().setCipher(cipher).build()); + observer.onCompleted(); + } + + @Override + public void assignVolume(FilerProto.AssignVolumeRequest request, + StreamObserver 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); + } + } + } +} diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 409aaa734a..e6051a71a9 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/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, 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, 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); } diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 14a40e7dcf..a6f20a0ddf 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/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, 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, diff --git a/test/volume_server/http/replication_lifecycle_test.go b/test/volume_server/http/replication_lifecycle_test.go index 3a4f392edd..2bf73d7046 100644 --- a/test/volume_server/http/replication_lifecycle_test.go +++ b/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)) + } } } diff --git a/test/volume_server/http/write_error_variants_test.go b/test/volume_server/http/write_error_variants_test.go index 2f858b41bc..bd37b2a15e 100644 --- a/test/volume_server/http/write_error_variants_test.go +++ b/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) + } +} diff --git a/weed/operation/upload_content.go b/weed/operation/upload_content.go index 081d295e66..a44dc38e8e 100644 --- a/weed/operation/upload_content.go +++ b/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}) diff --git a/weed/operation/upload_content_test.go b/weed/operation/upload_content_test.go index d80c8cd591..9e0a6e63a1 100644 --- a/weed/operation/upload_content_test.go +++ b/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{ diff --git a/weed/server/volume_server_handlers_write.go b/weed/server/volume_server_handlers_write.go index 149409fc09..87f0f0d9b8 100644 --- a/weed/server/volume_server_handlers_write.go +++ b/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 } diff --git a/weed/storage/needle/needle_parse_upload.go b/weed/storage/needle/needle_parse_upload.go index 99b4e13ca4..0707acef14 100644 --- a/weed/storage/needle/needle_parse_upload.go +++ b/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 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) } } diff --git a/weed/storage/needle/needle_parse_upload_stream_test.go b/weed/storage/needle/needle_parse_upload_stream_test.go index a3cd7ba759..4658cf125c 100644 --- a/weed/storage/needle/needle_parse_upload_stream_test.go +++ b/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) + } + } + } + } +} diff --git a/weed/storage/store.go b/weed/storage/store.go index 21c5efe5a1..6756983e0b 100644 --- a/weed/storage/store.go +++ b/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()) diff --git a/weed/storage/volume_write.go b/weed/storage/volume_write.go index f119425fff..f4597b2937 100644 --- a/weed/storage/volume_write.go +++ b/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 { diff --git a/weed/topology/store_replicate.go b/weed/topology/store_replicate.go index 7ef9ca5694..d130d3200a 100644 --- a/weed/topology/store_replicate.go +++ b/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 { diff --git a/weed/topology/store_replicate_test.go b/weed/topology/store_replicate_test.go index 74eadaa48c..d25c574028 100644 --- a/weed/topology/store_replicate_test.go +++ b/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) + } }) } }