diff --git a/.codespellrc b/.codespellrc index 82b2e1396..038fb6524 100644 --- a/.codespellrc +++ b/.codespellrc @@ -19,4 +19,5 @@ ignore-regex = \b[a-z]+[A-Z]\w*\b|\b[A-Z][a-z]+[A-Z]\w*\b # thirdparty: literal Maven groupId `org.apache.hadoop.thirdparty` (external, cannot rename) # unknwon: GitHub username / Go module path (`github.com/unknwon/goconfig`) # atleast: CLI mode literal string in test/benchmark/fuse_db/bin/sqlite_verify.py -ignore-words-list = visibles,fo,te,ser,bject,unparseable,keep-alives,tread,anc,ue,auther,thirdparty,unknwon,atleast +# sme: local variable for a *streamMutateError in mount tests +ignore-words-list = visibles,fo,te,ser,bject,unparseable,keep-alives,tread,anc,ue,auther,thirdparty,unknwon,atleast,sme diff --git a/seaweed-volume/src/storage/mod.rs b/seaweed-volume/src/storage/mod.rs index e3d192f18..32d1274be 100644 --- a/seaweed-volume/src/storage/mod.rs +++ b/seaweed-volume/src/storage/mod.rs @@ -9,6 +9,7 @@ pub mod store_ec_reconcile; pub mod super_block; pub mod types; pub mod volume; +pub mod volume_idx_rebuild; pub mod volume_idx_repair; pub mod volume_open; pub mod volume_report; diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index eef63bee4..943307cea 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -801,6 +801,30 @@ impl Volume { } if also_load_index { + // Adjust for existing volumes with .idx together with .dat files: + // an index already beside the data keeps serving after --dir.idx + // named a different directory. + if self.dir_idx != self.dir + && Path::new(&format!("{}.idx", self.data_file_name())).exists() + { + self.dir_idx = self.dir.clone(); + } + + // A changed --dir.idx leaves the new directory without an index. + // The .dat still holds every row, so rebuild rather than mount the + // volume with every needle invisible. + if !self.has_remote_file + && !Path::new(&self.file_name(".idx")).exists() + && self.current_dat_file_size()? > SUPER_BLOCK_SIZE as u64 + { + self.rebuild_idx_file()?; + info!( + volume_id = self.id.0, + idx = %self.file_name(".idx"), + "rebuilt the index from the data file" + ); + } + // Recover rows that deletes on a tiered read-only volume overwrote // at the front of .idx. Best effort: a volume that cannot be // repaired is still servable for everything the surviving rows @@ -3995,7 +4019,7 @@ impl Volume { /// Byte offset just past the needle's on-disk record. Deletion tombstones /// carry TombstoneFileSize (-1) in the .idx but are written with DataSize=0, /// so their on-disk record is sized as 0. Mirrors Go's needleDiskEnd. -fn needle_disk_end(offset: Offset, size: Size, version: Version) -> i64 { +pub(crate) fn needle_disk_end(offset: Offset, size: Size, version: Version) -> i64 { let on_disk_size = if size.is_deleted() { Size(0) } else { size }; offset.to_actual_offset() + get_actual_size(on_disk_size, version) } @@ -4166,6 +4190,11 @@ pub fn scan_volume_file( if size.0 == 0 && _id.is_empty() { break; // end of valid data } + // A negative size is a corrupt header, and body_length would advance the + // walk backwards from it. Go's scanners stop here by returning io.EOF. + if size.0 < 0 { + break; + } let body_length = needle::needle_body_length(size, version); let total_size = NEEDLE_HEADER_SIZE as i64 + body_length; @@ -5239,6 +5268,71 @@ mod tests { assert!(Path::new(&idx_dir_idx).exists()); } + #[test] + fn test_load_keeps_index_co_located_with_the_data() { + let root = TempDir::new().unwrap(); + let data_dir = root.path().join("data"); + let idx_dir = root.path().join("idx"); + fs::create_dir_all(&data_dir).unwrap(); + fs::create_dir_all(&idx_dir).unwrap(); + let data = data_dir.to_str().unwrap(); + let idx = idx_dir.to_str().unwrap(); + + let mut v = Volume::new( + data, + data, + "", + VolumeId(7), + NeedleMapKind::InMemory, + None, + None, + 0, + Version::current(), + ) + .unwrap(); + let payload = b"payload-beside-the-data".to_vec(); + let mut n = Needle { + id: NeedleId(42), + cookie: Cookie(0x55), + data: payload.clone(), + data_size: payload.len() as u32, + ..Needle::default() + }; + v.write_needle(&mut n, true, false).unwrap(); + v.sync_to_disk().unwrap(); + drop(v); + + // --dir.idx now names an empty directory: the index already beside the + // data keeps serving, and nothing lands in the new directory. + let reopened = Volume::new( + data, + idx, + "", + VolumeId(7), + NeedleMapKind::InMemory, + None, + None, + 0, + Version::current(), + ) + .unwrap(); + assert!( + Path::new(&format!("{data}/7.idx")).exists(), + "index stays with the data" + ); + assert!( + !Path::new(&format!("{idx}/7.idx")).exists(), + "nothing written to the new idx dir" + ); + + let mut got = Needle { + id: NeedleId(42), + ..Needle::default() + }; + reopened.read_needle(&mut got).unwrap(); + assert_eq!(got.data, payload); + } + #[test] fn test_relocate_index_to_noop_when_already_in_place() { let tmp = TempDir::new().unwrap(); diff --git a/seaweed-volume/src/storage/volume_idx_rebuild.rs b/seaweed-volume/src/storage/volume_idx_rebuild.rs new file mode 100644 index 000000000..d7fbb2188 --- /dev/null +++ b/seaweed-volume/src/storage/volume_idx_rebuild.rs @@ -0,0 +1,376 @@ +//! Rebuild a missing .idx from the .dat it indexes. Mirrors +//! `weed/storage/volume_idx_rebuild.go`. + +use std::fs::{self, OpenOptions}; +use std::io::{BufWriter, Write}; +use std::path::Path; + +use crate::storage::idx; +use crate::storage::needle::Needle; +use crate::storage::super_block::SuperBlock; +use crate::storage::types::*; +use crate::storage::volume::{ + fsync_dir, needle_disk_end, scan_volume_file, Volume, VolumeError, VolumeFileVisitor, +}; + +/// Writes one .idx row per .dat record, in .dat append order, which is the +/// shape the volume server's own writes leave behind. +struct VolumeFileScanner4RebuildIdx { + writer: W, + dat_size: i64, + version: Version, + stopped: bool, +} + +impl VolumeFileVisitor for VolumeFileScanner4RebuildIdx { + fn visit_super_block(&mut self, _sb: &SuperBlock) -> Result<(), VolumeError> { + Ok(()) + } + + fn read_needle_body(&self) -> bool { + false + } + + fn visit_needle(&mut self, n: &Needle, offset: i64) -> Result<(), VolumeError> { + // A record reaching past the end of .dat is a torn append or a corrupt + // header: nothing beyond it is indexable, and a row pointing past EOF + // would fail every read of that needle. The all-zero header case ends + // the walk upstream. + if self.stopped { + return Ok(()); + } + if needle_disk_end(Offset::from_actual_offset(offset), n.size, self.version) > self.dat_size + { + self.stopped = true; + return Ok(()); + } + let size = if n.size.is_valid() { + n.size + } else { + TOMBSTONE_FILE_SIZE + }; + idx::write_index_entry( + &mut self.writer, + n.id, + Offset::from_actual_offset(offset), + size, + )?; + Ok(()) + } +} + +impl Volume { + /// Regenerate the volume's .idx from its .dat. The whole index is derivable + /// from the data file, so a volume whose index directory has no .idx -- a + /// --dir.idx pointed at an empty directory, or a lost index -- comes back on + /// its own instead of mounting with every needle invisible. The rows go to a + /// temp file that is renamed in, so an interrupted rebuild leaves no partial + /// index behind. + pub(crate) fn rebuild_idx_file(&self) -> Result<(), VolumeError> { + let idx_path = self.file_name(".idx"); + let dat_path = self.file_name(".dat"); + let tmp_path = format!("{idx_path}.tmp"); + + let rebuild = || -> Result<(), VolumeError> { + if let Some(parent) = Path::new(&idx_path).parent() { + fs::create_dir_all(parent)?; + } + let tmp_file = OpenOptions::new() + .write(true) + .create(true) + .truncate(true) + .open(&tmp_path)?; + let mut scanner = VolumeFileScanner4RebuildIdx { + writer: BufWriter::new(&tmp_file), + dat_size: fs::metadata(&dat_path)?.len() as i64, + version: self.version(), + stopped: false, + }; + scan_volume_file(&dat_path, &mut scanner)?; + scanner.writer.flush()?; + drop(scanner); + tmp_file.sync_all()?; + fs::rename(&tmp_path, &idx_path)?; + fsync_dir(&idx_path)?; + Ok(()) + }; + + let result = rebuild(); + if result.is_err() { + let _ = fs::remove_file(&tmp_path); + } + result + } +} + +#[cfg(test)] +mod tests { + use crate::storage::needle::crc::CRC; + use crate::storage::needle::Needle; + use crate::storage::needle_map::NeedleMapKind; + use crate::storage::types::*; + use crate::storage::volume::Volume; + use std::fs; + use std::path::Path; + use tempfile::TempDir; + + fn needle(id: u64) -> Needle { + let data = format!("payload-{id}").into_bytes(); + Needle { + id: NeedleId(id), + cookie: Cookie(0x55), + data_size: data.len() as u32, + checksum: CRC::new(&data), + data, + ..Needle::default() + } + } + + // Pointing --dir.idx at a directory with no .idx used to mount the volume on + // an empty index; the index is derivable from the .dat, so it must be + // rebuilt in place instead. + #[test] + fn test_load_moved_idx_directory_rebuilds_idx() { + let root = TempDir::new().unwrap(); + let data_dir = root.path().join("data"); + let old_idx_dir = root.path().join("idxA"); + let new_idx_dir = root.path().join("idxB"); + for dir in [&data_dir, &old_idx_dir, &new_idx_dir] { + fs::create_dir_all(dir).unwrap(); + } + let data = data_dir.to_str().unwrap(); + let old_idx = old_idx_dir.to_str().unwrap(); + let new_idx = new_idx_dir.to_str().unwrap(); + + let mut v = Volume::new( + data, + old_idx, + "", + VolumeId(1), + NeedleMapKind::InMemory, + None, + None, + 0, + Version::current(), + ) + .unwrap(); + for id in 1..=3 { + v.write_needle(&mut needle(id), true, false).unwrap(); + } + v.delete_needle(&mut needle(2)).unwrap(); + v.sync_to_disk().unwrap(); + let (want_count, want_deleted) = (v.file_count(), v.deleted_count()); + drop(v); + + let seeded = fs::read(format!("{old_idx}/1.idx")).unwrap(); + + let reopened = Volume::new( + data, + new_idx, + "", + VolumeId(1), + NeedleMapKind::InMemory, + None, + None, + 0, + Version::current(), + ) + .unwrap(); + + assert!( + Path::new(&format!("{new_idx}/1.idx")).exists(), + "idx not rebuilt in the new idx dir" + ); + assert_eq!(reopened.file_count(), want_count); + assert_eq!(reopened.deleted_count(), want_deleted); + assert_eq!( + fs::read(format!("{new_idx}/1.idx")).unwrap(), + seeded, + "rebuilt idx differs from the one the server wrote" + ); + + for id in [1, 3] { + let mut got = needle(id); + got.data.clear(); + reopened.read_needle(&mut got).unwrap(); + assert_eq!(got.data, format!("payload-{id}").into_bytes()); + } + } + + // A .dat padded with zeros must not be indexed as needle 0 rows: the walk + // stops where the records do. + #[test] + fn test_rebuild_idx_stops_at_zero_padded_dat_tail() { + let root = TempDir::new().unwrap(); + let dir = root.path().to_str().unwrap(); + + let mut v = Volume::new( + dir, + dir, + "", + VolumeId(1), + NeedleMapKind::InMemory, + None, + None, + 0, + Version::current(), + ) + .unwrap(); + v.write_needle(&mut needle(1), true, false).unwrap(); + v.sync_to_disk().unwrap(); + drop(v); + + let seeded = fs::read(format!("{dir}/1.idx")).unwrap(); + let dat = fs::OpenOptions::new() + .write(true) + .open(format!("{dir}/1.dat")) + .unwrap(); + let dat_size = dat.metadata().unwrap().len(); + dat.set_len(dat_size + 4096).unwrap(); + drop(dat); + fs::remove_file(format!("{dir}/1.idx")).unwrap(); + + let reopened = Volume::new( + dir, + dir, + "", + VolumeId(1), + NeedleMapKind::InMemory, + None, + None, + 0, + Version::current(), + ) + .unwrap(); + drop(reopened); + + assert_eq!( + fs::read(format!("{dir}/1.idx")).unwrap(), + seeded, + "the zero-padded tail leaked into the rebuilt idx" + ); + } + + // A .dat whose last append was torn mid-body must not gain an index row + // that points past the end of the file. + #[test] + fn test_rebuild_idx_skips_truncated_dat_tail() { + let root = TempDir::new().unwrap(); + let dir = root.path().to_str().unwrap(); + + let mut v = Volume::new( + dir, + dir, + "", + VolumeId(1), + NeedleMapKind::InMemory, + None, + None, + 0, + Version::current(), + ) + .unwrap(); + v.write_needle(&mut needle(1), true, false).unwrap(); + v.sync_to_disk().unwrap(); + let kept = fs::read(format!("{dir}/1.idx")).unwrap(); + v.write_needle(&mut needle(2), true, false).unwrap(); + v.sync_to_disk().unwrap(); + drop(v); + + // Chop the second needle's body, leaving its header intact. + let dat = fs::OpenOptions::new() + .write(true) + .open(format!("{dir}/1.dat")) + .unwrap(); + let torn_size = dat.metadata().unwrap().len() - 8; + dat.set_len(torn_size).unwrap(); + drop(dat); + fs::remove_file(format!("{dir}/1.idx")).unwrap(); + + let reopened = Volume::new( + dir, + dir, + "", + VolumeId(1), + NeedleMapKind::InMemory, + None, + None, + 0, + Version::current(), + ) + .unwrap(); + drop(reopened); + + assert_eq!( + fs::read(format!("{dir}/1.idx")).unwrap(), + kept, + "the torn record leaked into the rebuilt idx" + ); + } + + // A corrupt header carrying a negative size advances the .dat walk + // backwards, which cycles forever between it and the record before it. + #[test] + fn test_rebuild_idx_stops_at_negative_size_header() { + let root = TempDir::new().unwrap(); + let dir = root.path().to_str().unwrap(); + + let mut v = Volume::new( + dir, + dir, + "", + VolumeId(1), + NeedleMapKind::InMemory, + None, + None, + 0, + Version::current(), + ) + .unwrap(); + v.write_needle(&mut needle(1), true, false).unwrap(); + v.sync_to_disk().unwrap(); + let kept = fs::read(format!("{dir}/1.idx")).unwrap(); + v.write_needle(&mut needle(2), true, false).unwrap(); + v.sync_to_disk().unwrap(); + drop(v); + + let rows = fs::read(format!("{dir}/1.idx")).unwrap(); + let (_, offset, _) = idx_entry_from_bytes(&rows[NEEDLE_MAP_ENTRY_SIZE..]); + let corrupt_at = offset.to_actual_offset(); + + // Overwrite the second needle's size field with a negative i32. + use std::io::{Seek, SeekFrom, Write}; + let mut dat = fs::OpenOptions::new() + .write(true) + .open(format!("{dir}/1.dat")) + .unwrap(); + dat.seek(SeekFrom::Start( + corrupt_at as u64 + COOKIE_SIZE as u64 + NEEDLE_ID_SIZE as u64, + )) + .unwrap(); + dat.write_all(&[0xff, 0xff, 0xf0, 0x00]).unwrap(); + drop(dat); + fs::remove_file(format!("{dir}/1.idx")).unwrap(); + + // Pre-fix the rebuild walked backwards from here and never terminated. + let reopened = Volume::new( + dir, + dir, + "", + VolumeId(1), + NeedleMapKind::InMemory, + None, + None, + 0, + Version::current(), + ) + .unwrap(); + drop(reopened); + + assert_eq!( + fs::read(format!("{dir}/1.idx")).unwrap(), + kept, + "the corrupt record leaked into the rebuilt idx" + ); + } +} diff --git a/weed/storage/disk_location.go b/weed/storage/disk_location.go index 59f121f14..ed17faa98 100644 --- a/weed/storage/disk_location.go +++ b/weed/storage/disk_location.go @@ -91,6 +91,9 @@ func NewDiskLocation(dir string, maxVolumeCount int32, minFreeSpace util.MinFree idxDir = dir } else { idxDir = util.ResolvePath(idxDir) + if err := os.MkdirAll(idxDir, 0755); err != nil { + glog.Fatalf("cannot create idx dir %s: %v", idxDir, err) + } } dirUuid, err := GenerateDirUuid(dir) if err != nil { diff --git a/weed/storage/disk_location_test.go b/weed/storage/disk_location_test.go index 2825af356..e26939d53 100644 --- a/weed/storage/disk_location_test.go +++ b/weed/storage/disk_location_test.go @@ -1,6 +1,8 @@ package storage import ( + "os" + "path/filepath" "reflect" "testing" "time" @@ -139,3 +141,17 @@ func TestCheckDiskSpaceProbesWithTheDiskTypeThreshold(t *testing.T) { } } } + +// -dir.idx naming a directory that does not exist yet used to leave every +// volume unable to open its index. +func TestNewDiskLocation_CreatesIdxDirectory(t *testing.T) { + root := t.TempDir() + idxDir := filepath.Join(root, "idx", "nested") + + loc := NewDiskLocation(root, 1, util.MinFreeSpace{}, idxDir, types.HardDriveType, nil, stats.DiskIOProbeConfig{}) + defer loc.Close() + + if _, err := os.Stat(idxDir); err != nil { + t.Fatalf("idx dir not created: %v", err) + } +} diff --git a/weed/storage/volume_idx_rebuild.go b/weed/storage/volume_idx_rebuild.go new file mode 100644 index 000000000..214a99d32 --- /dev/null +++ b/weed/storage/volume_idx_rebuild.go @@ -0,0 +1,98 @@ +package storage + +import ( + "bufio" + "fmt" + "io" + "os" + "path/filepath" + + "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" +) + +// VolumeFileScanner4RebuildIdx writes one .idx row per .dat record, in .dat +// append order, which is the shape the volume server's own writes leave behind. +type VolumeFileScanner4RebuildIdx struct { + writer *bufio.Writer + datSize int64 + version needle.Version +} + +func (scanner *VolumeFileScanner4RebuildIdx) VisitSuperBlock(superBlock super_block.SuperBlock) error { + return nil +} + +func (scanner *VolumeFileScanner4RebuildIdx) ReadNeedleBody() bool { + return false +} + +func (scanner *VolumeFileScanner4RebuildIdx) VisitNeedle(n *needle.Needle, offset int64, needleHeader, needleBody []byte) error { + // Stop at the first thing that is not a record: an all-zero header is + // unwritten space, a negative size is a corrupt header, and a record + // reaching past the end of .dat is a torn append. Indexing any of them + // fabricates rows, and io.EOF here stops the walk before it advances by a + // bad size -- a negative one moves the offset backwards. + if (n.Size == 0 && n.Id == 0) || n.Size < 0 { + return io.EOF + } + if needleDiskEnd(types.ToOffset(offset), n.Size, scanner.version) > scanner.datSize { + return io.EOF + } + size := n.Size + if !size.IsValid() { + size = types.TombstoneFileSize + } + _, err := scanner.writer.Write(needle_map.ToBytes(n.Id, types.ToOffset(offset), size)) + return err +} + +// rebuildIdxFile regenerates the volume's .idx from its .dat. The whole index +// is derivable from the data file, so a volume whose index directory has no +// .idx -- a -dir.idx pointed at an empty directory, or a lost index -- comes +// back on its own instead of taking the volume server down. The rows go to a +// temp file that is renamed in, so an interrupted rebuild leaves no partial +// index behind. +func (v *Volume) rebuildIdxFile() error { + if v.DataBackend == nil { + return fmt.Errorf("volume %d has no data backend", v.Id) + } + + datSize, _, err := v.DataBackend.GetStat() + if err != nil { + return fmt.Errorf("stat %s: %w", v.FileName(".dat"), err) + } + + idxFileName := v.FileName(".idx") + if err := os.MkdirAll(filepath.Dir(idxFileName), 0755); err != nil { + return fmt.Errorf("create idx dir for %s: %w", idxFileName, err) + } + tmpFileName := idxFileName + ".tmp" + tmpFile, err := os.OpenFile(tmpFileName, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644) + if err != nil { + return fmt.Errorf("create %s: %w", tmpFileName, err) + } + defer os.Remove(tmpFileName) + + scanner := &VolumeFileScanner4RebuildIdx{writer: bufio.NewWriter(tmpFile), datSize: datSize, version: v.Version()} + err = ScanVolumeFileFrom(scanner.version, v.DataBackend, int64(v.SuperBlock.BlockSize()), scanner) + if err == nil { + err = scanner.writer.Flush() + } + if err == nil { + err = tmpFile.Sync() + } + if closeErr := tmpFile.Close(); err == nil { + err = closeErr + } + if err != nil { + return fmt.Errorf("rebuild %s from %s: %w", idxFileName, v.FileName(".dat"), err) + } + + if err := os.Rename(tmpFileName, idxFileName); err != nil { + return fmt.Errorf("rename %s: %w", tmpFileName, err) + } + return fsyncDir(filepath.Dir(idxFileName)) +} diff --git a/weed/storage/volume_idx_rebuild_test.go b/weed/storage/volume_idx_rebuild_test.go new file mode 100644 index 000000000..a71abee83 --- /dev/null +++ b/weed/storage/volume_idx_rebuild_test.go @@ -0,0 +1,265 @@ +package storage + +import ( + "bytes" + "os" + "path/filepath" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" + "github.com/seaweedfs/seaweedfs/weed/storage/types" +) + +// Pointing -dir.idx at a directory with no .idx used to abort the whole volume +// server in checkIdxFile; the index is derivable from the .dat, so it must be +// rebuilt in place instead. +func TestLoad_MovedIdxDirectory_RebuildsIdx(t *testing.T) { + root := t.TempDir() + dataDir := filepath.Join(root, "data") + oldIdxDir := filepath.Join(root, "idxA") + newIdxDir := filepath.Join(root, "idxB") + for _, dir := range []string{dataDir, oldIdxDir, newIdxDir} { + if err := os.MkdirAll(dir, 0755); err != nil { + t.Fatalf("mkdir %s: %v", dir, err) + } + } + + v, err := NewVolume(dataDir, oldIdxDir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("create volume: %v", err) + } + for id := uint64(1); id <= 3; id++ { + if _, _, _, err := v.writeNeedle2(newRandomNeedle(id), true, false, false); err != nil { + t.Fatalf("seed write %d: %v", id, err) + } + } + if _, err := v.deleteNeedle2(newRandomNeedle(2)); err != nil { + t.Fatalf("seed delete: %v", err) + } + wantCount, wantDeleted := v.nm.FileCount(), v.nm.DeletedCount() + v.Close() + + if _, err := os.Stat(filepath.Join(oldIdxDir, "1.idx")); err != nil { + t.Fatalf("seeded idx should live in the old idx dir: %v", err) + } + + v2, err := NewVolume(dataDir, newIdxDir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("reload against the new idx dir: %v", err) + } + defer v2.Close() + + if _, err := os.Stat(filepath.Join(newIdxDir, "1.idx")); err != nil { + t.Fatalf("idx not rebuilt in the new idx dir: %v", err) + } + if got := v2.nm.FileCount(); got != wantCount { + t.Errorf("file count = %d, want %d", got, wantCount) + } + if got := v2.nm.DeletedCount(); got != wantDeleted { + t.Errorf("deleted count = %d, want %d", got, wantDeleted) + } + old, err := os.ReadFile(filepath.Join(oldIdxDir, "1.idx")) + if err != nil { + t.Fatalf("read the seeded idx: %v", err) + } + rebuilt, err := os.ReadFile(filepath.Join(newIdxDir, "1.idx")) + if err != nil { + t.Fatalf("read the rebuilt idx: %v", err) + } + if !bytes.Equal(old, rebuilt) { + t.Errorf("rebuilt idx (%d bytes) differs from the one the server wrote (%d bytes)", len(rebuilt), len(old)) + } + if v2.noWriteOrDelete { + t.Errorf("volume marked read-only after the rebuild") + } +} + +// A .dat padded with zeros must not be indexed as needle 0 rows: the walk stops +// where the records do. +func TestRebuildIdx_StopsAtZeroPaddedDatTail(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("create volume: %v", err) + } + if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false, false); err != nil { + t.Fatalf("seed write: %v", err) + } + v.Close() + + base := VolumeFileName(dir, "", 1) + seeded, err := os.ReadFile(base + ".idx") + if err != nil { + t.Fatalf("read the seeded idx: %v", err) + } + datSize, err := os.Stat(base + ".dat") + if err != nil { + t.Fatalf("stat dat: %v", err) + } + if err := os.Truncate(base+".dat", datSize.Size()+4096); err != nil { + t.Fatalf("pad dat: %v", err) + } + if err := os.Remove(base + ".idx"); err != nil { + t.Fatalf("drop idx: %v", err) + } + + v2, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("reload: %v", err) + } + defer v2.Close() + + rebuilt, err := os.ReadFile(base + ".idx") + if err != nil { + t.Fatalf("read the rebuilt idx: %v", err) + } + if !bytes.Equal(seeded, rebuilt) { + t.Errorf("rebuilt idx has %d bytes, want the %d the server wrote", len(rebuilt), len(seeded)) + } +} + +// A .dat whose last append was torn mid-body must not gain an index row that +// points past the end of the file. +func TestRebuildIdx_SkipsTruncatedDatTail(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("create volume: %v", err) + } + if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false, false); err != nil { + t.Fatalf("seed write: %v", err) + } + base := VolumeFileName(dir, "", 1) + kept, err := os.ReadFile(base + ".idx") + if err != nil { + t.Fatalf("read the seeded idx: %v", err) + } + if _, _, _, err := v.writeNeedle2(newRandomNeedle(2), true, false, false); err != nil { + t.Fatalf("seed torn write: %v", err) + } + v.Close() + + // Chop the second needle's body, leaving its header intact. + datSize, err := os.Stat(base + ".dat") + if err != nil { + t.Fatalf("stat dat: %v", err) + } + if err := os.Truncate(base+".dat", datSize.Size()-8); err != nil { + t.Fatalf("tear dat: %v", err) + } + if err := os.Remove(base + ".idx"); err != nil { + t.Fatalf("drop idx: %v", err) + } + + v2, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("reload: %v", err) + } + defer v2.Close() + + rebuilt, err := os.ReadFile(base + ".idx") + if err != nil { + t.Fatalf("read the rebuilt idx: %v", err) + } + if !bytes.Equal(kept, rebuilt) { + t.Errorf("rebuilt idx has %d bytes, want the %d covering only the intact needle", len(rebuilt), len(kept)) + } + if maxEnd := v2.nm.MaxNeedleEnd(); maxEnd > datSize.Size()-8 { + t.Errorf("rebuilt idx reaches %d, past the %d-byte .dat", maxEnd, datSize.Size()-8) + } +} + +// A corrupt header carrying a negative size advances the .dat walk backwards, +// which cycles forever between it and the record before it. +func TestRebuildIdx_StopsAtNegativeSizeHeader(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("create volume: %v", err) + } + if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false, false); err != nil { + t.Fatalf("seed write: %v", err) + } + base := VolumeFileName(dir, "", 1) + kept, err := os.ReadFile(base + ".idx") + if err != nil { + t.Fatalf("read the seeded idx: %v", err) + } + if _, _, _, err := v.writeNeedle2(newRandomNeedle(2), true, false, false); err != nil { + t.Fatalf("seed second write: %v", err) + } + nv, ok := v.nm.Get(2) + if !ok { + t.Fatalf("second needle missing from the index") + } + corruptAt := nv.Offset.ToActualOffset() + v.Close() + + // Overwrite the second needle's size field with a negative int32. + dat, err := os.OpenFile(base+".dat", os.O_WRONLY, 0644) + if err != nil { + t.Fatalf("open dat: %v", err) + } + if _, err := dat.WriteAt([]byte{0xff, 0xff, 0xf0, 0x00}, corruptAt+types.CookieSize+types.NeedleIdSize); err != nil { + t.Fatalf("corrupt size field: %v", err) + } + dat.Close() + if err := os.Remove(base + ".idx"); err != nil { + t.Fatalf("drop idx: %v", err) + } + + // Pre-fix the rebuild walked backwards from here and never terminated. + v2, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("reload: %v", err) + } + defer v2.Close() + + rebuilt, err := os.ReadFile(base + ".idx") + if err != nil { + t.Fatalf("read the rebuilt idx: %v", err) + } + if !bytes.Equal(kept, rebuilt) { + t.Errorf("rebuilt idx has %d bytes, want the %d covering only the intact needle", len(rebuilt), len(kept)) + } +} + +// A rebuild that cannot write skips just this volume; it used to call +// glog.Fatalf and take the whole volume server down with it. +func TestRebuildIdx_UnwritableIdxDirSkipsOnlyThisVolume(t *testing.T) { + if os.Geteuid() == 0 { + t.Skip("root ignores directory permissions") + } + root := t.TempDir() + dataDir := filepath.Join(root, "data") + idxDir := filepath.Join(root, "idx") + for _, dir := range []string{dataDir, idxDir} { + if err := os.MkdirAll(dir, 0755); err != nil { + t.Fatalf("mkdir %s: %v", dir, err) + } + } + + v, err := NewVolume(dataDir, dataDir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("create volume: %v", err) + } + if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false, false); err != nil { + t.Fatalf("seed write: %v", err) + } + v.Close() + if err := os.Remove(VolumeFileName(dataDir, "", 1) + ".idx"); err != nil { + t.Fatalf("drop idx: %v", err) + } + if err := os.Chmod(idxDir, 0555); err != nil { + t.Fatalf("seal idx dir: %v", err) + } + defer os.Chmod(idxDir, 0755) + + if _, err := NewVolume(dataDir, idxDir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0); err == nil { + t.Errorf("expected a load error for an unwritable idx dir") + } +} diff --git a/weed/storage/volume_loading.go b/weed/storage/volume_loading.go index 7d6ec9abe..e78d048f0 100644 --- a/weed/storage/volume_loading.go +++ b/weed/storage/volume_loading.go @@ -254,7 +254,13 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind glog.Errorf("skip remote volume %d (idx: %s): %v", v.Id, v.FileName(".idx"), err) return fmt.Errorf("check volume idx file %s: %w", v.FileName(".idx"), err) } - glog.Fatalf("check volume idx file %s: %v", v.FileName(".idx"), err) + // A changed -dir.idx leaves the new directory without an index. + // The .dat still holds every row, so rebuild rather than exit. + if rebuildErr := v.rebuildIdxFile(); rebuildErr != nil { + glog.Errorf("skip volume %d (idx: %s): %v", v.Id, v.FileName(".idx"), rebuildErr) + return fmt.Errorf("rebuild volume idx file %s: %w", v.FileName(".idx"), rebuildErr) + } + glog.V(0).Infof("volume %d: rebuilt %s from %s", v.Id, v.FileName(".idx"), v.FileName(".dat")) } // Recover rows that deletes on a tiered read-only volume overwrote at // the front of .idx. Best effort: a volume that cannot be repaired is