diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 3c0c3218fe..d2d02d35b0 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -150,6 +150,8 @@ struct SourceReadRequest { is_head: bool, has_range: bool, has_image_ops: bool, + /// No URL extension, but the stored name may still make it a transform. + may_have_image_ops: bool, bypass_cm: bool, } @@ -175,7 +177,10 @@ impl SourceReadRequest { if self.has_image_ops { return false; } - self.is_head || self.has_range || size.0 > STREAMING_THRESHOLD as i32 + // A range that may need a transform reads a small needle whole, like Go. + self.is_head + || (self.has_range && !self.may_have_image_ops) + || size.0 > STREAMING_THRESHOLD as i32 } } @@ -1461,6 +1466,7 @@ fn parse_read_request( is_head: method == Method::HEAD, has_range, has_image_ops: has_image_ops(&ext, query), + may_have_image_ops: ext.is_empty() && has_image_ops(".jpg", query), bypass_cm: query.cm.as_deref() == Some("false"), }; (ext, request_kind) @@ -6460,6 +6466,41 @@ mod tests { assert_eq!(status, StatusCode::NOT_FOUND); } + /// A range streams a small needle from the data file, unless it has no + /// URL extension and asks for a resize or crop the stored name may call + /// for: then the needle is read whole, like Go. + #[test] + fn test_range_reads_meta_first_unless_it_may_need_a_transform() { + let small = Size(4096); + let width = ReadQueryParams { + width: Some(100), + ..Default::default() + }; + let crop = ReadQueryParams { + crop_x2: Some(10), + crop_y2: Some(10), + ..Default::default() + }; + for (path, query, want) in [ + ("/1,abc", ReadQueryParams::default(), true), + ("/1,abc", width, false), + ("/1,abc", crop, false), + ( + "/1,abc.bin", + ReadQueryParams { + width: Some(100), + ..Default::default() + }, + true, + ), + ] { + let (_, request) = parse_read_request(path, true, &query, &Method::GET); + assert_eq!(request.try_meta_first(small), want, "{path}"); + let (_, request) = parse_read_request(path, false, &query, &Method::GET); + assert!(!request.try_meta_first(small), "{path} without a range"); + } + } + /// 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] diff --git a/weed/server/volume_server_handlers_read.go b/weed/server/volume_server_handlers_read.go index 99c51cb851..30d6968b52 100644 --- a/weed/server/volume_server_handlers_read.go +++ b/weed/server/volume_server_handlers_read.go @@ -251,7 +251,7 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) var count int var memoryCost types.Size - readOption.AttemptMetaOnly, readOption.MustMetaOnly = shouldAttemptStreamWrite(hasVolume, ext, r) + readOption.AttemptMetaOnly, readOption.MustMetaOnly, readOption.HasRange = shouldAttemptStreamWrite(hasVolume, ext, r) onReadSizeFn := func(size types.Size) { memoryCost = size atomic.AddInt64(&vs.inFlightDownloadDataSize, int64(memoryCost)) @@ -371,9 +371,11 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) } } -func shouldAttemptStreamWrite(hasLocalVolume bool, ext string, r *http.Request) (shouldAttempt bool, mustMetaOnly bool) { +// shouldAttemptStreamWrite picks the ReadOption flags that let a GET or HEAD +// skip the payload read: AttemptMetaOnly, MustMetaOnly and HasRange. +func shouldAttemptStreamWrite(hasLocalVolume bool, ext string, r *http.Request) (shouldAttempt bool, mustMetaOnly bool, hasRange bool) { if !hasLocalVolume { - return false, false + return false, false, false } if len(ext) > 0 { ext = strings.ToLower(ext) @@ -383,15 +385,25 @@ func shouldAttemptStreamWrite(hasLocalVolume bool, ext string, r *http.Request) 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 + return false, 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, false } - return true, false + if ext == "" { + // The stored name may still make this an image to resize or crop, + // which needs the payload. + _, _, _, resizeIfImage := shouldResizeImages(".jpg", r) + _, _, _, _, cropIfImage := shouldCropImages(".jpg", r) + if resizeIfImage || cropIfImage { + return true, false, false + } + } + // A range streams only its bytes, so a small needle is not read whole. + return true, false, r.Header.Get("Range") != "" } func (vs *VolumeServer) tryHandleChunkedFile(n *needle.Needle, fileName string, ext string, w http.ResponseWriter, r *http.Request) (processed bool) { diff --git a/weed/server/volume_server_handlers_read_test.go b/weed/server/volume_server_handlers_read_test.go index 769f870eb2..bb75ba3e29 100644 --- a/weed/server/volume_server_handlers_read_test.go +++ b/weed/server/volume_server_handlers_read_test.go @@ -49,28 +49,42 @@ func TestInternalErrorWritesTheReadError(t *testing.T) { // 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. +// read into a header read followed by a trailer read. A ranged GET skips the +// payload at any size, since it streams only the range. func TestShouldAttemptStreamWrite(t *testing.T) { for _, tc := range []struct { name string method string target string ext string + rangeHeader string hasVolume bool wantAttempt bool wantMustMeta bool + wantRange 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}, + {"HEAD", http.MethodHead, "/1,abc.jpg", ".jpg", "", true, true, false, false}, + {"GET", http.MethodGet, "/1,abc.jpg", ".jpg", "", true, true, false, false}, + {"ranged GET", http.MethodGet, "/1,abc", "", "bytes=0-99", true, true, false, true}, + {"ranged HEAD", http.MethodHead, "/1,abc", "", "bytes=0-99", true, true, false, false}, + {"ranged GET of a resized image", http.MethodGet, "/1,abc.jpg?width=100", ".jpg", "bytes=0-99", true, false, false, false}, + {"ranged GET to resize, no extension", http.MethodGet, "/1,abc?width=100", "", "bytes=0-99", true, true, false, false}, + {"ranged GET to crop, no extension", http.MethodGet, "/1,abc?crop_x1=0&crop_y1=0&crop_x2=10&crop_y2=10", "", "bytes=0-99", true, true, false, false}, + {"ranged GET with a width on a non-image", http.MethodGet, "/1,abc.bin?width=100", ".bin", "bytes=0-99", true, true, false, true}, + {"ranged GET without a local volume", http.MethodGet, "/1,abc", "", "bytes=0-99", false, false, false, false}, + {"GET of a resized image", http.MethodGet, "/1,abc.jpg?width=100", ".jpg", "", true, false, false, false}, + {"HEAD of a resized image", http.MethodHead, "/1,abc.jpg?width=100", ".jpg", "", true, false, 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, false}, + {"HEAD without a local volume", http.MethodHead, "/1,abc.jpg", ".jpg", "", false, 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) + if tc.rangeHeader != "" { + r.Header.Set("Range", tc.rangeHeader) + } + attempt, mustMeta, hasRange := shouldAttemptStreamWrite(tc.hasVolume, tc.ext, r) + if attempt != tc.wantAttempt || mustMeta != tc.wantMustMeta || hasRange != tc.wantRange { + t.Errorf("%s: attempt %v, must be metadata only %v, range %v; want %v, %v, %v", + tc.name, attempt, mustMeta, hasRange, tc.wantAttempt, tc.wantMustMeta, tc.wantRange) } } } diff --git a/weed/storage/store.go b/weed/storage/store.go index 4f6a7cc72c..21ac971ed7 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -37,16 +37,20 @@ type ReadOption struct { // request ReadDeleted bool // AttemptMetaOnly alone skips the payload only for needles larger than - // PagedReadLimit. Setting MustMetaOnly as well skips it at any size. + // PagedReadLimit. Setting MustMetaOnly or HasRange as well skips it at + // any size: MustMetaOnly when the request ends at the metadata, HasRange + // when a range of the payload is read next and reports the outcome. // 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 + HasRange bool // response IsMetaOnly bool // read status VolumeRevision uint16 - IsOutOfRange bool // whether read over MaxPossibleVolumeSize + metaOffset Offset // the record a metadata-only read resolved + IsOutOfRange bool // whether read over MaxPossibleVolumeSize // If HasSlowRead is set to true: // * read requests and write requests compete for the lock. diff --git a/weed/storage/volume_read.go b/weed/storage/volume_read.go index 3112cabb15..2a00cbe1c5 100644 --- a/weed/storage/volume_read.go +++ b/weed/storage/volume_read.go @@ -11,6 +11,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/storage/backend" "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/needle_map" "github.com/seaweedfs/seaweedfs/weed/storage/super_block" . "github.com/seaweedfs/seaweedfs/weed/storage/types" ) @@ -53,7 +54,7 @@ func (v *Volume) readNeedle(n *needle.Needle, readOption *ReadOption, onReadSize } // 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) { + if readOption != nil && readOption.AttemptMetaOnly && v.Version() != needle.Version1 && (readOption.MustMetaOnly || readOption.HasRange || readSize > PagedReadLimit) { readOption.VolumeRevision = v.SuperBlock.CompactionRevision err = n.ReadNeedleMeta(v.DataBackend, nv.Offset.ToActualOffset(), readSize, v.Version()) if err == needle.ErrorSizeMismatch && OffsetSize == 4 { @@ -66,6 +67,7 @@ func (v *Volume) readNeedle(n *needle.Needle, readOption *ReadOption, onReadSize } if !n.IsCompressed() && !n.IsChunkedManifest() { readOption.IsMetaOnly = true + readOption.metaOffset = nv.Offset // A must-meta-only read ends here, so its success may reset the // streak. For a streamed GET the meta read precedes the payload read, // which reports the outcome itself. @@ -128,6 +130,31 @@ func (v *Volume) readNeedleMetaAt(n *needle.Needle, offset int64, size int32) (e return nil } +// verifyMovedRecord checks that the record a compaction moved is the one n's +// metadata came from, not a replacement of the same size, and sets +// IsOutOfRange for where it now lies. +func (v *Volume) verifyMovedRecord(n *needle.Needle, moved *needle_map.NeedleValue, found bool, readOption *ReadOption) error { + if !found || moved.Offset.IsZero() || moved.Size != n.Size { + return ErrorNotFound + } + m := new(needle.Needle) + readOption.IsOutOfRange = false + err := m.ReadNeedleMeta(v.DataBackend, moved.Offset.ToActualOffset(), moved.Size, v.Version()) + if err == needle.ErrorSizeMismatch && OffsetSize == 4 { + readOption.IsOutOfRange = true + err = m.ReadNeedleMeta(v.DataBackend, moved.Offset.ToActualOffset()+int64(MaxPossibleVolumeSize), moved.Size, v.Version()) + } + if err != nil { + v.checkReadWriteError(err) + return fmt.Errorf("ReadNeedleMeta of moved needle: %w", err) + } + // Compaction rewrites a pre-3.09 checksum in the current encoding. + if m.AppendAtNs != n.AppendAtNs || (m.Checksum != n.Checksum && uint32(n.Checksum) != m.Checksum.Value()) { + return ErrorNotFound + } + return nil +} + // read fills in Needle content by looking up n.Id from NeedleMapper func (v *Volume) readNeedleDataInto(n *needle.Needle, readOption *ReadOption, writer io.Writer, offset int64, size int64) (err error) { @@ -172,6 +199,11 @@ func (v *Volume) readNeedleDataInto(n *needle.Needle, readOption *ReadOption, wr if readSize == 0 { return nil } + // Stream the record the metadata came from: the index may already point + // at a replacement, whose bytes must not go out under the old headers. + if !readOption.metaOffset.IsZero() { + nv = &needle_map.NeedleValue{Key: n.Id, Offset: readOption.metaOffset, Size: n.Size} + } actualOffset := nv.Offset.ToActualOffset() if readOption.IsOutOfRange { @@ -201,12 +233,16 @@ func (v *Volume) readNeedleDataInto(n *needle.Needle, readOption *ReadOption, wr return ErrorNotFound } refreshed, found := v.nm.Get(n.Id) - if !found || refreshed.Offset.IsZero() { - return ErrorNotFound + if err := v.verifyMovedRecord(n, refreshed, found, readOption); err != nil { + return err } nv = refreshed actualOffset = nv.Offset.ToActualOffset() + if readOption.IsOutOfRange { + actualOffset += int64(MaxPossibleVolumeSize) + } readOption.VolumeRevision = v.SuperBlock.CompactionRevision + readOption.metaOffset = nv.Offset } header := make([]byte, NeedleHeaderSize) headerCount, headerErr := v.DataBackend.ReadAt(header, actualOffset) @@ -263,14 +299,18 @@ func (v *Volume) readNeedleDataInto(n *needle.Needle, readOption *ReadOption, wr } // the volume is compacted nv, ok = v.nm.Get(n.Id) - if !ok || nv.Offset.IsZero() { + if err := v.verifyMovedRecord(n, nv, ok, readOption); err != nil { if readOption.HasSlowRead { v.dataFileAccessLock.RUnlock() } - return ErrorNotFound + return err } actualOffset = nv.Offset.ToActualOffset() + if readOption.IsOutOfRange { + actualOffset += int64(MaxPossibleVolumeSize) + } readOption.VolumeRevision = v.SuperBlock.CompactionRevision + readOption.metaOffset = nv.Offset } count, err := n.ReadNeedleData(v.DataBackend, actualOffset, buf, x) if readOption.HasSlowRead { diff --git a/weed/storage/volume_read_test.go b/weed/storage/volume_read_test.go index 491245bc5d..3955c192f1 100644 --- a/weed/storage/volume_read_test.go +++ b/weed/storage/volume_read_test.go @@ -8,6 +8,7 @@ import ( "path/filepath" "reflect" "strings" + "syscall" "testing" "time" @@ -394,6 +395,201 @@ func TestReadNeedleMustMetaOnly(t *testing.T) { } } +// HasRange skips the payload at any size, like MustMetaOnly, but the range +// read that follows reports the outcome, so the metadata read leaves the io +// error streak alone. +func TestReadNeedleHasRange(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 := &needle.Needle{Id: 1, Cookie: 0x1234, Data: bytes.Repeat([]byte("0123456789"), 400)} + plain.Checksum = needle.NewCRC(plain.Data) + if _, _, _, err := v.writeNeedle2(plain, true, false, false); 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) + } + compressed := newRandomNeedle(3) + compressed.SetIsCompressed() + if _, _, _, err := v.writeNeedle2(compressed, true, false, false); err != nil { + t.Fatalf("write compressed needle: %v", err) + } + + v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) + option := &ReadOption{AttemptMetaOnly: true, HasRange: true, ReadBufferSize: 1024} + got := &needle.Needle{Id: plain.Id} + if _, err := v.readNeedle(got, option, nil); err != nil { + t.Fatalf("metadata read: %v", err) + } + if !option.IsMetaOnly || len(got.Data) != 0 || got.Cookie != plain.Cookie || got.DataSize != plain.DataSize { + t.Errorf("metadata read: IsMetaOnly %v, %d bytes of data, cookie %x, data size %d", option.IsMetaOnly, len(got.Data), got.Cookie, got.DataSize) + } + if _, count, _ := v.getIoErrorState(); count != 1 { + t.Errorf("io error streak %d after the metadata read, want 1", count) + } + + var buf bytes.Buffer + if err := v.readNeedleDataInto(got, option, &buf, 1500, 2000); err != nil { + t.Fatalf("range read: %v", err) + } + if !bytes.Equal(buf.Bytes(), plain.Data[1500:3500]) { + t.Errorf("range read returned %d bytes that differ from the needle", buf.Len()) + } + if _, count, _ := v.getIoErrorState(); count != 0 { + t.Errorf("io error streak %d after the range read, want 0", count) + } + + // A chunk manifest or a compressed needle is still read whole. + for _, want := range []*needle.Needle{manifest, compressed} { + option = &ReadOption{AttemptMetaOnly: true, HasRange: true} + got = &needle.Needle{Id: want.Id} + if _, err := v.readNeedle(got, option, nil); err != nil || option.IsMetaOnly || !bytes.Equal(got.Data, want.Data) { + t.Errorf("ranged read of needle %d: IsMetaOnly %v, %d bytes, %v", want.Id, option.IsMetaOnly, len(got.Data), err) + } + } +} + +// A range streams from the record its metadata read resolved, not from a +// replacement written in between, whatever the replacement's size. +func TestReadNeedleDataIntoKeepsTheMetaRecord(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() + + write := func(data []byte) { + t.Helper() + n := &needle.Needle{Id: 1, Cookie: 0x1234, Data: data} + n.Checksum = needle.NewCRC(n.Data) + if _, _, _, err := v.writeNeedle2(n, true, false, false); err != nil { + t.Fatalf("write needle: %v", err) + } + } + old := bytes.Repeat([]byte("old-"), 1000) + for _, replacement := range [][]byte{bytes.Repeat([]byte("new-"), 1000), bytes.Repeat([]byte("short"), 10)} { + write(old) + option := &ReadOption{AttemptMetaOnly: true, HasRange: true, ReadBufferSize: 1024} + got := &needle.Needle{Id: 1} + if _, err := v.readNeedle(got, option, nil); err != nil || !option.IsMetaOnly { + t.Fatalf("metadata read: IsMetaOnly %v, %v", option.IsMetaOnly, err) + } + write(replacement) + var buf bytes.Buffer + if err := v.readNeedleDataInto(got, option, &buf, 100, 2000); err != nil { + t.Fatalf("range read after a %d-byte replacement: %v", len(replacement), err) + } + if !bytes.Equal(buf.Bytes(), old[100:2100]) { + t.Errorf("range read after a %d-byte replacement returned bytes of another record", len(replacement)) + } + } +} + +// A compaction moves the record between the parts of a multi-range reply; +// every later part follows it, not only the first. A pre-3.09 checksum, +// which compaction rewrites in the current encoding, still matches. +func TestReadNeedleDataIntoFollowsACompaction(t *testing.T) { + for _, legacy := range []bool{false, true} { + t.Run(fmt.Sprintf("legacy checksum %v", legacy), func(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() + + gone := &needle.Needle{Id: 1, Cookie: 0x1234, Data: bytes.Repeat([]byte("gone"), 1000)} + gone.Checksum = needle.NewCRC(gone.Data) + data := bytes.Repeat([]byte("0123456789"), 400) + kept := &needle.Needle{Id: 2, Cookie: 0x1234, Data: data} + kept.Checksum = needle.NewCRC(kept.Data) + if legacy { + kept.Checksum = needle.CRC(kept.Checksum.Value()) + } + for _, n := range []*needle.Needle{gone, kept} { + if _, _, _, err := v.writeNeedle2(n, true, false, false); err != nil { + t.Fatalf("write needle %d: %v", n.Id, err) + } + } + if _, err := v.deleteNeedle2(newEmptyNeedle(1)); err != nil { + t.Fatalf("delete: %v", err) + } + + option := &ReadOption{AttemptMetaOnly: true, HasRange: true, ReadBufferSize: 1024} + got := &needle.Needle{Id: kept.Id} + if _, err := v.readNeedle(got, option, nil); err != nil || !option.IsMetaOnly { + t.Fatalf("metadata read: IsMetaOnly %v, %v", option.IsMetaOnly, err) + } + if err := v.CompactByIndex(nil); err != nil { + t.Fatalf("compact: %v", err) + } + if err := v.CommitCompact(); err != nil { + t.Fatalf("commit compact: %v", err) + } + + for _, part := range [][2]int64{{0, 100}, {2000, 500}} { + var buf bytes.Buffer + if err := v.readNeedleDataInto(got, option, &buf, part[0], part[1]); err != nil { + t.Fatalf("part at %d: %v", part[0], err) + } + if !bytes.Equal(buf.Bytes(), data[part[0]:part[0]+part[1]]) { + t.Errorf("part at %d returned bytes of another record", part[0]) + } + } + }) + } +} + +// A compaction that drops the record the metadata came from leaves only its +// replacement; a range then fails rather than send the replacement's bytes, +// even when the sizes match. +func TestReadNeedleDataIntoRefusesAReplacementAfterCompaction(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() + + write := func(data []byte) { + t.Helper() + n := &needle.Needle{Id: 1, Cookie: 0x1234, Data: data} + n.Checksum = needle.NewCRC(n.Data) + if _, _, _, err := v.writeNeedle2(n, true, false, false); err != nil { + t.Fatalf("write needle: %v", err) + } + } + write(bytes.Repeat([]byte("old-"), 1000)) + option := &ReadOption{AttemptMetaOnly: true, HasRange: true, ReadBufferSize: 1024} + got := &needle.Needle{Id: 1} + if _, err := v.readNeedle(got, option, nil); err != nil || !option.IsMetaOnly { + t.Fatalf("metadata read: IsMetaOnly %v, %v", option.IsMetaOnly, err) + } + write(bytes.Repeat([]byte("new-"), 1000)) + if err := v.CompactByIndex(nil); err != nil { + t.Fatalf("compact: %v", err) + } + if err := v.CommitCompact(); err != nil { + t.Fatalf("commit compact: %v", err) + } + + var buf bytes.Buffer + if err := v.readNeedleDataInto(got, option, &buf, 100, 2000); err != ErrorNotFound || buf.Len() != 0 { + t.Errorf("range read after the record was replaced and compacted: %d bytes, %v; want %v", buf.Len(), err, ErrorNotFound) + } +} + // 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.