From c1ccbcda13ab43d037f56e0cb13249e6f3026abf Mon Sep 17 00:00:00 2001 From: Eliah Rusin Date: Thu, 24 Sep 2026 02:08:36 +0300 Subject: [PATCH] volume server: the write queue answers uploads with the needle's real ETag (#11414) With SEAWEED_WRITE_QUEUE=1 every upload came back with ETag "00000000". The upload handler built the needle with Needle::default(), so its checksum was CRC(0), and handed a clone of it to the queue. The CRC was only computed in the write path, on the worker's clone, and WriteResult carries no checksum back, so n.etag() in the handler formatted the zero checksum. The direct path writes through &mut n and was correct. Compute the checksum in the handler while building the needle, the way Go's CreateNeedleFromRequest does, over the same bytes the write path hashes (the stored data, gzipped or not). The ETag and the has-name flag are read before the write, so the needle is moved into the queue instead of cloned, which also drops a full payload copy per queued upload. Co-authored-by: Claude Fable 5.1 --- seaweed-volume/src/server/handlers.rs | 17 ++-- seaweed-volume/tests/http_integration.rs | 99 ++++++++++++++++++++++++ 2 files changed, 111 insertions(+), 5 deletions(-) diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index a204b2dc2..55ab9ea5d 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -2664,8 +2664,15 @@ pub async fn post_handler( // sees it the same way the primary did. let fsync = form_value("fsync").as_deref() == Some("true"); + // Go computes the checksum while building the needle (CreateNeedleFromRequest). + // The write queue takes the needle away, so the response fields that come + // from it are read here, before the write, rather than from a copy. + n.checksum = crate::storage::needle::crc::CRC::new(&n.data); + let needle_etag = n.etag(); + let needle_has_name = n.has_name(); + let write_result = if let Some(wq) = state.write_queue.get() { - wq.submit(vid, n.clone(), fsync).await + wq.submit(vid, n, fsync).await } else { let mut store = state.store.write().unwrap(); store.write_volume_needle(vid, &mut n, fsync) @@ -2716,22 +2723,22 @@ pub async fn post_handler( let resp = match write_result { Ok((_offset, _size, is_unchanged)) => { if is_unchanged { - let etag = format!("\"{}\"", n.etag()); + let etag = format!("\"{}\"", needle_etag); (StatusCode::NO_CONTENT, [(header::ETAG, etag)]).into_response() } else { // Go only includes contentMd5 when the client provided Content-MD5 let result = UploadResult { - name: if n.has_name() { + name: if needle_has_name { filename.clone() } else { String::new() }, size: original_data_size, // H3: use original size, not compressed - etag: n.etag(), + etag: needle_etag.clone(), mime: mime_type.clone(), content_md5: original_content_md5.clone(), }; - let etag = n.etag(); + let etag = needle_etag; let etag_header = if etag.starts_with('"') { etag.clone() } else { diff --git a/seaweed-volume/tests/http_integration.rs b/seaweed-volume/tests/http_integration.rs index bcb4a8f86..43897c9b2 100644 --- a/seaweed-volume/tests/http_integration.rs +++ b/seaweed-volume/tests/http_integration.rs @@ -1255,3 +1255,102 @@ async fn non_ascii_fid_and_ttl_are_rejected_not_panicked() { "a non-ASCII ttl must behave like any other invalid ttl, not panic" ); } + +// ============================================================================ +// The write queue answers an upload with the needle's real ETag +// +// The queue worker computes the CRC on the needle it was handed, so the handler +// has to know the checksum before it submits. Without that every queued upload +// came back as "00000000". The direct path is the reference: same payload, same +// ETag, for a plain body and for one the handler gzips before storing. +// ============================================================================ + +#[tokio::test] +async fn write_queue_upload_returns_same_etag_as_direct_write() { + use seaweed_volume::server::write_queue::WriteQueue; + + let (direct_state, _direct_tmp) = test_state(); + let (queued_state, _queued_tmp) = test_state(); + let wq = WriteQueue::new(queued_state.clone(), 128); + let _ = queued_state.write_queue.set(wq); + + let compressible = "seaweedfs ".repeat(200).into_bytes(); + let uploads: [(&str, &[u8]); 2] = [ + ("/1,01637037d6", b"hello, seaweedfs!"), + ("/1/02637037d6/notes.txt", &compressible), + ]; + + for (uri, payload) in uploads { + let mut etags = Vec::new(); + for state in [&direct_state, &queued_state] { + let app = build_admin_router(state.clone()); + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri(uri) + .body(Body::from(payload.to_vec())) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::CREATED); + + let header = response + .headers() + .get("ETag") + .expect("upload response has no ETag") + .to_str() + .unwrap() + .to_string(); + let body = body_bytes(response).await; + let json: serde_json::Value = + serde_json::from_slice(&body).expect("POST response is not valid JSON"); + let etag = json["eTag"].as_str().unwrap().to_string(); + assert_eq!(header, format!("\"{}\"", etag)); + etags.push(etag); + } + assert_ne!(etags[0], "00000000", "{}: direct ETag is the zero CRC", uri); + assert_eq!( + etags[1], etags[0], + "{}: queued upload must return the direct path's ETag", + uri + ); + } + + // The second upload really was stored gzipped, so its ETag is the CRC of + // the compressed bytes on both paths. + let app = build_admin_router(queued_state.clone()); + let response = app + .oneshot( + Request::builder() + .uri(uploads[1].0) + .header("Accept-Encoding", "gzip") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response.headers()["Content-Encoding"], "gzip"); + + // Re-uploading the same bytes is the unchanged path: 204 with the same ETag. + let (uri, payload) = uploads[0]; + let mut etags = Vec::new(); + for state in [&direct_state, &queued_state] { + let app = build_admin_router(state.clone()); + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri(uri) + .body(Body::from(payload.to_vec())) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::NO_CONTENT); + etags.push(response.headers()["ETag"].to_str().unwrap().to_string()); + } + assert_eq!(etags[1], etags[0], "unchanged upload ETag differs"); +}