diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 4eb390f314..6accfed9fe 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -207,9 +207,15 @@ fn read_needle_for_get( if request.try_meta_first(plan.size()) { plan.read_meta(&mut n)?; if n.cookie != cookie { + // The meta read answered the request. + plan.note_meta_only_success(); return Ok(None); } if request.served_from_source(&n) { + if request.is_head { + // HEAD ends here; no payload read will record the success. + plan.note_meta_only_success(); + } let info = plan.into_stream_info(&n); return Ok(Some((n, Some(info)))); } @@ -237,6 +243,9 @@ struct StreamingBody { hold_read_lock_for_stream: bool, _held_read_lease: Option, io_unavailable: Arc, + /// Chunk read failures feed the volume's io-error streak, and finishing + /// the stream clears it — Go readNeedleDataInto parity. + io_errors: Arc, /// Pending chunk read; it hands `buf` back with the result. pending: Option)>>, /// Chunk buffer, reclaimed once the previous frame has been written out. @@ -298,12 +307,18 @@ impl http_body::Body for StreamingBody { ), ))); } + // Hyper may not poll again once Content-Length is + // satisfied, so the read's success is recorded here. + self.io_errors.check_read_write_error(None); } return std::task::Poll::Ready(Some(Ok(http_body::Frame::data( chunk, )))); } - Ok(Err(e)) => return std::task::Poll::Ready(Some(Err(e))), + Ok(Err(e)) => { + self.io_errors.check_read_write_error(Some(&e)); + return std::task::Poll::Ready(Some(Err(e))); + } Err(e) => { return std::task::Poll::Ready(Some(Err(std::io::Error::other(e)))); } @@ -314,6 +329,7 @@ impl http_body::Body for StreamingBody { let total = self.data_size as usize; if self.pos >= total { + self.io_errors.check_read_write_error(None); return std::task::Poll::Ready(None); } @@ -1105,6 +1121,10 @@ async fn get_or_head_handler_inner( let (etag, last_modified_str) = etag_and_last_modified(&n); if let Some(resp) = not_modified_response(&n, &headers, &etag, &last_modified_str) { + // A request that ends here was answered by the meta read alone. + if let Some(io_errors) = strategy.io_errors() { + io_errors.check_read_write_error(None); + } return resp; } @@ -1198,6 +1218,19 @@ struct ReadPlan { strategy: ReadStrategy, } +impl ReadStrategy { + /// The tracker a terminal meta read reports success to, when the reply + /// is served from the data file. + fn io_errors(&self) -> Option<&Arc> { + match self { + ReadStrategy::Stream(info) + | ReadStrategy::HeadFromMeta(info) + | ReadStrategy::RangeFromSource(info) => Some(&info.io_errors), + ReadStrategy::Buffered => None, + } + } +} + /// The 401 reply for a bad read JWT; Go checks it before NewVolumeId, so a /// bad path with JWT enabled is a 401, not a 400. fn reject_read_jwt( @@ -1722,6 +1755,7 @@ fn stream_response( data_file_access_control: info.data_file_access_control, hold_read_lock_for_stream: !state.has_slow_read, io_unavailable: info.io_unavailable, + io_errors: info.io_errors, pending: None, buf: bytes::BytesMut::new(), state: tracking_state, @@ -2121,9 +2155,21 @@ fn handle_range_request_from_source( return Ok(Vec::new()); } let mut buf = vec![0u8; length as usize]; - info.source - .read_exact_at(&mut buf, info.data_file_offset + start as u64)?; - Ok(buf) + match info + .source + .read_exact_at(&mut buf, info.data_file_offset + start as u64) + { + // A finished range read clears the streak a failed meta read + // counted, like readNeedleDataInto's terminal report. + Ok(()) => { + info.io_errors.check_read_write_error(None); + Ok(buf) + } + Err(e) => { + info.io_errors.check_read_write_error(Some(&e)); + Err(e) + } + } }; if ranges.len() == 1 { @@ -5225,6 +5271,7 @@ mod tests { hold_read_lock_for_stream: true, _held_read_lease: None, io_unavailable: Arc::default(), + io_errors: Arc::default(), pending: None, buf: bytes::BytesMut::new(), state: None, @@ -5771,6 +5818,44 @@ mod tests { assert_eq!(status, StatusCode::NOT_FOUND); } + /// A ranged read served from the data file reports its outcome to the + /// io-error streak, like readNeedleDataInto: a finished range clears the + /// count a failed meta read left. + #[test] + fn test_range_from_source_reports_to_io_error_streak() { + use std::io::Write; + + let tmp = tempfile::TempDir::new().unwrap(); + let file = std::fs::OpenOptions::new() + .read(true) + .write(true) + .create(true) + .open(tmp.path().join("1.dat")) + .unwrap(); + let data: Vec = (0..256u32).map(|i| i as u8).collect(); + (&file).write_all(&data).unwrap(); + + let io_errors = Arc::new(crate::storage::io_error::IoErrorTracker::default()); + let eio = crate::storage::volume::media_error_for_test(); + io_errors.check_read_write_error(Some(&eio)); + io_errors.check_read_write_error(Some(&eio)); + assert_eq!(io_errors.get_io_error_state().1, 2); + + let info = crate::storage::volume::NeedleStreamInfo { + source: crate::storage::volume::NeedleStreamSource::Local(file), + data_file_offset: 0, + data_size: data.len() as u32, + data_file_access_control: Arc::default(), + io_unavailable: Arc::default(), + io_errors: io_errors.clone(), + needle_id: NeedleId(1), + checksum: 0, + }; + let resp = handle_range_request_from_source("bytes=10-19", info, HeaderMap::new(), None); + assert_eq!(resp.status(), StatusCode::PARTIAL_CONTENT); + assert_eq!(io_errors.get_io_error_state().1, 0); + } + /// Range is read as Go reads it: a byte its parseRange cannot parse is a /// 416, Unicode whitespace around a range is trimmed, an empty value is no /// range, and HEAD ignores it. diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 83bc73093d..7fc7be57de 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -836,7 +836,7 @@ impl NeedleReadPlan { pub(crate) fn read_meta(&mut self, n: &mut Needle) -> Result<(), VolumeError> { let (source, size, version) = (&self.source, self.size, self.version); let read_at = |buf: &mut [u8], at: u64| Ok(source.read_exact_at(buf, at)?); - with_offset_retry(&mut self.offset, |off| { + let result = with_offset_retry(&mut self.offset, |off| { if version == VERSION_1 { // A V1 body is all data, so its "meta" is the whole record. let mut buf = vec![0u8; get_actual_size(size, version) as usize]; @@ -847,13 +847,26 @@ impl NeedleReadPlan { } else { Volume::read_needle_meta_blob_and_parse(read_at, n, off, size, version) } - })?; + }); + // Report failures only: a successful meta read precedes the payload + // read, so clearing the streak here would reset the count before + // every failing download and the volume could never reach quarantine. + if let Err(VolumeError::Io(e)) = &result { + self.io_errors.check_read_write_error(Some(e)); + } + result?; if needle_expired(n) { return Err(VolumeError::NotFound); } Ok(()) } + /// A successful meta read ended the request (e.g. HEAD), so it may reset + /// the streak — unlike a meta read that precedes a payload read. + pub(crate) fn note_meta_only_success(&self) { + self.io_errors.check_read_write_error(None); + } + /// For a reply served from the data file, after `read_meta` filled `n`. pub(crate) fn into_stream_info(self, n: &Needle) -> NeedleStreamInfo { // V1 data starts right after the header, V2/V3 after the DataSize field. @@ -868,6 +881,7 @@ impl NeedleReadPlan { data_size: n.data_size, data_file_access_control: self.data_file_access_control, io_unavailable: self.io_unavailable, + io_errors: self.io_errors, needle_id: self.needle_id, checksum: n.checksum.0, } @@ -1144,6 +1158,9 @@ pub struct NeedleStreamInfo { pub data_file_access_control: Arc, /// Checked before each chunk: the volume can become unavailable mid-stream. pub(crate) io_unavailable: Arc, + /// Streamed chunks report read failures here and clear the streak when + /// the whole body has been emitted, mirroring readNeedleDataInto. + pub(crate) io_errors: Arc, pub needle_id: NeedleId, /// Checksum stored in the needle tail, verified once the last chunk has /// been read — before that frame is emitted. @@ -2918,18 +2935,52 @@ impl Volume { } // Cookie validation for existing needle (matches Go: check whenever nm.Get returns ok) - if let Some(nm) = &io.nm - && let Some(nv) = nm.get(n.id)? + let indexed = match io.nm.as_ref() { + Some(nm) => nm.get(n.id)?, + None => None, + }; + if let Some(nv) = indexed + && !nv.offset.is_zero() { let mut existing = Needle::default(); // Read only the header to check cookie self.read_needle_header_unlocked(io, &mut existing, nv.offset.to_actual_offset())?; - if n.cookie.0 == 0 && !check_cookie { - n.cookie = existing.cookie; + if existing.id != n.id && existing.size != nv.size { + // a 4-byte index entry can index past the 32 GiB wrap: the + // low offset may hold another needle while the real record + // sits at the wrapped offset, where reads already retry + #[cfg(not(feature = "5bytes"))] + if nv.offset.to_actual_offset() < MAX_POSSIBLE_VOLUME_SIZE as i64 { + let mut wrapped = Needle::default(); + match self.read_needle_header_unlocked( + io, + &mut wrapped, + nv.offset.to_actual_offset() + MAX_POSSIBLE_VOLUME_SIZE as i64, + ) { + Ok(()) => existing = wrapped, + Err(VolumeError::Io(e)) if e.kind() == io::ErrorKind::UnexpectedEof => {} + Err(e) => return Err(e), + } + } } - if existing.cookie != n.cookie { - return Err(VolumeError::CookieMismatch(n.cookie.0)); + + if existing.id != n.id && existing.size != nv.size { + // the entry resolves to a record that is neither the indexed + // key nor the indexed size, so it is stale rather than a + // duplicate — skip the cookie check and let publish_write + // overwrite it once the append commits + warn!( + "volume {} index entry for needle {} resolves to needle {}; overwriting stale index entry", + self.id.0, n.id.0, existing.id.0 + ); + } else { + if n.cookie.0 == 0 && !check_cookie { + n.cookie = existing.cookie; + } + if existing.cookie != n.cookie { + return Err(VolumeError::CookieMismatch(n.cookie.0)); + } } } Ok(false) @@ -6067,7 +6118,7 @@ fn preallocate_file(file: &File, size: u64) { /// An OS error the platform reports for failing storage media, which is /// what counts toward the I/O error streak. #[cfg(test)] -fn media_error_for_test() -> io::Error { +pub(crate) fn media_error_for_test() -> io::Error { #[cfg(unix)] { io::Error::from_raw_os_error(libc::EIO) @@ -9044,6 +9095,107 @@ mod tests { assert!(matches!(err, VolumeError::CookieMismatch(_))); } + /// A damaged index entry resolving to a different needle must miss on + /// reads and be dropped by the next write for the key, not rejected as a + /// cookie conflict. + #[test] + fn test_stale_index_entry_heals_on_write() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let v = make_test_volume(dir); + + let mut first = Needle { + id: NeedleId(2), + cookie: Cookie(0x2222), + data: b"second".to_vec(), + data_size: 6, + ..Needle::default() + }; + v.write_needle(&mut first, true, false).unwrap(); + + // point key 9 at needle 2's record with a size that does not match + // the record, as a corrupt index entry would look + { + let mut io = v.io.write().unwrap(); + let nm = io.nm.as_mut().unwrap(); + let nv = nm.get(NeedleId(2)).unwrap().unwrap(); + nm.put(NeedleId(9), nv.offset, Size(nv.size.0 + 1)).unwrap(); + } + + let mut probe = Needle { + id: NeedleId(9), + ..Needle::default() + }; + assert!(v.read_needle(&mut probe).is_err()); + + let mut n = Needle { + id: NeedleId(9), + cookie: Cookie(0xabcd), + data: b"ninth".to_vec(), + data_size: 5, + ..Needle::default() + }; + v.write_needle(&mut n, true, false).unwrap(); + + let mut back = Needle { + id: NeedleId(9), + ..Needle::default() + }; + v.read_needle(&mut back).unwrap(); + assert_eq!(back.data, b"ninth"); + assert_eq!(back.cookie, Cookie(0xabcd)); + } + + /// A stale index entry is only replaced once the new record commits: a + /// durable write rolled back by a failed fsync leaves the index exactly + /// as it found it. + #[test] + fn test_stale_index_entry_survives_failed_write() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let v = make_test_volume(dir); + + let mut existing = Needle { + id: NeedleId(2), + cookie: Cookie(0x2222), + data: b"second".to_vec(), + data_size: 6, + ..Needle::default() + }; + v.write_needle(&mut existing, true, false).unwrap(); + + let stale_nv = { + let mut io = v.io.write().unwrap(); + let nm = io.nm.as_mut().unwrap(); + let nv = nm.get(NeedleId(2)).unwrap().unwrap(); + nm.put(NeedleId(9), nv.offset, Size(nv.size.0 + 1)).unwrap(); + nm.get(NeedleId(9)).unwrap().unwrap() + }; + + v.fail_next_fsync_for_test(true); + let mut replacement = Needle { + id: NeedleId(9), + cookie: Cookie(0xabcd), + data: b"ninth".to_vec(), + data_size: 5, + ..Needle::default() + }; + assert!(v.write_needle(&mut replacement, true, true).is_err()); + v.fail_next_fsync_for_test(false); + + let now = { + let io = v.io.read().unwrap(); + io.nm + .as_ref() + .unwrap() + .get(NeedleId(9)) + .unwrap() + .expect("a failed write must not drop the index entry it meant to replace") + }; + assert_eq!(now.offset, stale_nv.offset); + assert_eq!(now.size, stale_nv.size); + } + // A size disagreeing with the blob's own header indexes the needle at the wrong // length and, on v3, stamps the append timestamp into the middle of the needle. #[test] diff --git a/weed/server/volume_server_handlers_read.go b/weed/server/volume_server_handlers_read.go index aab0245f5e..0c46428920 100644 --- a/weed/server/volume_server_handlers_read.go +++ b/weed/server/volume_server_handlers_read.go @@ -232,6 +232,12 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) return } if n.Cookie != cookie { + // A meta-only read ended the request; record its success. + if readOption.IsMetaOnly { + if v := vs.store.GetVolume(volumeId); v != nil { + v.NoteMetaOnlySuccess() + } + } glog.V(0).Infof("request %s with cookie:%x expected:%x from %s agent %s", r.URL.Path, cookie, n.Cookie, r.RemoteAddr, r.UserAgent()) NotFound(w) return @@ -241,6 +247,11 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) if r.Header.Get("If-Modified-Since") != "" { if t, parseError := time.Parse(http.TimeFormat, r.Header.Get("If-Modified-Since")); parseError == nil { if t.Unix() >= int64(n.LastModified) { + if readOption.IsMetaOnly { + if v := vs.store.GetVolume(volumeId); v != nil { + v.NoteMetaOnlySuccess() + } + } w.WriteHeader(http.StatusNotModified) return } @@ -248,6 +259,11 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) } } if inm := r.Header.Get("If-None-Match"); inm == "\""+n.Etag()+"\"" { + if readOption.IsMetaOnly { + if v := vs.store.GetVolume(volumeId); v != nil { + v.NoteMetaOnlySuccess() + } + } w.WriteHeader(http.StatusNotModified) return } @@ -479,6 +495,10 @@ func (vs *VolumeServer) streamWriteResponseContent(filename string, mimeType str AdjustPassthroughHeaders(w, r, filename) if r.Method == http.MethodHead { + // HEAD ends at the metadata read; record its success. + if v := vs.store.GetVolume(volumeId); v != nil { + v.NoteMetaOnlySuccess() + } w.Header().Set("Content-Length", strconv.FormatInt(totalSize, 10)) return } diff --git a/weed/storage/volume_idx_repair_test.go b/weed/storage/volume_idx_repair_test.go index 197ebfade4..6b155477c5 100644 --- a/weed/storage/volume_idx_repair_test.go +++ b/weed/storage/volume_idx_repair_test.go @@ -1,6 +1,8 @@ package storage import ( + "bytes" + "errors" "os" "path/filepath" "testing" @@ -219,3 +221,96 @@ func TestRepairIdxHeadTombstones_KeepsDeletedNeedlesDeleted(t *testing.T) { t.Fatalf("needle 1 should have been recovered: %v", err) } } + +// TestStaleIndexEntry_HealsOnWrite covers a damaged index entry that resolves +// to a different needle: reads must miss instead of serving the wrong needle, +// and the next write for the key must re-index it rather than be rejected as +// a cookie conflict. +func TestStaleIndexEntry_HealsOnWrite(t *testing.T) { + dir := t.TempDir() + written := writeTestVolume(t, dir, 5) + + v := mustReload(t, dir) + defer v.Close() + + // point key 9 at needle 2's record with a size that does not match the + // record, as a corrupt index entry would look + const staleKey = 9 + nv, ok := v.nm.Get(types.Uint64ToNeedleId(2)) + if !ok { + t.Fatal("needle 2 missing from index") + } + if err := v.nm.Put(types.Uint64ToNeedleId(staleKey), nv.Offset, nv.Size+1); err != nil { + t.Fatalf("plant stale index entry: %v", err) + } + + if _, err := v.readNeedle(newEmptyNeedle(staleKey), nil, nil); err == nil { + t.Fatal("stale-index read served needle 2's data") + } + + stream := newEmptyNeedle(staleKey) + stream.DataSize = written[2].DataSize + if err := v.readNeedleDataInto(stream, &ReadOption{ReadBufferSize: 1024}, &bytes.Buffer{}, 0, int64(stream.DataSize)); err != ErrorNotFound { + t.Fatalf("stale-index stream: want ErrorNotFound, got %v", err) + } + + n := newRandomNeedle(staleKey) + n.Cookie = 0xABCD + if _, _, _, err := v.writeNeedle2(n, true, false, false); err != nil { + t.Fatalf("write over stale index entry: %v", err) + } + if got := mustRead(t, v, staleKey); !bytes.Equal(got, n.Data) { + t.Fatalf("needle %d after re-index: got %d bytes, want %d", staleKey, len(got), len(n.Data)) + } +} + +// TestStaleIndexEntry_SurvivesFailedWrite checks that a stale index entry is +// only replaced once the new record commits: a durable write rolled back by +// a failed fsync leaves the index exactly as it found it. +func TestStaleIndexEntry_SurvivesFailedWrite(t *testing.T) { + v, counting := newCountingVolume(t) + + existing := fixedNeedle(2, "existing") + if _, _, _, err := v.writeNeedle2(existing, true, true, true); err != nil { + t.Fatalf("write needle 2: %v", err) + } + nv, ok := v.nm.Get(existing.Id) + if !ok { + t.Fatal("needle 2 missing from index") + } + + staleId := types.Uint64ToNeedleId(9) + if err := v.nm.Put(staleId, nv.Offset, nv.Size+1); err != nil { + t.Fatalf("plant stale index entry: %v", err) + } + + counting.syncErr = errors.New("batch fsync failed") + if _, _, _, err := v.writeNeedle2(fixedNeedle(9, "replacement"), true, true, false); err == nil { + t.Fatal("a write whose fsync failed must not be acknowledged") + } + + entry, ok := v.nm.Get(staleId) + if !ok { + t.Fatal("a failed write dropped the index entry it meant to replace") + } + if entry.Offset != nv.Offset || entry.Size != nv.Size+1 { + t.Fatalf("index entry changed under a failed write: %+v, want %+v", entry, nv) + } +} + +// TestStaleIndexEntry_KeepsCookieConflictForSameKey checks that an entry +// resolving to the same needle id still enforces the cookie check: only +// cross-key corruption is dropped. +func TestStaleIndexEntry_KeepsCookieConflictForSameKey(t *testing.T) { + dir := t.TempDir() + writeTestVolume(t, dir, 3) + + v := mustReload(t, dir) + defer v.Close() + + n := newRandomNeedle(2) + n.Cookie = 0xABCD + if _, _, _, err := v.writeNeedle2(n, true, false, false); err == nil { + t.Fatal("write with a foreign cookie over an existing key should be rejected") + } +} diff --git a/weed/storage/volume_read.go b/weed/storage/volume_read.go index f93c1a570e..3112cabb15 100644 --- a/weed/storage/volume_read.go +++ b/weed/storage/volume_read.go @@ -61,10 +61,17 @@ func (v *Volume) readNeedle(n *needle.Needle, readOption *ReadOption, onReadSize err = n.ReadNeedleMeta(v.DataBackend, nv.Offset.ToActualOffset()+int64(MaxPossibleVolumeSize), readSize, v.Version()) } if err != nil { + v.checkReadWriteError(err) return 0, err } if !n.IsCompressed() && !n.IsChunkedManifest() { readOption.IsMetaOnly = true + // 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. + if readOption.MustMetaOnly { + v.checkReadWriteError(nil) + } } } if readOption == nil || !readOption.IsMetaOnly { @@ -91,6 +98,13 @@ func (v *Volume) readNeedle(n *needle.Needle, readOption *ReadOption, onReadSize return -1, ErrorNotFound } +// NoteMetaOnlySuccess records a request that ended at the metadata read +// (e.g. a HEAD or a cookie mismatch), so its success may reset the streak — +// unlike a meta read that precedes a payload read. +func (v *Volume) NoteMetaOnlySuccess() { + v.checkReadWriteError(nil) +} + // read needle at a specific offset func (v *Volume) readNeedleMetaAt(n *needle.Needle, offset int64, size int32) (err error) { v.dataFileAccessLock.RLock() @@ -176,6 +190,55 @@ func (v *Volume) readNeedleDataInto(n *needle.Needle, readOption *ReadOption, wr } }() + // The header is not part of the streamed body, so verify the indexed + // offset really holds this needle before sending another needle's data. + // A slow read dropped the access lock after the index lookup; take it + // back for the probe so a compaction commit cannot swap the backend + // under it, and resolve the offset the way the streaming loop does. + probeHeader := func() error { + if readOption.VolumeRevision != v.SuperBlock.CompactionRevision { + if v.nm == nil { + return ErrorNotFound + } + refreshed, found := v.nm.Get(n.Id) + if !found || refreshed.Offset.IsZero() { + return ErrorNotFound + } + nv = refreshed + actualOffset = nv.Offset.ToActualOffset() + readOption.VolumeRevision = v.SuperBlock.CompactionRevision + } + header := make([]byte, NeedleHeaderSize) + headerCount, headerErr := v.DataBackend.ReadAt(header, actualOffset) + if headerErr == io.EOF && headerCount == NeedleHeaderSize { + headerErr = nil + } + if headerErr != nil { + if headerErr != io.EOF { + v.checkReadWriteError(headerErr) + } + return fmt.Errorf("ReadNeedleData header: %w", headerErr) + } + probe := new(needle.Needle) + probe.ParseNeedleHeader(header) + if probe.Id != n.Id && probe.Size != nv.Size { + glog.V(0).Infof("volume %d index entry for needle %x resolves to needle %x; reporting not-found", + v.Id, uint64(n.Id), uint64(probe.Id)) + return ErrorNotFound + } + return nil + } + if readOption.HasSlowRead { + v.dataFileAccessLock.RLock() + } + probeErr := probeHeader() + if readOption.HasSlowRead { + v.dataFileAccessLock.RUnlock() + } + if probeErr != nil { + return probeErr + } + // read needle data crc := needle.CRC(0) var pending []byte @@ -255,11 +318,6 @@ func (v *Volume) readNeedleDataInto(n *needle.Needle, readOption *ReadOption, wr break } } - // Whole-needle read completed without a backend error — clear any - // pending EIO streak. If a non-EIO failure happens later (CRC etc.) - // we still return that error to the caller, but the disk itself - // produced clean bytes. - v.checkReadWriteError(nil) if checkCRC && (n.Checksum != crc && uint32(n.Checksum) != crc.Value()) { // the crc.Value() function is to be deprecated. this double checking is for backward compatibility // with seaweed version using crc.Value() instead of uint32(crc), which appears in commit 056c480eb @@ -268,6 +326,9 @@ func (v *Volume) readNeedleDataInto(n *needle.Needle, readOption *ReadOption, wr stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorCRC).Inc() return fmt.Errorf("ReadNeedleData checksum %v expected %v for Needle: %v,%v", crc, n.Checksum, v.Id, n) } + // The read succeeded only once the checksum passes; a failed needle + // leaves the io error streak untouched, matching the streamed path. + v.checkReadWriteError(nil) if pending != nil { if _, err = writer.Write(pending); err != nil { return fmt.Errorf("ReadNeedleData write: %w", err) diff --git a/weed/storage/volume_write.go b/weed/storage/volume_write.go index a04fa746cc..88072c9aa3 100644 --- a/weed/storage/volume_write.go +++ b/weed/storage/volume_write.go @@ -4,6 +4,7 @@ import ( "bytes" "errors" "fmt" + "io" "os" "github.com/seaweedfs/seaweedfs/weed/glog" @@ -322,22 +323,45 @@ func (v *Volume) appendWriteRequest(n *needle.Needle, checkCookie bool, fsync bo // check whether existing needle cookie matches nv, ok := v.nm.Get(n.Id) - if ok { + if ok && !nv.Offset.IsZero() { existingNeedle, _, _, existingNeedleReadErr := needle.ReadNeedleHeader(v.DataBackend, v.Version(), nv.Offset.ToActualOffset()) if existingNeedleReadErr != nil { err = fmt.Errorf("reading existing needle: %w", existingNeedleReadErr) return } - if n.Cookie == 0 && !checkCookie { - // this is from batch deletion, and read back again when tailing a remote volume - // which only happens when checkCookie == false and fsync == false - n.Cookie = existingNeedle.Cookie + if existingNeedle.Id != n.Id && existingNeedle.Size != nv.Size && + OffsetSize == 4 && nv.Offset.ToActualOffset() < int64(MaxPossibleVolumeSize) { + // a 4-byte index entry can index past the 32 GiB wrap: the low + // offset may hold another needle while the real record sits at + // the wrapped offset, where reads already retry on a mismatch + wrapped, _, _, wrappedErr := needle.ReadNeedleHeader(v.DataBackend, v.Version(), + nv.Offset.ToActualOffset()+int64(MaxPossibleVolumeSize)) + if wrappedErr == nil { + existingNeedle = wrapped + } else if wrappedErr != io.EOF { + err = fmt.Errorf("reading existing needle at wrapped offset: %w", wrappedErr) + return + } } - if existingNeedle.Cookie != n.Cookie { - glog.V(0).Infof("write cookie mismatch: existing %s, new %s", - needle.NewFileIdFromNeedle(v.Id, existingNeedle), needle.NewFileIdFromNeedle(v.Id, n)) - err = fmt.Errorf("mismatching cookie %x", n.Cookie) - return + if existingNeedle.Id != n.Id && existingNeedle.Size != nv.Size { + // the entry resolves to a record that is neither the indexed key + // nor the indexed size, so it is stale rather than a duplicate — + // skip the cookie check and let the publish step overwrite it + // once the append commits + glog.V(0).Infof("volume %d index entry for %s resolves to %s; overwriting stale index entry", + v.Id, needle.NewFileIdFromNeedle(v.Id, n), needle.NewFileIdFromNeedle(v.Id, existingNeedle)) + } else { + if n.Cookie == 0 && !checkCookie { + // this is from batch deletion, and read back again when tailing a remote volume + // which only happens when checkCookie == false and fsync == false + n.Cookie = existingNeedle.Cookie + } + if existingNeedle.Cookie != n.Cookie { + glog.V(0).Infof("write cookie mismatch: existing %s, new %s", + needle.NewFileIdFromNeedle(v.Id, existingNeedle), needle.NewFileIdFromNeedle(v.Id, n)) + err = fmt.Errorf("mismatching cookie %x", n.Cookie) + return + } } }