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 <noreply@anthropic.com>

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 {

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