fix(volume): skip unnecessary needle data reads on delete (#11711)

* fix(volume): do not read a needle's data to delete it

A delete read the needle's whole payload from disk only to discard it.
The delete handler and BatchDelete need the needle's cookie and its
chunk-manifest flag, which live in the header and trailer, but fetched
them with a full needle read, the same call a GET uses.

readNeedle now reads the header and trailer only when MustMetaOnly is
set, whatever the needle's size, and the delete paths ask for that. The
Rust volume server honours must_meta_only the same way for its delete
paths. A compressed needle or a chunk manifest still gets the full read,
as before, because their callers need the bytes.

HEAD is unchanged. It used to set MustMetaOnly, which nothing read; it
now asks for AttemptMetaOnly alone, so it keeps reading a needle up to
PagedReadLimit whole. On a spinning disk that one read is cheaper than a
header read followed by a trailer read.

Version 1 needles are always read in full. They have no data size field
and no metadata after the data, which ReadNeedleMeta relies on, so a
metadata-only read of one failed with "invalid needle meta size". That
already happened for version 1 needles above PagedReadLimit read with
AttemptMetaOnly.

A delete therefore no longer verifies the data CRC of the needle it
removes. The cookie check is unchanged.

* fix(volume): read transformed images fully on HEAD too

A HEAD carrying resize or crop params reports the transformed length,
which a metadata read cannot produce. Go and Rust both answered such
HEADs from the source header, returning the stored size instead of the
cropped or resized one.

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

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

---------

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

seaweed-volume/src/server/grpc_server.rs

@@ -986,7 +986,12 @@ impl VolumeServer for VolumeGrpcService {
                 let original_cookie = n.cookie;
                 if !is_ec_volume {
                     let store = self.state.store.read().unwrap();
-                    match store.read_volume_needle(file_id.volume_id, &mut n) {
+                    let mut read_option = crate::server::handlers::delete_read_option();
+                    match store.read_volume_needle_with_option(
+                        file_id.volume_id,
+                        &mut n,
+                        &mut read_option,
+                    ) {
                         Ok(_) => {}
                         Err(e) => {
                             results.push(volume_server_pb::DeleteResult {
@@ -10997,6 +11002,50 @@ mod tests {
         (r.status, r.error.clone())
     }
 
+    /// A batch delete only needs the needle's cookie, which is stored in the
+    /// header, so it must not read the payload.
+    #[tokio::test]
+    async fn batch_delete_reads_needle_meta_only() {
+        use crate::storage::volume::needle_read_hook;
+        use std::sync::Mutex;
+
+        const ID: u64 = 0x6e7a_0b01;
+        let (service, _tmp) = make_local_service_with_volume("", None);
+        let data = vec![7u8; 512 * 1024];
+        let mut n = Needle {
+            id: NeedleId(ID),
+            cookie: Cookie(0x3344),
+            data: data.clone(),
+            data_size: data.len() as u32,
+            ..Needle::default()
+        };
+        service
+            .state
+            .store
+            .write()
+            .unwrap()
+            .write_volume_needle(VolumeId(1), &mut n, false)
+            .unwrap();
+
+        let reads = Arc::new(Mutex::new(Vec::new()));
+        let _hook = needle_read_hook::register(NeedleId(ID), {
+            let reads = reads.clone();
+            move |len| reads.lock().unwrap().push(len)
+        });
+
+        let (status, _) = batch_delete_1(&service, ID, 0x3345, false).await;
+        assert_eq!(status, 400);
+        let (status, error) = batch_delete_1(&service, ID, 0x3344, false).await;
+        assert_eq!(status, 202, "{error}");
+
+        let reads = reads.lock().unwrap().clone();
+        assert!(
+            !reads.is_empty() && reads.iter().all(|&len| len < 4096),
+            "a batch delete must not read the needle's payload ({} bytes): {reads:?}",
+            data.len()
+        );
+    }
+
     const GO_EC_NEEDLE_NOT_FOUND: &str =
         "locate in local ec volume: FindNeedleFromEcx: needle not found";
 

seaweed-volume/src/server/handlers.rs

@@ -161,9 +161,10 @@ impl SourceReadRequest {
 
     /// HEAD, a range, or a stream.
     fn served_from_source(&self, n: &Needle) -> bool {
-        // HEAD is meta-only unless the payload is needed to expand a manifest.
         if self.is_head {
-            return !(n.is_chunk_manifest() && !self.bypass_cm);
+            // HEAD is meta-only unless the payload is needed to expand a
+            // manifest or to answer an image transform's length.
+            return !self.has_image_ops && !(n.is_chunk_manifest() && !self.bypass_cm);
         }
         self.direct(n) && (self.has_range || n.data_size > STREAMING_THRESHOLD)
     }
@@ -171,8 +172,10 @@ impl SourceReadRequest {
     /// Meta first only if the needle could be served from source; its data
     /// size is below its index size.
     fn try_meta_first(&self, size: Size) -> bool {
-        self.is_head
-            || (!self.has_image_ops && (self.has_range || size.0 > STREAMING_THRESHOLD as i32))
+        if self.has_image_ops {
+            return false;
+        }
+        self.is_head || self.has_range || size.0 > STREAMING_THRESHOLD as i32
     }
 }
 
@@ -3058,6 +3061,20 @@ struct DeleteResult {
     size: i64,
 }
 
+/// Read option for deletes: read only the needle's header and trailer.
+///
+/// A delete needs to know that the needle exists, what its cookie is, and
+/// whether it is a chunk manifest. None of that requires the payload. If the
+/// needle turns out to be a chunk manifest, the read loads its data as well,
+/// because the delete needs the manifest to delete the chunks.
+pub(crate) fn delete_read_option() -> ReadOption {
+    ReadOption {
+        attempt_meta_only: true,
+        must_meta_only: true,
+        ..ReadOption::default()
+    }
+}
+
 pub async fn delete_handler(
     State(state): State<Arc<VolumeServerState>>,
     request: Request<Body>,
@@ -3226,7 +3243,7 @@ pub async fn delete_handler(
     let original_cookie = cookie;
     {
         let store = state.store.read().unwrap();
-        match store.read_volume_needle(vid, &mut n) {
+        match store.read_volume_needle_with_option(vid, &mut n, &mut delete_read_option()) {
             Ok(_) => {}
             Err(_) => {
                 let result = DeleteResult { size: 0 };
@@ -5523,6 +5540,17 @@ mod tests {
         (status, headers, body)
     }
 
+    /// The public router does not accept DELETE requests, so this calls the
+    /// handler directly, the same way the admin router does.
+    async fn send_delete(state: &Arc<VolumeServerState>, path: &str) -> StatusCode {
+        let request = Request::builder()
+            .method(Method::DELETE)
+            .uri(path)
+            .body(Body::empty())
+            .unwrap();
+        delete_handler(State(state.clone()), request).await.status()
+    }
+
     /// A GET whose needle read is parked must not hold the store lock: a
     /// writer, here a real append to the same volume, must get through.
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@@ -5627,6 +5655,43 @@ mod tests {
         );
     }
 
+    /// A delete needs the needle's cookie and its chunk-manifest flag. Both
+    /// are stored in the header and trailer, so the delete must not read the
+    /// payload.
+    #[tokio::test]
+    async fn test_delete_reads_needle_meta_only() {
+        use crate::storage::volume::needle_read_hook;
+        use std::sync::Mutex;
+
+        const ID: u64 = 0x6e7a_0a01;
+        let tmp = tempfile::TempDir::new().unwrap();
+        let state = volume_test_state(&tmp);
+        let data: Vec<u8> = (0..512 * 1024).map(|i| (i % 251) as u8).collect();
+        let path = put_test_needle(&state, ID, &data);
+
+        let reads = Arc::new(Mutex::new(Vec::new()));
+        let _hook = needle_read_hook::register(NeedleId(ID), {
+            let reads = reads.clone();
+            move |len| reads.lock().unwrap().push(len)
+        });
+
+        let wrong_cookie = format!("/1,{:x}{:08x}", ID, TEST_COOKIE + 1);
+        assert_eq!(
+            send_delete(&state, &wrong_cookie).await,
+            StatusCode::BAD_REQUEST
+        );
+        assert_eq!(send_delete(&state, &path).await, StatusCode::ACCEPTED);
+
+        let reads = reads.lock().unwrap().clone();
+        assert!(
+            !reads.is_empty() && reads.iter().all(|&len| len < 4096),
+            "a delete must not read the needle's payload ({} bytes): {reads:?}",
+            data.len()
+        );
+        let (status, _, _) = send_read(&state, Method::GET, &path, None).await;
+        assert_eq!(status, StatusCode::NOT_FOUND);
+    }
+
     /// HEAD and ranged reads keep answering from the needle meta and the
     /// data file, and a missing or wrong-cookie needle is still a 404.
     #[tokio::test]
@@ -6105,6 +6170,29 @@ mod tests {
         assert_eq!(taken(&reads), vec![1]);
     }
 
+    /// A HEAD asking for an image transform needs the payload too: a
+    /// meta-only read could not produce the transformed length.
+    #[tokio::test]
+    async fn test_head_with_image_op_reads_full() {
+        let tmp = tempfile::TempDir::new().unwrap();
+        let state = volume_test_state(&tmp);
+        let png = solid_png(8, 8);
+        let stored = put_test_needle(&state, 0x6e7a_0a30, &png);
+        let crop = format!(
+            "{}.png?crop_x1=0&crop_y1=0&crop_x2=2&crop_y2=2",
+            stored
+        );
+
+        let (get_status, _, cropped) = send_read(&state, Method::GET, &crop, None).await;
+        assert_eq!(get_status, StatusCode::OK);
+        assert_ne!(cropped.len(), png.len(), "the crop must change the length");
+
+        let (status, headers, body) = send_read(&state, Method::HEAD, &crop, None).await;
+        assert_eq!(status, StatusCode::OK);
+        assert!(body.is_empty());
+        assert_eq!(headers[header::CONTENT_LENGTH], cropped.len().to_string());
+    }
+
     /// Go takes the transform extension from the URL first — including a fid
     /// suffix like /vid,fid.png — before the resolved filename's.
     #[tokio::test]

seaweed-volume/src/storage/store.rs

@@ -736,6 +736,20 @@ impl Store {
         vol.read_needle(n)
     }
 
+    /// Read a needle from a volume using the given read option.
+    ///
+    /// On return, `read_option.is_meta_only` tells the caller whether the
+    /// payload was skipped.
+    pub fn read_volume_needle_with_option(
+        &self,
+        vid: VolumeId,
+        n: &mut Needle,
+        read_option: &mut ReadOption,
+    ) -> Result<i32, VolumeError> {
+        let (_, vol) = self.find_volume(vid).ok_or(VolumeError::NotFound)?;
+        vol.read_needle_with_option(n, read_option)
+    }
+
     /// Read a needle from a volume, optionally reading deleted needles.
     pub fn read_volume_needle_opt(
         &self,

seaweed-volume/src/storage/volume.rs

@@ -2009,12 +2009,29 @@ impl Volume {
             return Ok(0);
         }
 
-        match self.read_needle_data_at_unlocked(
-            n,
-            nv.offset.to_actual_offset(),
-            read_size,
-            read_option,
-        ) {
+        // With `must_meta_only`, read the header and trailer first. If the
+        // needle is compressed or is a chunk manifest, the caller needs its
+        // data, so read the payload as well.
+        //
+        // A version 1 needle has no data size field and no metadata after its
+        // data, so the metadata reader would treat the whole payload as
+        // metadata. It is always read in full.
+        let offset = nv.offset.to_actual_offset();
+        // `is_meta_only` is a result: clear it in case the option is reused.
+        read_option.is_meta_only = false;
+        let meta_first = read_option.must_meta_only && self.version() != VERSION_1;
+        let mut read = if meta_first {
+            self.read_needle_meta_at_unlocked(n, offset, read_size)
+        } else {
+            Ok(())
+        };
+        if read.is_ok() && meta_first && !n.is_compressed() && !n.is_chunk_manifest() {
+            read_option.is_meta_only = true;
+        }
+        if read.is_ok() && !read_option.is_meta_only {
+            read = self.read_needle_data_at_unlocked(n, offset, read_size, read_option);
+        }
+        match read {
             Ok(()) => self.check_read_write_error(None),
             Err(VolumeError::Io(ref e)) => {
                 self.check_read_write_error(Some(e));
@@ -6080,6 +6097,120 @@ mod tests {
         assert_eq!(read_n.cookie, Cookie(0x12345678));
     }
 
+    /// With `must_meta_only` set, the payload is skipped whatever the needle's
+    /// size. This is the option a delete uses. A chunk manifest is the
+    /// exception: its data is still read, because a delete needs it to find
+    /// the chunks.
+    #[test]
+    fn test_read_needle_must_meta_only() {
+        let tmp = TempDir::new().unwrap();
+        let mut v = make_test_volume(tmp.path().to_str().unwrap());
+        let data = vec![7u8; 8192];
+        let mut plain = Needle {
+            id: NeedleId(1),
+            cookie: Cookie(0x1234),
+            data: data.clone(),
+            data_size: data.len() as u32,
+            ..Needle::default()
+        };
+        v.write_needle(&mut plain, true, false).unwrap();
+        let mut manifest = Needle {
+            id: NeedleId(2),
+            data: b"{}".to_vec(),
+            data_size: 2,
+            ..Needle::default()
+        };
+        manifest.set_is_chunk_manifest();
+        v.write_needle(&mut manifest, true, false).unwrap();
+
+        let read = |id: u64, read_option: &mut ReadOption| {
+            let mut n = Needle {
+                id: NeedleId(id),
+                ..Needle::default()
+            };
+            v.read_needle_with_option(&mut n, read_option).unwrap();
+            n
+        };
+        let meta_only = || ReadOption {
+            attempt_meta_only: true,
+            must_meta_only: true,
+            ..ReadOption::default()
+        };
+
+        let mut read_option = meta_only();
+        let n = read(1, &mut read_option);
+        assert!(read_option.is_meta_only);
+        assert!(n.data.is_empty(), "{} bytes of data read", n.data.len());
+        assert_eq!(n.cookie, Cookie(0x1234));
+        assert_eq!(n.data_size as usize, data.len());
+
+        // Without `must_meta_only`, the needle is read in full.
+        let mut read_option = ReadOption::default();
+        let n = read(1, &mut read_option);
+        assert!(!read_option.is_meta_only);
+        assert_eq!(n.data, data);
+
+        let mut read_option = meta_only();
+        let n = read(2, &mut read_option);
+        assert!(!read_option.is_meta_only);
+        assert_eq!(n.data, b"{}");
+
+        // The option is reused: a metadata-only result from the first read
+        // must not carry over and skip the chunk manifest's data.
+        let mut read_option = meta_only();
+        read(1, &mut read_option);
+        assert!(read_option.is_meta_only);
+        let n = read(2, &mut read_option);
+        assert!(!read_option.is_meta_only);
+        assert_eq!(n.data, b"{}");
+    }
+
+    /// A version 1 needle is a header, the data and a checksum: it has no data
+    /// size field and no metadata after the data, so the metadata reader would
+    /// take the whole payload for metadata and reject anything over 128 KiB.
+    /// A read with `must_meta_only` must fall back to a full read.
+    #[test]
+    fn test_read_needle_must_meta_only_version_1() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let spec = VolumeSpec {
+            version: VERSION_1,
+            ..VolumeSpec::default()
+        };
+        let mut v = Volume::new(dir, dir, VolumeId(1), NeedleMapKind::InMemory, &spec).unwrap();
+        assert_eq!(v.version(), VERSION_1);
+
+        for (id, len) in [(1u64, 11usize), (2, 256 * 1024)] {
+            let data: Vec<u8> = b"hello world".iter().copied().cycle().take(len).collect();
+            // The server only writes version 2 and 3 records, so build the
+            // version 1 record by hand.
+            let size = Size(len as i32);
+            let mut blob = vec![0u8; NEEDLE_HEADER_SIZE];
+            Cookie(0x1234).to_bytes(&mut blob[0..COOKIE_SIZE]);
+            NeedleId(id).to_bytes(&mut blob[COOKIE_SIZE..COOKIE_SIZE + NEEDLE_ID_SIZE]);
+            size.to_bytes(&mut blob[COOKIE_SIZE + NEEDLE_ID_SIZE..NEEDLE_HEADER_SIZE]);
+            blob.extend_from_slice(&data);
+            blob.extend_from_slice(&CRC::new(&data).0.to_be_bytes());
+            blob.resize(get_actual_size(size, VERSION_1) as usize, 0);
+            v.write_needle_blob_and_index(NeedleId(id), &blob, size)
+                .unwrap();
+
+            let mut n = Needle {
+                id: NeedleId(id),
+                ..Needle::default()
+            };
+            let mut read_option = ReadOption {
+                attempt_meta_only: true,
+                must_meta_only: true,
+                ..ReadOption::default()
+            };
+            v.read_needle_with_option(&mut n, &mut read_option).unwrap();
+            assert!(!read_option.is_meta_only);
+            assert_eq!(n.cookie, Cookie(0x1234));
+            assert_eq!(n.data, data, "{len} byte needle");
+        }
+    }
+
     /// A durable write goes down the same path and lands the same data; the
     /// flush is extra work, not different work.
     #[test]

weed/server/volume_grpc_batch_delete.go

@@ -60,7 +60,7 @@ func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.B
 		if !req.SkipCookieCheck {
 			cookie := n.Cookie
 			if !isEcVolume {
-				if _, err := vs.store.ReadVolumeNeedle(volumeId, n, nil, nil); err != nil {
+				if _, err := vs.store.ReadVolumeNeedle(volumeId, n, deleteReadOption(), nil); err != nil {
 					resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{
 						FileId: fid,
 						Status: http.StatusNotFound,

weed/server/volume_server_handlers_read.go

@@ -308,14 +308,19 @@ func shouldAttemptStreamWrite(hasLocalVolume bool, ext string, r *http.Request)
 	if len(ext) > 0 {
 		ext = strings.ToLower(ext)
 	}
-	if r.Method == http.MethodHead {
-		return true, true
-	}
 	_, _, _, shouldResize := shouldResizeImages(ext, r)
 	_, _, _, _, shouldCrop := shouldCropImages(ext, r)
 	if shouldResize || shouldCrop {
+		// A transformed image's length is not the needle's; the payload is
+		// needed to answer — also true for HEAD, which reports it.
 		return false, false
 	}
+	if r.Method == http.MethodHead {
+		// Attempt only, so a needle up to PagedReadLimit is still read whole:
+		// on a spinning disk one read is cheaper than a header read followed
+		// by a trailer read.
+		return true, false
+	}
 	return true, false
 }
 

weed/server/volume_server_handlers_read_test.go

@@ -0,0 +1,35 @@
+package weed_server
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+// A HEAD only attempts a metadata-only read, like a GET: a needle up to
+// PagedReadLimit is read whole. Demanding metadata only would turn that one
+// read into a header read followed by a trailer read.
+func TestShouldAttemptStreamWrite(t *testing.T) {
+	for _, tc := range []struct {
+		name         string
+		method       string
+		target       string
+		ext          string
+		hasVolume    bool
+		wantAttempt  bool
+		wantMustMeta bool
+	}{
+		{"HEAD", http.MethodHead, "/1,abc.jpg", ".jpg", true, true, false},
+		{"GET", http.MethodGet, "/1,abc.jpg", ".jpg", true, true, false},
+		{"GET of a resized image", http.MethodGet, "/1,abc.jpg?width=100", ".jpg", true, false, false},
+		{"HEAD of a resized image", http.MethodHead, "/1,abc.jpg?width=100", ".jpg", true, false, false},
+		{"HEAD of a cropped image", http.MethodHead, "/1,abc.jpg?crop_x1=0&crop_y1=0&crop_x2=100&crop_y2=100", ".jpg", true, false, false},
+		{"HEAD without a local volume", http.MethodHead, "/1,abc.jpg", ".jpg", false, false, false},
+	} {
+		r := httptest.NewRequest(tc.method, tc.target, nil)
+		attempt, mustMeta := shouldAttemptStreamWrite(tc.hasVolume, tc.ext, r)
+		if attempt != tc.wantAttempt || mustMeta != tc.wantMustMeta {
+			t.Errorf("%s: attempt %v, must be metadata only %v; want %v, %v", tc.name, attempt, mustMeta, tc.wantAttempt, tc.wantMustMeta)
+		}
+	}
+}

weed/server/volume_server_handlers_write.go

@@ -115,7 +115,7 @@ func (vs *VolumeServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	_, ok := vs.store.ReadVolumeNeedle(volumeId, n, nil, nil)
+	_, ok := vs.store.ReadVolumeNeedle(volumeId, n, deleteReadOption(), nil)
 	if ok != nil {
 		m := make(map[string]uint32)
 		m["size"] = 0
@@ -161,6 +161,17 @@ func (vs *VolumeServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
 
 }
 
+// deleteReadOption returns the read option used by deletes: read only the
+// needle's header and trailer.
+//
+// A delete needs to know that the needle exists, what its cookie is, and
+// whether it is a chunk manifest. None of that requires the payload. If the
+// needle turns out to be a chunk manifest, the read loads its data as well,
+// because the delete needs the manifest to delete the chunks.
+func deleteReadOption() *storage.ReadOption {
+	return &storage.ReadOption{AttemptMetaOnly: true, MustMetaOnly: true}
+}
+
 func writeDeleteResult(err error, count int64, w http.ResponseWriter, r *http.Request) {
 	if err == nil {
 		m := make(map[string]int64)

weed/storage/store.go

@@ -34,7 +34,12 @@ const (
 
 type ReadOption struct {
 	// request
-	ReadDeleted     bool
+	ReadDeleted bool
+	// AttemptMetaOnly alone skips the payload only for needles larger than
+	// PagedReadLimit. Setting MustMetaOnly as well skips it at any size.
+	// A compressed needle or a chunk manifest is always read in full, because
+	// the caller needs its data. After the read, IsMetaOnly reports whether
+	// the payload was skipped.
 	AttemptMetaOnly bool
 	MustMetaOnly    bool
 	// response

weed/storage/volume_read.go

@@ -51,7 +51,9 @@ func (v *Volume) readNeedle(n *needle.Needle, readOption *ReadOption, onReadSize
 	if onReadSizeFn != nil {
 		onReadSizeFn(readSize)
 	}
-	if readOption != nil && readOption.AttemptMetaOnly && readSize > PagedReadLimit {
+	// A version 1 needle has no data size field and no metadata after its
+	// data, which ReadNeedleMeta relies on, so it is always read in full.
+	if readOption != nil && readOption.AttemptMetaOnly && v.Version() != needle.Version1 && (readOption.MustMetaOnly || readSize > PagedReadLimit) {
 		readOption.VolumeRevision = v.SuperBlock.CompactionRevision
 		err = n.ReadNeedleMeta(v.DataBackend, nv.Offset.ToActualOffset(), readSize, v.Version())
 		if err == needle.ErrorSizeMismatch && OffsetSize == 4 {

weed/storage/volume_read_test.go

@@ -322,3 +322,125 @@ func TestReadNeedleDataIntoChecksumMismatchHoldsLastPage(t *testing.T) {
 		})
 	}
 }
+
+// With MustMetaOnly set, the payload is skipped whatever the needle's size.
+// This is the option a delete uses. A chunk manifest is the exception: its
+// data is still read, because a delete needs it to find the chunks.
+func TestReadNeedleMustMetaOnly(t *testing.T) {
+	dir := t.TempDir()
+
+	v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	if err != nil {
+		t.Fatalf("volume creation: %v", err)
+	}
+	defer v.Close()
+
+	plain := newRandomNeedle(1)
+	plain.Cookie = 0x1234
+	plain.Name = []byte("a.jpg")
+	plain.NameSize = uint8(len(plain.Name))
+	plain.SetHasName()
+	plainOffset, _, _, err := v.writeNeedle2(plain, true, false, false)
+	if err != nil {
+		t.Fatalf("write needle: %v", err)
+	}
+	manifest := newRandomNeedle(2)
+	manifest.SetIsChunkManifest()
+	if _, _, _, err := v.writeNeedle2(manifest, true, false, false); err != nil {
+		t.Fatalf("write chunk manifest: %v", err)
+	}
+
+	read := func(id uint64, option *ReadOption) (*needle.Needle, error) {
+		n := &needle.Needle{Id: types.Uint64ToNeedleId(id)}
+		_, err := v.readNeedle(n, option, nil)
+		return n, err
+	}
+	metaOnly := func() *ReadOption { return &ReadOption{AttemptMetaOnly: true, MustMetaOnly: true} }
+
+	option := metaOnly()
+	got, err := read(1, option)
+	if err != nil {
+		t.Fatalf("metadata read: %v", err)
+	}
+	if !option.IsMetaOnly || len(got.Data) != 0 {
+		t.Errorf("metadata read: IsMetaOnly %v, %d bytes of data", option.IsMetaOnly, len(got.Data))
+	}
+	if got.Cookie != plain.Cookie || got.Size != plain.Size || got.DataSize != plain.DataSize || string(got.Name) != "a.jpg" {
+		t.Errorf("metadata read: cookie %x size %d data size %d name %q", got.Cookie, got.Size, got.DataSize, got.Name)
+	}
+
+	// With AttemptMetaOnly alone, a small needle is still read in full.
+	option = &ReadOption{AttemptMetaOnly: true}
+	if got, err = read(1, option); err != nil || option.IsMetaOnly || !bytes.Equal(got.Data, plain.Data) {
+		t.Errorf("attempted metadata read of a small needle: IsMetaOnly %v, %d bytes, %v", option.IsMetaOnly, len(got.Data), err)
+	}
+
+	option = metaOnly()
+	if got, err = read(2, option); err != nil || option.IsMetaOnly || !bytes.Equal(got.Data, manifest.Data) {
+		t.Errorf("metadata read of a chunk manifest: IsMetaOnly %v, %d bytes, %v", option.IsMetaOnly, len(got.Data), err)
+	}
+
+	// A meta-only read never looks at the payload. Corrupt the payload and
+	// check that only a full read reports the damage.
+	damaged := []byte{^plain.Data[0]}
+	if _, err := v.DataBackend.WriteAt(damaged, int64(plainOffset)+types.NeedleHeaderSize+types.DataSizeSize); err != nil {
+		t.Fatalf("damage needle: %v", err)
+	}
+	if _, err := read(1, nil); !errors.Is(err, needle.ErrorCorrupted) {
+		t.Errorf("full read of a damaged needle: %v, want %v", err, needle.ErrorCorrupted)
+	}
+	if _, err := read(1, metaOnly()); err != nil {
+		t.Errorf("metadata read of a damaged needle: %v", err)
+	}
+}
+
+// A version 1 needle has no data size field and no metadata after its data,
+// so the metadata-only reader cannot parse it. A metadata-only read of a
+// version 1 volume must fall back to a full read instead of failing.
+func TestReadNeedleMetaOnlyVersion1(t *testing.T) {
+	dir := t.TempDir()
+
+	v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	if err != nil {
+		t.Fatalf("volume creation: %v", err)
+	}
+	defer v.Close()
+	// New volumes are always created with the current version, so switch this
+	// one to version 1 before anything is written to it.
+	v.volumeInfo.Version = uint32(needle.Version1)
+	if v.Version() != needle.Version1 {
+		t.Fatalf("volume version %d, want %d", v.Version(), needle.Version1)
+	}
+
+	// The payload starts with bytes that would be a huge data size if the
+	// metadata-only reader took them for the version 2 data size field.
+	small := &needle.Needle{Id: 1, Cookie: 0x1234, Data: []byte("hello world")}
+	small.Checksum = needle.NewCRC(small.Data)
+	if _, _, _, err := v.writeNeedle2(small, true, false, false); err != nil {
+		t.Fatalf("write small needle: %v", err)
+	}
+	large := &needle.Needle{Id: 2, Cookie: 0x5678, Data: bytes.Repeat([]byte("hello world "), PagedReadLimit/12+1)}
+	large.Checksum = needle.NewCRC(large.Data)
+	if _, _, _, err := v.writeNeedle2(large, true, false, false); err != nil {
+		t.Fatalf("write large needle: %v", err)
+	}
+
+	for _, tc := range []struct {
+		name   string
+		want   *needle.Needle
+		option *ReadOption
+	}{
+		{"must be metadata only, small needle", small, &ReadOption{AttemptMetaOnly: true, MustMetaOnly: true}},
+		{"must be metadata only, large needle", large, &ReadOption{AttemptMetaOnly: true, MustMetaOnly: true}},
+		{"attempt metadata only, large needle", large, &ReadOption{AttemptMetaOnly: true}},
+	} {
+		got := &needle.Needle{Id: tc.want.Id}
+		if _, err := v.readNeedle(got, tc.option, nil); err != nil {
+			t.Errorf("%s: %v", tc.name, err)
+			continue
+		}
+		if tc.option.IsMetaOnly || got.Cookie != tc.want.Cookie || !bytes.Equal(got.Data, tc.want.Data) {
+			t.Errorf("%s: IsMetaOnly %v, cookie %x, %d bytes of data", tc.name, tc.option.IsMetaOnly, got.Cookie, len(got.Data))
+		}
+	}
+}