mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-02 02:10:45 +01:00
volume: serve a ranged GET without reading the whole needle (#12301)
* volume: serve a ranged GET without reading the whole needle A ranged GET read every needle up to 1MB in full, checksum and all, then copied out the requested bytes. Read the metadata and stream only the range instead, as the Go server already does above 1MB and the Rust server does at any size. A small range of a 790KB needle goes from ~340us to ~160us. The metadata read stays non-terminal: the range read that follows reports the io outcome, so a failing disk still builds its error streak. HEAD, compressed needles and chunk manifests keep their current reads. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * volume: read the whole needle for a range that may need an image transform Without a URL extension, the stored name can still make a ranged GET with width, height or crop parameters an image transform, which needs the payload. Keep the full read there instead of streaming the range. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * volume: stream a range from the record its metadata came from The streamed read looked the needle up again, so a replacement written after the metadata read sent new bytes under the old length, checksum and headers. Read the record the metadata read resolved, as the Rust server does, and after a compaction refuse a record whose size changed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * rust volume: read the whole needle for a range that may need an image transform A ranged GET with no URL extension but width, height or crop parameters took the metadata-first path, so a stored image name never got its transform. Read a small needle whole there, as the Go server does. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * volume: keep later parts of a multi-range reply on a moved record When a compaction moved the record before one part of a multi-range reply, that part re-resolved it, but the next part went back to the offset the metadata read saw. Remember the re-resolved offset too. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * volume: check a compaction-moved record is the one the metadata came from After a compaction, a range re-resolved the needle and accepted any record of the same size, so a replacement written in between could send its bytes under the old headers. Compare the moved record's checksum and append time with the metadata already read, and fail otherwise. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * volume: accept a re-encoded checksum and count read errors on a moved record Compaction rewrites a pre-3.09 checksum in the current encoding, so the moved-record check now accepts either. A failed read of the moved record's metadata now feeds the io-error streak and surfaces as that error instead of not-found. The pinned record is a fresh value rather than a write through the index's pointer. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * volume: keep a moved record's wrapped offset when a range follows it With 4-byte offsets, a record past the 32GB wrap is read at offset plus MaxPossibleVolumeSize. The compaction refresh dropped that, so it checked and streamed the wrong location. Retry the wrapped offset in the moved record check, as the metadata read does, and apply it to the stream. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
4be191173a
commit
a1a208506e
6 files changed
+331
-24
No files matched your search
@@ -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]
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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.
|
||||
|
||||
Reference in new issue
Block a user