diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index f363648c3..ee3db150b 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/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"; diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 8441c9430..2e56f3ed1 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/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>, request: Request, @@ -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, 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 = (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] diff --git a/seaweed-volume/src/storage/store.rs b/seaweed-volume/src/storage/store.rs index bd0b1fb3f..fe8891491 100644 --- a/seaweed-volume/src/storage/store.rs +++ b/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 { + 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, diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 4477a8ad7..e91721ec6 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/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 = 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] diff --git a/weed/server/volume_grpc_batch_delete.go b/weed/server/volume_grpc_batch_delete.go index e925df895..06a2eeb3e 100644 --- a/weed/server/volume_grpc_batch_delete.go +++ b/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, diff --git a/weed/server/volume_server_handlers_read.go b/weed/server/volume_server_handlers_read.go index 581fb5f9b..957f9a17e 100644 --- a/weed/server/volume_server_handlers_read.go +++ b/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 } diff --git a/weed/server/volume_server_handlers_read_test.go b/weed/server/volume_server_handlers_read_test.go new file mode 100644 index 000000000..83f7e3bc5 --- /dev/null +++ b/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) + } + } +} diff --git a/weed/server/volume_server_handlers_write.go b/weed/server/volume_server_handlers_write.go index b380f03ed..149409fc0 100644 --- a/weed/server/volume_server_handlers_write.go +++ b/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) diff --git a/weed/storage/store.go b/weed/storage/store.go index d4447cbc3..a7a54091c 100644 --- a/weed/storage/store.go +++ b/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 diff --git a/weed/storage/volume_read.go b/weed/storage/volume_read.go index 672d774f6..f93c1a570 100644 --- a/weed/storage/volume_read.go +++ b/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 { diff --git a/weed/storage/volume_read_test.go b/weed/storage/volume_read_test.go index be3971e74..491245bc5 100644 --- a/weed/storage/volume_read_test.go +++ b/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)) + } + } +}