volume: never finish serving a needle whose data fails its CRC (#11467)

* storage: hold back last chunk until CRC verifies on whole-needle reads

Above PagedReadLimit the needle is streamed: headers and body go out
before the checksum is computed, so a corrupted needle was served as
200 with bad bytes and readers could not fall back to a replica. The
final chunk is now written only after the checksum verifies; on a
mismatch the response ends short of Content-Length and the client sees
a failed transfer.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* rust volume: verify needle checksum on streamed reads (parity)

Mirror the Go fix: carry the needle checksum in NeedleStreamInfo and
have StreamingBody accumulate the CRC and verify it before emitting the
last frame; a mismatch ends the body with an error so the client sees
the transfer fail rather than receiving corrupt bytes that look
complete.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* server: abort the transfer when a committed stream fails

A writeFn failure after any byte or WriteHeader call leaves the declared
status and Content-Length already sent; http.Error's text then joins the
body and can exactly fill the withheld tail of a corrupted needle read —
the client sees a complete 200 instead of a failed transfer to retry.

Track whether the response is committed (headers sent, or bytes buffered
for the deferred flush) and panic with http.ErrAbortHandler instead of
appending an error body; pre-commit failures keep the 500 path.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* server: drop the response writer wrapper from the committed-response check

Counting buffered writes is enough: with no bytes buffered the status and
headers cannot have gone out, and the range branches commit via the
explicit WriteHeader call before writeFn runs. The extra ResponseWriter
wrapper added a new Write sink site that CodeQL flags.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

seaweed-volume/src/server/handlers.rs

@@ -21,6 +21,7 @@ use super::volume_server::{VolumeServerState, normalize_outgoing_http_url, to_ht
 use crate::config::ReadMode;
 use crate::metrics;
 use crate::pb::volume_server_pb;
+use crate::storage::needle::CRC;
 use crate::storage::needle::needle::Needle;
 use crate::storage::types::*;
 
@@ -167,6 +168,10 @@ struct StreamingBody {
     /// Compaction revision at the time of the initial read; if the volume's revision
     /// changes between chunks, the needle may have moved and we must re-lookup its offset.
     compaction_revision: u16,
+    /// Checksum stored in the needle tail.
+    expected_checksum: u32,
+    /// Running checksum over the emitted data.
+    crc: CRC,
 }
 
 impl http_body::Body for StreamingBody {
@@ -187,7 +192,32 @@ impl http_body::Body for StreamingBody {
                         match result {
                             Ok(Ok(chunk)) => {
                                 let len = chunk.len();
+                                self.crc = self.crc.update(&chunk);
                                 self.pos += len;
+                                // The last frame is held back until the checksum verifies
+                                // (Go parity: readNeedleDataInto). On a mismatch it is
+                                // never emitted, so the response ends short of the declared
+                                // Content-Length and the reader sees a failed transfer.
+                                if self.pos >= self.data_size as usize {
+                                    let ok = self.expected_checksum == self.crc.0
+                                        || self.expected_checksum == self.crc.legacy_value();
+                                    if !ok {
+                                        metrics::HANDLER_COUNTER
+                                            .with_label_values(&[metrics::ERROR_CRC])
+                                            .inc();
+                                        return std::task::Poll::Ready(Some(Err(
+                                            std::io::Error::new(
+                                                std::io::ErrorKind::InvalidData,
+                                                format!(
+                                                    "needle data checksum {} expected {} for needle {}",
+                                                    self.crc.0,
+                                                    self.expected_checksum,
+                                                    self.needle_id
+                                                ),
+                                            ),
+                                        )));
+                                    }
+                                }
                                 return std::task::Poll::Ready(Some(Ok(http_body::Frame::data(
                                     chunk,
                                 ))));
@@ -1512,6 +1542,8 @@ async fn get_or_head_handler_inner(
             volume_id: info.volume_id,
             needle_id: info.needle_id,
             compaction_revision: info.compaction_revision,
+            expected_checksum: info.checksum,
+            crc: CRC(0),
         };
 
         let body = Body::new(streaming);
@@ -4583,4 +4615,127 @@ mod tests {
 
         server.abort();
     }
+
+    fn streaming_test_state() -> Arc<VolumeServerState> {
+        use crate::security::{Guard, SigningKey};
+        use crate::server::volume_server::RuntimeMetricsConfig;
+        use crate::storage::needle_map::NeedleMapKind;
+        use crate::storage::store::Store;
+        use std::sync::RwLock;
+        use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU32};
+
+        Arc::new(VolumeServerState {
+            store: RwLock::new(Store::new(NeedleMapKind::InMemory)),
+            guard: RwLock::new(Guard::new(
+                &[],
+                SigningKey(vec![]),
+                0,
+                SigningKey(vec![]),
+                0,
+            )),
+            is_stopping: RwLock::new(false),
+            maintenance: AtomicBool::new(false),
+            state_version: AtomicU32::new(0),
+            concurrent_upload_limit: 0,
+            concurrent_download_limit: 0,
+            inflight_upload_data_timeout: std::time::Duration::ZERO,
+            inflight_download_data_timeout: std::time::Duration::ZERO,
+            inflight_upload_bytes: AtomicI64::new(0),
+            inflight_download_bytes: AtomicI64::new(0),
+            upload_notify: tokio::sync::Notify::new(),
+            download_notify: tokio::sync::Notify::new(),
+            data_center: String::new(),
+            rack: String::new(),
+            file_size_limit_bytes: 0,
+            maintenance_byte_per_second: 0,
+            is_heartbeating: AtomicBool::new(false),
+            has_master: false,
+            pre_stop_seconds: 0,
+            volume_state_notify: tokio::sync::Notify::new(),
+            write_queue: std::sync::OnceLock::new(),
+            read_mode: crate::config::ReadMode::Local,
+            allow_untrusted_remote_endpoints: false,
+            master_url: String::new(),
+            master_urls: Vec::new(),
+            seed_master_set: std::collections::HashSet::new(),
+            current_master_url: tokio::sync::RwLock::new(String::new()),
+            self_url: String::new(),
+            http_client: reqwest::Client::new(),
+            outgoing_http_scheme: "http".to_string(),
+            outgoing_grpc_tls: None,
+            metrics_runtime: RwLock::new(RuntimeMetricsConfig::default()),
+            metrics_notify: tokio::sync::Notify::new(),
+            fix_jpg_orientation: false,
+            has_slow_read: false,
+            read_buffer_size_bytes: 4 * 1024 * 1024,
+            security_file: String::new(),
+            cli_white_list: vec![],
+            state_file_path: String::new(),
+        })
+    }
+
+    /// Go's readNeedleDataInto holds back the last chunk of a whole-needle read
+    /// until the checksum verifies; a streamed needle that fails its CRC must
+    /// likewise end short of the declared length instead of looking complete.
+    #[tokio::test]
+    async fn test_streaming_body_checksum_failure_drops_last_chunk() {
+        use futures::StreamExt;
+        use std::io::Write;
+
+        let data: Vec<u8> = (0..10_000u32).map(|i| (i % 251) as u8).collect();
+        let mut tmp = tempfile::NamedTempFile::new().unwrap();
+        tmp.write_all(&data).unwrap();
+        let file = tmp.reopen().unwrap();
+
+        let control = Arc::new(crate::storage::volume::DataFileAccessControl::default());
+        let mk_body = |expected_checksum: u32, file: &std::fs::File| StreamingBody {
+            source: crate::storage::volume::NeedleStreamSource::Local(file.try_clone().unwrap()),
+            data_offset: 0,
+            data_size: data.len() as u32,
+            pos: 0,
+            chunk_size: 4096,
+            data_file_access_control: control.clone(),
+            hold_read_lock_for_stream: true,
+            _held_read_lease: None,
+            pending: None,
+            state: None,
+            tracked_bytes: 0,
+            server_state: streaming_test_state(),
+            volume_id: VolumeId(1),
+            needle_id: NeedleId(7),
+            compaction_revision: 0,
+            expected_checksum,
+            crc: CRC(0),
+        };
+
+        // Healthy needle: every byte is delivered.
+        let body = Body::new(mk_body(CRC::new(&data).0, &file));
+        let mut stream = body.into_data_stream();
+        let mut got = 0usize;
+        while let Some(item) = stream.next().await {
+            got += item.expect("healthy stream errored").len();
+        }
+        assert_eq!(got, data.len());
+
+        // Corrupted needle: the last chunk fails the checksum and is never
+        // emitted — the body ends short of the data size.
+        let body = Body::new(mk_body(CRC::new(&data).0 ^ 0xff, &file));
+        let mut stream = body.into_data_stream();
+        let mut got = 0usize;
+        let mut saw_err = false;
+        while let Some(item) = stream.next().await {
+            match item {
+                Ok(bytes) => got += bytes.len(),
+                Err(e) => {
+                    assert!(
+                        e.to_string().contains("checksum"),
+                        "unexpected stream error: {e}"
+                    );
+                    saw_err = true;
+                }
+            }
+        }
+        assert!(saw_err, "corrupted stream must fail");
+        assert!(got < data.len(), "delivered {got} of {} bytes", data.len());
+    }
 }

seaweed-volume/src/storage/volume.rs

@@ -606,6 +606,9 @@ pub struct NeedleStreamInfo {
     /// streaming, the needle's disk offset must be re-read from the needle map because
     /// compaction may have moved the needle to a different location.
     pub compaction_revision: u16,
+    /// Checksum stored in the needle tail, verified once the last chunk has
+    /// been read — before that frame is emitted.
+    pub checksum: u32,
 }
 
 #[derive(Clone)]
@@ -1811,6 +1814,7 @@ impl Volume {
             volume_id: self.id,
             needle_id: n.id,
             compaction_revision: self.super_block.compaction_revision,
+            checksum: n.checksum.0,
         })
     }