From ab738bfd3339dcf4de6dd3e41ce8e9230a791e35 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 20 Oct 2026 23:13:51 -0700 Subject: [PATCH] rust volume: repair an unindexed .dat tail at load (#11878) * rust volume: repair an unindexed .dat tail at load Mirror Go's repairUnindexedDatTail (#11870): a crash between the .dat\nappend and the .idx append leaves a .dat tail no index entry covers,\nwhich sent the volume read-only on the next load and left acknowledged\nneedles unreadable. Before the index loads, re-index the complete\nneedles in the tail (a size 0 record as the tombstone or empty write\nresolveEmptyTailRecords reads it as) and cut only an incomplete last\nrecord. A tail an interrupted append does not explain is left as it is\nand the integrity check quarantines the volume read-only as before. Generated with [Devin](https://devin.ai)\n\nCo-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * rust volume: tests for the unindexed .dat tail repair Reload volumes through the real load path after dropping .idx entries\nand cutting or damaging the .dat, mirroring Go's\nvolume_checking_tail_repair_test.go: complete unindexed needles are\nre-indexed (deletes as tombstones, empty writes as empty needles), an\nincomplete last record is cut, and a tail an interrupted append does\nnot explain is left untouched with the volume read-only. Generated with [Devin](https://devin.ai)\n\nCo-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume: re-index a torn size 0 record as a live empty needle A delete and a write with no data append the identical size 0 record, so a recovered one cannot be told apart from the prior index state alone: the write path accepts an empty overwrite of a live key and indexes it with size 0, while a delete of the same key publishes a tombstone. Inferring a tombstone from prior live state turned a lost index row after an empty overwrite into a deletion -- the object read back not-found although the write was answered. Every re-indexed record is now indexed at its on-disk size, the entry publishWriteRequest writes for an empty write. The other misread is bounded: a recovered delete reads its key back as an empty object rather than not-found, and the size 0 record carries no data to resurrect. Review feedback on the Rust mirror (#11878): the same bug exists here; fixing both sides identically. Generated with [Devin](https://devin.ai)\n\nCo-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * rust volume: re-index a torn size 0 record as a live empty needle Mirrors the Go fix in this pull request. A delete and a write with no data append the identical size 0 record, so resolve_empty_tail_records could not tell them apart from the prior index state: it treated a zero-size record of a live id as a tombstone, although the write path indexes the same bytes as a live size 0 entry. A crash after an empty overwrite of a live key therefore recovered the object as deleted. Every re-indexed record is now indexed at its on-disk size, the entry publish_write writes for an empty write. A recovered delete reads its key back as an empty object rather than not-found; the size 0 record carries no data to resurrect. Review: greptile-apps[bot] and devin-ai-integration[bot] on #11878. Generated with [Devin](https://devin.ai)\n\nCo-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume: repair a .dat tail left past an empty .idx The writable index loader creates the .idx before the first needle append, so a crash after that append leaves a volume whose data nothing indexes. CheckVolumeDataIntegrity skipped an empty index outright, so the volume loaded writable and the surviving first needle stayed invisible while later writes appended past it. A zero-size index now runs the same repair with tailEntryPos -1: the walk starts at the first record past the superblock with no anchor needle and no append timestamp to compare against, gets the same complete-record validation and incomplete-tail policy, and re-verifies through the normal tail check once it wrote entries. A tail the repair cannot explain fails the check, so the volume loads read-only instead of writable on an unaccounted-for .dat. Review feedback on the Rust mirror (#11878): the same bug exists here; fixing both sides identically. Generated with [Devin](https://devin.ai)\n\nCo-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * rust volume: repair a .dat tail left past an empty .idx Mirrors the Go fix in this pull request. The writable index loader creates the .idx before the first needle append, so a crash after that append leaves a volume whose data nothing indexes. Both the repair and the integrity check skipped an empty index, so the volume loaded writable and the surviving first needle stayed invisible while later writes appended past it. An empty .idx now runs the same repair with no anchor: the walk starts at the first record past the superblock, and a tail it cannot explain fails the load like any other repair failure, since every later check skips an empty index. Review: devin-ai-integration[bot] on #11878. Generated with [Devin](https://devin.ai)\n\nCo-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume: sync the .dat before publishing re-indexed .idx rows A power loss after the repaired .idx rows become durable but before the .dat tail records they name are flushed leaves the index pointing past the end of the data file; the next load then keeps the volume read-only until an operator steps in. Sync the .dat first so the rows never outlive the records they reference. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * rust volume: sync the .dat before publishing re-indexed .idx rows Mirrors the Go fix: a power loss after the repaired .idx rows become durable but before the .dat tail records they name are flushed leaves the index pointing past the end of the data file. Sync the .dat first. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- seaweed-volume/src/storage/mod.rs | 1 + seaweed-volume/src/storage/volume.rs | 51 +- .../src/storage/volume_tail_repair.rs | 1148 +++++++++++++++++ weed/storage/volume_checking.go | 156 +-- .../volume_checking_tail_repair_test.go | 156 ++- 5 files changed, 1427 insertions(+), 85 deletions(-) create mode 100644 seaweed-volume/src/storage/volume_tail_repair.rs diff --git a/seaweed-volume/src/storage/mod.rs b/seaweed-volume/src/storage/mod.rs index c2d0561fc4..dab21dd2d9 100644 --- a/seaweed-volume/src/storage/mod.rs +++ b/seaweed-volume/src/storage/mod.rs @@ -18,3 +18,4 @@ pub mod volume_idx_repair; pub mod volume_open; pub mod volume_report; pub mod volume_report_hash; +pub mod volume_tail_repair; diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 1096010150..d8e8f8aff3 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -29,6 +29,7 @@ use crate::storage::needle_map::{CompactNeedleMap, NeedleMap, NeedleMapKind, Red use crate::storage::super_block::{ReplicaPlacement, SUPER_BLOCK_SIZE, SuperBlock}; use crate::storage::types::*; use crate::storage::volume_open::open_volume_file; +use crate::storage::volume_tail_repair::TailRepairError; // ============================================================================ // Errors @@ -1724,6 +1725,46 @@ impl Volume { ), } + // A crash between the .dat append and the .idx append leaves a + // .dat tail no index entry covers. Re-index complete needles and + // cut an incomplete remainder before the index loads, so the + // recovered rows land in the needle map and inside its append + // offset. Mirrors Go running repairUnindexedDatTail inside + // CheckVolumeDataIntegrity ahead of the needle map load. Only a + // volume about to load writable is repaired. + if !self.has_remote_file() && self.no_write_flags() == (false, false) { + // An empty .idx gets the same repair: every check below + // skips an empty index on its own, so a tail the repair + // cannot explain there fails the load like in Go rather + // than warn and load writable. + let idx_empty = fs::metadata(self.file_name(".idx")) + .map(|m| m.len() == 0) + .unwrap_or(false); + match self.repair_unindexed_dat_tail(io) { + Ok(()) => {} + Err(e) if e.is_not_torn() && !idx_empty => warn!( + volume_id = self.id.0, + error = %e, + "unindexed .dat tail not repaired" + ), + // The index refusing a read-write open is the same failure + // Go reports opening it ahead of the check: it fails the + // load rather than only the repair. + Err(TailRepairError::OpenIndex(e)) => return Err(e), + Err(e) => { + // A repair that could not complete must not load the + // volume writable on files it half-fixed: the check + // fails as in Go, so the volume loads read-only. + self.set_no_write_flag(NO_WRITE_OR_DELETE, true); + warn!( + volume_id = self.id.0, + error = %e, + "repair unindexed .dat tail" + ); + } + } + } + self.load_index(io)?; // Match Go: CheckVolumeDataIntegrity after loading index (volume_loading.go L154-159) @@ -2082,7 +2123,7 @@ impl Volume { self.current_dat_file_size_io(&io) } - fn current_dat_file_size_io(&self, io: &VolumeIo) -> io::Result { + pub(crate) fn current_dat_file_size_io(&self, io: &VolumeIo) -> io::Result { if let Some(ref f) = io.dat_file { Ok(f.metadata()?.len()) } else if let Some(ref remote_dat_file) = io.remote_dat_file { @@ -2306,7 +2347,7 @@ impl Volume { self.read_needle_blob_unlocked(&io, offset, size) } - fn read_needle_blob_unlocked( + pub(crate) fn read_needle_blob_unlocked( &self, io: &VolumeIo, offset: i64, @@ -2726,7 +2767,7 @@ impl Volume { /// Flush the .dat, the first half of a durable write. The .idx is flushed /// separately by flush_idx once the row is published; the two are split so /// nothing is indexed before the bytes it points at are down. - fn flush_dat(&self, io: &VolumeIo) -> io::Result<()> { + pub(crate) fn flush_dat(&self, io: &VolumeIo) -> io::Result<()> { #[cfg(test)] self.dat_syncs_for_test.fetch_add(1, Ordering::Relaxed); #[cfg(test)] @@ -2778,7 +2819,7 @@ impl Volume { } /// Take an unflushed append back off the .dat after its sync failed. - fn truncate_dat(&self, io: &VolumeIo, len: u64) -> io::Result<()> { + pub(crate) fn truncate_dat(&self, io: &VolumeIo, len: u64) -> io::Result<()> { #[cfg(test)] if io.fail_truncate_for_test { return Err(io::Error::other("injected truncate failure")); @@ -3512,7 +3553,7 @@ impl Volume { /// key-sorted .idx (e.g. legacy weed fix output) falls back to a single /// linear scan for the maximum offset. Returns -1 when no entry has a /// non-zero offset (a pre-allocated / all-zero index). - fn find_dat_tail_entry_offset( + pub(crate) fn find_dat_tail_entry_offset( &self, io: &VolumeIo, idx_file: &mut File, diff --git a/seaweed-volume/src/storage/volume_tail_repair.rs b/seaweed-volume/src/storage/volume_tail_repair.rs new file mode 100644 index 0000000000..d747de1f5d --- /dev/null +++ b/seaweed-volume/src/storage/volume_tail_repair.rs @@ -0,0 +1,1148 @@ +//! Repair of the .dat tail a crash between the needle append and the .idx +//! append leaves. Mirrors `repairUnindexedDatTail` and its helpers in +//! weed/storage/volume_checking.go (#11864). +//! +//! A write appends the needle to .dat, then its entry to .idx, then answers +//! the client. A process killed between the first two steps leaves a needle, +//! or part of one, past the end of the needle at the highest indexed offset. +//! After a power loss a complete needle can survive whose .idx entry did +//! not, although the client was answered. So a complete needle is never +//! dropped: every complete needle in the tail gets the .idx entry the volume +//! server would have written, in .dat order, the .dat synced first so the +//! rows never become durable ahead of the records they name, and only an +//! incomplete last record is cut. A size 0 record is indexed at size 0, the +//! same live entry +//! the write path publishes for an empty write: a delete appends the +//! identical record, so the two cannot be told apart, and resurrecting an +//! empty placeholder beats a wrong tombstone, which drops an acknowledged +//! write's object entirely. + +use std::fmt; +use std::fs::{self, File, OpenOptions}; +use std::io::{self, Read, Seek, SeekFrom, Write}; +use std::path::Path; + +use tracing::warn; + +use crate::storage::needle::needle::{Needle, NeedleError, get_actual_size}; +use crate::storage::super_block::SUPER_BLOCK_SIZE; +use crate::storage::types::*; +use crate::storage::volume::{Volume, VolumeError, VolumeIo, needle_disk_end}; + +/// What stopped the unindexed-.dat-tail repair. Mirrors the sentinels Go +/// distinguishes: a `NotTorn` tail is left untouched for the integrity check +/// to quarantine read-only, a `Corrupt` record is damaged bytes (Go errors +/// wrapping needle.ErrorCorrupted) that `walk_unindexed_dat_tail` resolves +/// into a cut or a `NotTorn`, an `OpenIndex` failure is the .idx refusing a +/// read-write open — which in Go fails the whole load before the check ever +/// runs — and `Failed` is any other repair failure, which fails the check so +/// the volume loads read-only. +#[derive(Debug)] +pub(crate) enum TailRepairError { + NotTorn(String), + Corrupt(String), + OpenIndex(VolumeError), + Failed(VolumeError), +} + +impl fmt::Display for TailRepairError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + TailRepairError::NotTorn(why) | TailRepairError::Corrupt(why) => { + write!(f, "{why}: not a torn tail") + } + TailRepairError::OpenIndex(e) | TailRepairError::Failed(e) => write!(f, "{e}"), + } + } +} + +impl TailRepairError { + /// Mirrors Go's `errors.Is(err, errTailNotTorn)`: a tail that is not the + /// result of an interrupted append keeps the volume loading read-only + /// through the integrity check; a `Corrupt` record is resolved inside + /// the walk and everything else is a real repair failure. + pub(crate) fn is_not_torn(&self) -> bool { + matches!(self, TailRepairError::NotTorn(_)) + } +} + +impl From for TailRepairError { + fn from(e: io::Error) -> Self { + TailRepairError::Failed(VolumeError::Io(e)) + } +} + +impl From for TailRepairError { + fn from(e: VolumeError) -> Self { + TailRepairError::Failed(e) + } +} + +impl Volume { + /// Re-index the complete needles a crash left past the last indexed + /// needle of a local version 3 volume, and cut an incomplete remainder. + /// Mirrors Go's repairUnindexedDatTail. + /// + /// Runs ahead of the index load so the re-indexed rows land in the + /// needle map and inside its append offset. Only a volume about to load + /// writable is repaired; the caller checks the no-write flags. A crash + /// after the very first append leaves the needle past an index with no + /// entry at all: the walk then starts at the first record past the + /// superblock with no anchor needle and no append timestamp to compare + /// against, and a tail it cannot explain fails the load like any other + /// repair failure, since every later check skips an empty index. Nothing + /// is changed when the tail is anything but what an interrupted append + /// leaves, and the integrity check that follows then marks the volume + /// read-only exactly as before. + pub(crate) fn repair_unindexed_dat_tail(&self, io: &VolumeIo) -> Result<(), TailRepairError> { + let version = self.version(); + if version != VERSION_3 || !self.has_data_backend_io(io) { + return Ok(()); + } + let idx_path = self.file_name(".idx"); + if !Path::new(&idx_path).exists() { + return Ok(()); + } + let idx_size = fs::metadata(&idx_path).map(|m| m.len()).unwrap_or(0) as i64; + if idx_size % NEEDLE_MAP_ENTRY_SIZE as i64 != 0 { + // a partial .idx entry is left alone, as today + return Ok(()); + } + let file_size = self.current_dat_file_size_io(io)? as i64; + if file_size <= SUPER_BLOCK_SIZE as i64 || file_size > MAX_POSSIBLE_VOLUME_SIZE as i64 { + // past MaxPossibleVolumeSize the .idx offsets wrap: an entry's + // offset does not name its needle's position on its own + return Ok(()); + } + + // Go opens the index O_RDWR ahead of the check when the volume is + // writable, and a failure there fails the load, not just the repair. + let mut idx_file = OpenOptions::new() + .read(true) + .write(true) + .open(&idx_path) + .map_err(|e| TailRepairError::OpenIndex(VolumeError::Io(e)))?; + + let mut last_append_at_ns = 0u64; + let start = if idx_size == 0 { + // The index holds no row at all, so no entry can anchor the + // walk: it starts at the first record past the superblock. + // Mirrors Go's tailEntryPos < 0. + self.super_block().block_size() as i64 + } else { + let tail_entry_pos = + self.find_dat_tail_entry_offset(io, &mut idx_file, idx_size, version)?; + if tail_entry_pos < 0 { + // pre-allocated / all-zero index, no anchor to repair past + return Ok(()); + } + let mut buf = [0u8; NEEDLE_MAP_ENTRY_SIZE]; + idx_file.seek(SeekFrom::Start(tail_entry_pos as u64))?; + idx_file.read_exact(&mut buf)?; + let (key, offset, size) = idx_entry_from_bytes(&buf); + if offset.is_zero() { + return Ok(()); + } + let start = needle_disk_end(offset, size, version); + if start >= file_size { + // the .dat ends at the last needle (clean), or the index + // references bytes the .dat does not have: not a torn tail + return Ok(()); + } + + // Anchor: the needle at the highest indexed offset must be the one + // its entry describes, whole and with a valid CRC. + let on_disk_size = if size.is_deleted() { Size(0) } else { size }; + let anchor = self + .read_tail_needle(io, offset.to_actual_offset(), on_disk_size, version) + .map_err(|e| { + TailRepairError::NotTorn(format!( + "last indexed needle {key} at {}: {e}", + offset.to_actual_offset() + )) + })?; + if anchor.id != key { + return Err(TailRepairError::NotTorn(format!( + "needle at {} is id {:?}, its index entry says id {:?}", + offset.to_actual_offset(), + anchor.id, + key + ))); + } + last_append_at_ns = anchor.append_at_ns; + start + }; + if start >= file_size { + // no record past the superblock: nothing to index + return Ok(()); + } + + let (entries, cut_at) = + self.walk_unindexed_dat_tail(io, version, start, file_size, last_append_at_ns)?; + let reindexed = entries.len() / NEEDLE_MAP_ENTRY_SIZE; + if reindexed > 0 { + // The new .idx rows must not become durable before the .dat + // records they name: a power loss between the two would leave + // the .idx pointing past the end of the .dat. + self.flush_dat(io).map_err(|e| { + TailRepairError::Failed(VolumeError::Io(io::Error::new( + e.kind(), + format!("sync {} before re-indexing: {e}", self.file_name(".dat")), + ))) + })?; + Self::append_index_entries(&mut idx_file, &entries, idx_size)?; + } + let mut cut = 0i64; + if cut_at < file_size { + self.truncate_dat(io, cut_at as u64).map_err(|e| { + TailRepairError::Failed(VolumeError::Io(io::Error::new( + e.kind(), + format!("truncate {} to {cut_at}: {e}", self.file_name(".dat")), + ))) + })?; + self.flush_dat(io).map_err(|e| { + TailRepairError::Failed(VolumeError::Io(io::Error::new( + e.kind(), + format!( + "sync {} after truncating to {cut_at}: {e}", + self.file_name(".dat") + ), + ))) + })?; + cut = file_size - cut_at; + } + warn!( + "volume {}: repaired the unindexed tail of {}: re-indexed {} needle(s), cut {} bytes ({} -> {} bytes)", + self.id.0, + self.file_name(".dat"), + reindexed, + cut, + file_size, + file_size - cut + ); + Ok(()) + } + + /// Read and verify the whole record of a needle of the given on-disk + /// size at offset: header, body, CRC and timestamp. Mirrors Go's + /// readTailNeedle. + fn read_tail_needle( + &self, + io: &VolumeIo, + at: i64, + size: Size, + version: Version, + ) -> Result { + let blob = self + .read_needle_blob_unlocked(io, at, size) + .map_err(TailRepairError::Failed)?; + let mut n = Needle::default(); + n.read_bytes(&blob, at, size, version) + .map_err(|e| match e { + NeedleError::CrcMismatch { .. } + | NeedleError::IndexOutOfRange(_) + | NeedleError::TailTooShort => TailRepairError::Corrupt(e.to_string()), + other => TailRepairError::Failed(VolumeError::Io(io::Error::new( + io::ErrorKind::InvalidData, + other.to_string(), + ))), + })?; + if size.0 > 0 && n.data.is_empty() { + // A needle with no data is written with size 0, so a record with + // a body whose data size reads 0 is damaged. The CRC covers only + // the data, and read_bytes does not check it for empty data. + return Err(TailRepairError::Corrupt(format!( + "needle {:?} at {at} has size {} but no data", + n.id, size.0 + ))); + } + Ok(n) + } + + /// Walk the needles from start to file_size and return the .idx entries + /// of the complete ones, and the offset from which the rest is an + /// incomplete last record (file_size when there is none). Fails with + /// NotTorn when the tail is damaged in a way an interrupted append does + /// not explain. Mirrors Go's walkUnindexedDatTail. + fn walk_unindexed_dat_tail( + &self, + io: &VolumeIo, + version: Version, + start: i64, + file_size: i64, + last_append_at_ns: u64, + ) -> Result<(Vec, i64), TailRepairError> { + let mut entries = Vec::new(); + let mut last_append_at_ns = last_append_at_ns; + let mut at = start; + while at < file_size { + if file_size - at < NEEDLE_HEADER_SIZE as i64 { + return Ok((entries, at)); // a partial header + } + let mut header = [0u8; NEEDLE_HEADER_SIZE]; + self.read_exact_at_backend(io, &mut header, at as u64) + .map_err(|e| { + TailRepairError::Failed(VolumeError::Io(io::Error::new( + io::ErrorKind::InvalidData, + format!("read needle header at {at}: {e}"), + ))) + })?; + let (_, id, size) = Needle::parse_header(&header); + if id.is_empty() || size.0 < 0 { + return Err(TailRepairError::NotTorn(format!( + "record at {at} has id {id:?} size {}", + size.0 + ))); + } + let end = at + get_actual_size(size, version); + if end > file_size { + return Ok((entries, at)); // the last record runs past the end of the file + } + let n = match self.read_tail_needle(io, at, size, version) { + Ok(n) => n, + Err(e @ TailRepairError::Corrupt(_)) => { + if end == file_size { + return Ok((entries, at)); // the last record is damaged: it never completed + } + return Err(TailRepairError::NotTorn(format!( + "record {id:?} at {at} is damaged and {} bytes follow it: {e}", + file_size - end + ))); + } + Err(e) => return Err(e), // a read error says nothing about the bytes: leave them + }; + if n.append_at_ns <= last_append_at_ns { + return Err(TailRepairError::NotTorn(format!( + "record {:?} at {at} was appended at {}, not after {last_append_at_ns}", + n.id, n.append_at_ns + ))); + } + last_append_at_ns = n.append_at_ns; + // A size 0 record is a delete or an empty write; the two append + // the identical record, so it is indexed at size 0 the way the + // write path indexes an empty write. + let mut entry = [0u8; NEEDLE_MAP_ENTRY_SIZE]; + idx_entry_to_bytes(&mut entry, n.id, Offset::from_actual_offset(at), size); + entries.extend_from_slice(&entry); + at = end; + } + Ok((entries, file_size)) + } + + /// Write entries at the end (at) of the .idx and sync it. A failed write + /// can leave part of the entries behind, which would make the next load + /// reject the .idx as a partial entry; they are removed again. Mirrors + /// Go's appendIndexEntries. + fn append_index_entries( + idx_file: &mut File, + entries: &[u8], + at: i64, + ) -> Result<(), TailRepairError> { + let write = (|| -> io::Result<()> { + idx_file.seek(SeekFrom::Start(at as u64))?; + idx_file.write_all(entries) + })(); + if let Err(e) = write { + return Err(match idx_file.set_len(at as u64) { + Ok(()) => TailRepairError::Failed(VolumeError::Io(io::Error::new( + e.kind(), + format!("append {} bytes to index: {e}", entries.len()), + ))), + Err(te) => TailRepairError::Failed(VolumeError::Io(io::Error::new( + e.kind(), + format!( + "append {} bytes to index: {e}; truncate back to {at}: {te}", + entries.len() + ), + ))), + }); + } + idx_file.sync_all().map_err(|e| { + TailRepairError::Failed(VolumeError::Io(io::Error::new( + e.kind(), + format!("sync index: {e}"), + ))) + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::storage::needle_map::NeedleMapKind; + use std::collections::HashMap; + use crate::storage::volume::VolumeSpec; + use tempfile::TempDir; + + // A torn tail is what a crash between the two appends of a write leaves: + // the needle is in the .dat, its entry never reached the .idx. These + // tests write needles (and deletes) normally, drop the last .idx + // entries, cut or damage the .dat as a crash would, and reload the + // volume through the real load path — the same fixture Go's + // volume_checking_tail_repair_test.go builds. + + enum Op { + Write(u64), + Del(u64), + Empty(u64), // a write with no data appends a size 0 record like a delete + } + + struct TornFixture { + _tmp: TempDir, + dir: String, + starts: Vec, // .dat offset of each op's record + end: i64, // .dat size after every op + written: HashMap>, + } + + impl TornFixture { + fn dat_path(&self) -> String { + format!("{}/1.dat", self.dir) + } + fn idx_path(&self) -> String { + format!("{}/1.idx", self.dir) + } + fn dat_size(&self) -> i64 { + fs::metadata(self.dat_path()).unwrap().len() as i64 + } + fn idx_size(&self) -> i64 { + fs::metadata(self.idx_path()).unwrap().len() as i64 + } + /// An offset inside the body of op i's record: past its header, + /// short of its end. + fn mid_record(&self, i: usize) -> i64 { + let end = if i + 1 < self.starts.len() { + self.starts[i + 1] + } else { + self.end + }; + self.starts[i] + + NEEDLE_HEADER_SIZE as i64 + + (end - self.starts[i] - NEEDLE_HEADER_SIZE as i64) / 2 + } + fn truncate_dat(&self, size: i64) { + OpenOptions::new() + .write(true) + .open(self.dat_path()) + .unwrap() + .set_len(size as u64) + .unwrap(); + } + /// Damage the data of the record at the given offset so its CRC no + /// longer matches. + fn flip_data_byte(&self, at: i64) { + // Positional writes without Unix-only FileExt so the helper also + // builds on Windows; single-threaded, no reader can move the + // offset between seek and write. + let mut file = OpenOptions::new() + .read(true) + .write(true) + .open(self.dat_path()) + .unwrap(); + let pos = (at + NEEDLE_HEADER_SIZE as i64 + 4) as u64; // past the 4-byte DataSize field + let mut b = [0u8; 1]; + file.seek(SeekFrom::Start(pos)).unwrap(); + file.read_exact(&mut b).unwrap(); + b[0] ^= 0xff; + file.seek(SeekFrom::Start(pos)).unwrap(); + file.write_all(&b).unwrap(); + } + /// Zero the DataSize field and the first data byte (read as the + /// flags once DataSize is 0): the header still claims a body, the + /// body parses as an empty needle. + fn zero_data_size(&self, at: i64) { + let mut file = OpenOptions::new() + .write(true) + .open(self.dat_path()) + .unwrap(); + file.seek(SeekFrom::Start(at as u64 + NEEDLE_HEADER_SIZE as u64)) + .unwrap(); + file.write_all(&[0u8; 5]).unwrap(); + } + } + + fn open_volume(dir: &str) -> Volume { + Volume::new( + dir, + dir, + VolumeId(1), + NeedleMapKind::InMemory, + &VolumeSpec::default(), + ) + .unwrap() + } + + fn new_needle(id: u64) -> Needle { + let data = format!("needle-{id}-payload").into_bytes(); + Needle { + id: NeedleId(id), + cookie: Cookie(0x1234_0000 + id as u32), + data, + ..Needle::default() + } + } + + /// Run ops against a fresh volume (each op appends one .dat record and + /// one .idx entry), drop it, and remove the last `unindexed` entries. + fn build_fixture(ops: &[Op], unindexed: usize) -> TornFixture { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap().to_string(); + let mut f = TornFixture { + _tmp: tmp, + dir, + starts: Vec::new(), + end: 0, + written: HashMap::new(), + }; + { + let v = open_volume(&f.dir); + for op in ops { + f.starts.push(v.current_dat_file_size().unwrap() as i64); + match op { + Op::Write(id) => { + let mut n = new_needle(*id); + v.write_needle(&mut n, true, false).unwrap(); + f.written.insert(*id, n.data.clone()); + } + Op::Del(id) => { + v.delete_needle(&mut Needle { + id: NeedleId(*id), + ..Needle::default() + }) + .unwrap(); + } + Op::Empty(id) => { + let mut n = Needle { + id: NeedleId(*id), + cookie: Cookie(0x1234_0000 + *id as u32), + ..Needle::default() + }; + v.write_needle(&mut n, true, false).unwrap(); + f.written.insert(*id, n.data.clone()); + } + } + } + f.end = v.current_dat_file_size().unwrap() as i64; + } + if unindexed > 0 { + let idx_file = OpenOptions::new().write(true).open(f.idx_path()).unwrap(); + let idx_size = idx_file.metadata().unwrap().len(); + idx_file + .set_len(idx_size - unindexed as u64 * NEEDLE_MAP_ENTRY_SIZE as u64) + .unwrap(); + } + f + } + + fn assert_readable(v: &Volume, id: u64, want: &[u8]) { + let mut n = Needle { + id: NeedleId(id), + ..Needle::default() + }; + v.read_needle(&mut n).unwrap(); + assert_eq!(n.data, want, "needle {id} reads back different data"); + } + + fn assert_not_readable(v: &Volume, id: u64) { + let mut n = Needle { + id: NeedleId(id), + ..Needle::default() + }; + assert!( + v.read_needle(&mut n).is_err(), + "needle {id} reads back; it should be deleted or absent" + ); + } + + /// A recovered size 0 record indexes as a live empty needle: a delete + /// and a write with no data append the identical record, so the repair + /// cannot tell them apart and indexes the entry the write path would + /// have written. + fn assert_empty(v: &Volume, id: u64) { + let mut n = Needle { + id: NeedleId(id), + ..Needle::default() + }; + v.read_needle(&mut n) + .unwrap_or_else(|e| panic!("read needle {id}: {e}")); + assert!( + n.data.is_empty(), + "needle {id} reads back {} bytes, want an empty needle", + n.data.len() + ); + } + + fn writes(ids: &[u64]) -> Vec { + ids.iter().map(|id| Op::Write(*id)).collect() + } + + struct RepairCase { + name: &'static str, + ops: Vec, + unindexed: usize, + damage: Option, + want_dat_size: fn(&TornFixture) -> i64, + reindexed: usize, + readable: &'static [u64], + empty: &'static [u64], // readable as an empty live needle + absent: &'static [u64], + } + + struct ReadOnlyCase { + name: &'static str, + ops: Vec, + unindexed: usize, + damage: fn(&TornFixture), + } + + #[test] + fn volume_load_repairs_unindexed_dat_tail() { + let cases = vec![ + RepairCase { + name: "one complete unindexed needle is re-indexed", + ops: writes(&[1, 2, 3, 4]), + unindexed: 1, + damage: None, + want_dat_size: |f| f.end, + reindexed: 1, + readable: &[1, 2, 3, 4], + empty: &[], + absent: &[], + }, + RepairCase { + name: "two complete unindexed needles are re-indexed", + ops: writes(&[1, 2, 3, 4, 5]), + unindexed: 2, + damage: None, + want_dat_size: |f| f.end, + reindexed: 2, + readable: &[1, 2, 3, 4, 5], + empty: &[], + absent: &[], + }, + RepairCase { + name: "an unindexed delete is re-indexed as an empty live needle", + ops: { + let mut ops = writes(&[1, 2, 3]); + ops.push(Op::Del(2)); + ops + }, + unindexed: 1, + damage: None, + want_dat_size: |f| f.end, + reindexed: 1, + readable: &[1, 3], + empty: &[2], + absent: &[], + }, + RepairCase { + name: "an unindexed empty write of a live id stays an empty needle", + ops: { + let mut ops = writes(&[1, 2, 3]); + ops.push(Op::Empty(2)); + ops + }, + unindexed: 1, + damage: None, + want_dat_size: |f| f.end, + reindexed: 1, + readable: &[1, 3], + empty: &[2], + absent: &[], + }, + RepairCase { + name: "a partial header is cut", + ops: writes(&[1, 2, 3, 4]), + unindexed: 1, + damage: Some(|f| f.truncate_dat(f.starts[3] + 7)), + want_dat_size: |f| f.starts[3], + reindexed: 0, + readable: &[1, 2, 3], + empty: &[], + absent: &[4], + }, + RepairCase { + name: "a partial body is cut", + ops: writes(&[1, 2, 3, 4]), + unindexed: 1, + damage: Some(|f| f.truncate_dat(f.mid_record(3))), + want_dat_size: |f| f.starts[3], + reindexed: 0, + readable: &[1, 2, 3], + empty: &[], + absent: &[4], + }, + RepairCase { + name: "a complete needle is re-indexed and the partial one after it cut", + ops: writes(&[1, 2, 3, 4, 5]), + unindexed: 2, + damage: Some(|f| f.truncate_dat(f.mid_record(4))), + want_dat_size: |f| f.starts[4], + reindexed: 1, + readable: &[1, 2, 3, 4], + empty: &[], + absent: &[5], + }, + RepairCase { + name: "a last needle with a bad CRC is cut", + ops: writes(&[1, 2, 3, 4]), + unindexed: 1, + damage: Some(|f| f.flip_data_byte(f.starts[3])), + want_dat_size: |f| f.starts[3], + reindexed: 0, + readable: &[1, 2, 3], + empty: &[], + absent: &[4], + }, + RepairCase { + name: "a last needle whose data size field reads 0 is cut", + ops: writes(&[1, 2, 3, 4]), + unindexed: 1, + damage: Some(|f| f.zero_data_size(f.starts[3])), + want_dat_size: |f| f.starts[3], + reindexed: 0, + readable: &[1, 2, 3], + empty: &[], + absent: &[4], + }, + RepairCase { + name: "an unindexed empty write of a new id stays an empty needle", + ops: { + let mut ops = writes(&[1, 2]); + ops.push(Op::Empty(3)); + ops + }, + unindexed: 1, + damage: None, + want_dat_size: |f| f.end, + reindexed: 1, + readable: &[1, 2, 3], + empty: &[], + absent: &[], + }, + RepairCase { + name: "an unindexed empty write after a delete of the same id stays an empty needle", + ops: { + let mut ops = writes(&[1, 2]); + ops.push(Op::Del(2)); + ops.push(Op::Empty(2)); + ops + }, + unindexed: 1, + damage: None, + want_dat_size: |f| f.end, + reindexed: 1, + readable: &[1, 2], + empty: &[], + absent: &[], + }, + RepairCase { + name: "an unindexed write and delete of the same id are re-indexed in order", + ops: { + let mut ops = writes(&[1, 2, 3]); + ops.push(Op::Del(3)); + ops + }, + unindexed: 2, + damage: None, + want_dat_size: |f| f.end, + reindexed: 2, + readable: &[1, 2], + empty: &[3], + absent: &[], + }, + RepairCase { + name: "an unindexed empty write and its delete are re-indexed in order", + ops: { + let mut ops = writes(&[1, 2]); + ops.push(Op::Empty(3)); + ops.push(Op::Del(3)); + ops + }, + unindexed: 2, + damage: None, + want_dat_size: |f| f.end, + reindexed: 2, + readable: &[1, 2], + empty: &[3], + absent: &[], + }, + RepairCase { + name: "a partial needle after an indexed tombstone is cut", + ops: { + let mut ops = writes(&[1, 2, 3]); + ops.push(Op::Del(2)); + ops.push(Op::Write(4)); + ops + }, + unindexed: 1, + damage: Some(|f| f.truncate_dat(f.mid_record(4))), + want_dat_size: |f| f.starts[4], + reindexed: 0, + readable: &[1, 3], + empty: &[], + absent: &[2, 4], + }, + ]; + for tc in cases { + let f = build_fixture(&tc.ops, tc.unindexed); + if let Some(damage) = tc.damage { + damage(&f); + } + let idx_before = f.idx_size(); + + { + let v = open_volume(&f.dir); + assert!( + !v.is_read_only(), + "{}: volume with an unindexed .dat tail loaded read-only", + tc.name + ); + assert_eq!( + f.dat_size(), + (tc.want_dat_size)(&f), + "{}: .dat size after load", + tc.name + ); + assert_eq!( + f.idx_size(), + idx_before + tc.reindexed as i64 * NEEDLE_MAP_ENTRY_SIZE as i64, + "{}: .idx size after load ({} entries re-indexed)", + tc.name, + tc.reindexed + ); + for &id in tc.readable { + assert_readable(&v, id, &f.written[&id]); + } + for &id in tc.empty { + assert_empty(&v, id); + } + for &id in tc.absent { + assert_not_readable(&v, id); + } + let mut extra = new_needle(100); + v.write_needle(&mut extra, true, false) + .expect("write after repair"); + assert_readable(&v, 100, &extra.data); + } + + // the repaired volume reloads clean, and keeps what it indexed + let v = open_volume(&f.dir); + assert!( + !v.is_read_only(), + "{}: repaired volume loaded read-only on the next load", + tc.name + ); + for &id in tc.readable { + assert_readable(&v, id, &f.written[&id]); + } + assert_readable(&v, 100, b"needle-100-payload"); + } + } + + /// An empty index can hide writes a crash stranded in the .dat: the + /// writable index loader creates the .idx before the first needle + /// append, so a crash after that append leaves a volume whose data + /// nothing indexes. The same repair runs from the first record past + /// the superblock. + #[test] + fn volume_load_repairs_tail_past_empty_index() { + let cases = vec![ + RepairCase { + name: "the only write is re-indexed", + ops: writes(&[7]), + unindexed: 1, + damage: None, + want_dat_size: |f| f.end, + reindexed: 1, + readable: &[7], + empty: &[], + absent: &[], + }, + RepairCase { + name: "every write is re-indexed", + ops: writes(&[1, 2, 3]), + unindexed: 3, + damage: None, + want_dat_size: |f| f.end, + reindexed: 3, + readable: &[1, 2, 3], + empty: &[], + absent: &[], + }, + RepairCase { + name: "an unindexed delete is re-indexed as an empty live needle", + ops: { + let mut ops = writes(&[1, 2, 3]); + ops.push(Op::Del(2)); + ops + }, + unindexed: 4, + damage: None, + want_dat_size: |f| f.end, + reindexed: 4, + readable: &[1, 3], + empty: &[2], + absent: &[], + }, + RepairCase { + name: "a complete needle is re-indexed and the partial one after it cut", + ops: writes(&[1, 2]), + unindexed: 2, + damage: Some(|f| f.truncate_dat(f.mid_record(1))), + want_dat_size: |f| f.starts[1], + reindexed: 1, + readable: &[1], + empty: &[], + absent: &[2], + }, + RepairCase { + name: "a last needle with a bad CRC is cut", + ops: writes(&[1, 2]), + unindexed: 2, + damage: Some(|f| f.flip_data_byte(f.starts[1])), + want_dat_size: |f| f.starts[1], + reindexed: 1, + readable: &[1], + empty: &[], + absent: &[2], + }, + ]; + for tc in cases { + let f = build_fixture(&tc.ops, tc.unindexed); + if let Some(damage) = tc.damage { + damage(&f); + } + + { + let v = open_volume(&f.dir); + assert!( + !v.is_read_only(), + "{}: volume with an unindexed .dat tail loaded read-only", + tc.name + ); + assert_eq!( + f.dat_size(), + (tc.want_dat_size)(&f), + "{}: .dat size after load", + tc.name + ); + assert_eq!( + f.idx_size(), + tc.reindexed as i64 * NEEDLE_MAP_ENTRY_SIZE as i64, + "{}: .idx size after load ({} entries re-indexed)", + tc.name, + tc.reindexed + ); + for &id in tc.readable { + assert_readable(&v, id, &f.written[&id]); + } + for &id in tc.empty { + assert_empty(&v, id); + } + for &id in tc.absent { + assert_not_readable(&v, id); + } + } + + // the repaired volume reloads clean, and keeps what it indexed + let v = open_volume(&f.dir); + assert!( + !v.is_read_only(), + "{}: repaired volume loaded read-only on the next load", + tc.name + ); + for &id in tc.readable { + assert_readable(&v, id, &f.written[&id]); + } + for &id in tc.empty { + assert_empty(&v, id); + } + } + } + + /// A .dat an interrupted append does not explain, over an index with no + /// rows at all, loads the volume read-only: the checks that would + /// quarantine it otherwise all skip an empty index. + #[test] + fn volume_load_keeps_unexplained_tail_past_empty_index_read_only() { + let cases = vec![ + ReadOnlyCase { + name: "zero bytes where the next record would be (id 0)", + ops: writes(&[1, 2]), + unindexed: 2, + damage: |f| { + let mut file = OpenOptions::new().append(true).open(f.dat_path()).unwrap(); + file.write_all(&[0u8; 64]).unwrap(); + }, + }, + ReadOnlyCase { + name: "a damaged record more data follows", + ops: writes(&[1, 2]), + unindexed: 2, + damage: |f| f.flip_data_byte(f.starts[0]), + }, + ]; + for tc in cases { + let f = build_fixture(&tc.ops, tc.unindexed); + (tc.damage)(&f); + let (dat_before, idx_before) = (f.dat_size(), f.idx_size()); + assert_eq!(idx_before, 0, "fixture did not leave an empty .idx"); + + let v = open_volume(&f.dir); + assert!( + v.is_read_only(), + "{}: volume with an unexplained tail past an empty index loaded writable", + tc.name + ); + assert_eq!( + f.dat_size(), + dat_before, + "{}: .dat changed on load", + tc.name + ); + assert_eq!( + f.idx_size(), + idx_before, + "{}: .idx changed on load", + tc.name + ); + } + } + + /// Every tail an interrupted append does not explain is left exactly as + /// it is, and the volume loads read-only as before. + #[test] + fn volume_load_keeps_unexplained_tail_read_only() { + let cases = vec![ + ReadOnlyCase { + name: "the last indexed needle's id differs from its entry", + ops: writes(&[1, 2, 3, 4]), + unindexed: 1, + damage: |f| { + let mut file = OpenOptions::new().write(true).open(f.dat_path()).unwrap(); + file.seek(SeekFrom::Start( + (f.starts[2] + COOKIE_SIZE as i64 + NEEDLE_ID_SIZE as i64 - 1) as u64, + )) + .unwrap(); + file.write_all(&[0xff]).unwrap(); + }, + }, + ReadOnlyCase { + name: "the last indexed needle fails its CRC", + ops: writes(&[1, 2, 3, 4]), + unindexed: 1, + damage: |f| f.flip_data_byte(f.starts[2]), + }, + ReadOnlyCase { + name: "a zeroed data size field followed by more data", + ops: writes(&[1, 2, 3, 4, 5]), + unindexed: 2, + damage: |f| f.zero_data_size(f.starts[3]), + }, + ReadOnlyCase { + name: "a bad CRC followed by more data", + ops: writes(&[1, 2, 3, 4, 5]), + unindexed: 2, + damage: |f| f.flip_data_byte(f.starts[3]), + }, + ReadOnlyCase { + name: "zero bytes past the last needle (id 0)", + ops: writes(&[1, 2, 3]), + unindexed: 0, + damage: |f| { + let mut file = OpenOptions::new().append(true).open(f.dat_path()).unwrap(); + file.write_all(&[0u8; 64]).unwrap(); + }, + }, + ReadOnlyCase { + name: "the index ends past the data", + ops: writes(&[1, 2, 3]), + unindexed: 0, + damage: |f| f.truncate_dat(f.end - 10), + }, + ]; + for tc in cases { + let f = build_fixture(&tc.ops, tc.unindexed); + (tc.damage)(&f); + let (dat_before, idx_before) = (f.dat_size(), f.idx_size()); + + let v = open_volume(&f.dir); + assert!( + v.is_read_only(), + "{}: volume with an unexplained tail loaded writable", + tc.name + ); + assert_eq!( + f.dat_size(), + dat_before, + "{}: .dat changed on load", + tc.name + ); + assert_eq!( + f.idx_size(), + idx_before, + "{}: .idx changed on load", + tc.name + ); + } + } + + #[test] + fn volume_load_leaves_clean_volume_untouched() { + let f = build_fixture( + &{ + let mut ops = writes(&[1, 2, 3]); + ops.push(Op::Del(2)); + ops + }, + 0, + ); + let (dat_before, idx_before) = (f.dat_size(), f.idx_size()); + + let v = open_volume(&f.dir); + assert!(!v.is_read_only(), "clean volume loaded read-only"); + assert_eq!(f.dat_size(), dat_before, "clean .dat changed"); + assert_eq!(f.idx_size(), idx_before, "clean .idx changed"); + assert_readable(&v, 1, &f.written[&1]); + assert_readable(&v, 3, &f.written[&3]); + assert_not_readable(&v, 2); + } + + /// A volume persisted read-only is not repaired: the tail stays for + /// whoever marked it. + #[test] + fn volume_load_leaves_read_only_volume_tail_alone() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap().to_string(); + { + let v = open_volume(&dir); + for id in [1u64, 2, 99] { + let mut n = new_needle(id); + v.write_needle(&mut n, true, false).unwrap(); + } + v.set_read_only_persist(false, true).unwrap(); + } + let idx_path = format!("{dir}/1.idx"); + let idx_len = fs::metadata(&idx_path).unwrap().len(); + OpenOptions::new() + .write(true) + .open(&idx_path) + .unwrap() + .set_len(idx_len - NEEDLE_MAP_ENTRY_SIZE as u64) + .unwrap(); + let dat_path = format!("{dir}/1.dat"); + let (dat_before, idx_before) = ( + fs::metadata(&dat_path).unwrap().len(), + fs::metadata(&idx_path).unwrap().len(), + ); + + let v = open_volume(&dir); + assert!( + v.is_read_only(), + "volume persisted read-only loaded writable" + ); + assert_eq!(fs::metadata(&dat_path).unwrap().len(), dat_before); + assert_eq!(fs::metadata(&idx_path).unwrap().len(), idx_before); + } +} diff --git a/weed/storage/volume_checking.go b/weed/storage/volume_checking.go index 1bb1157346..f65d3254e0 100644 --- a/weed/storage/volume_checking.go +++ b/weed/storage/volume_checking.go @@ -145,7 +145,25 @@ func CheckVolumeDataIntegrity(v *Volume, indexFile *os.File) (lastAppendAtNs uin return 0, fmt.Errorf("verifyIndexFileIntegrity %s failed: %v", indexFile.Name(), err) } if indexSize == 0 { - return 0, nil + if v.noWriteOrDelete || v.noWriteCanDelete { + return 0, nil + } + // An empty index can hide writes a crash stranded in the .dat: the + // writable index loader creates the .idx before the first needle + // append, so a crash after that append leaves a volume whose data + // nothing indexes — and every check below skips an empty index. The + // same repair runs from the first record past the superblock; a tail + // it cannot explain fails the check, since an unaccounted-for .dat + // must not load writable either. + if _, _, err = repairUnindexedDatTail(v, indexFile, 0, -1); err != nil { + return 0, fmt.Errorf("CheckVolumeDataIntegrity %s: repair unindexed .dat tail: %v", indexFile.Name(), err) + } + if indexSize, err = verifyIndexFileIntegrity(indexFile); err != nil { + return 0, fmt.Errorf("verifyIndexFileIntegrity %s after re-indexing failed: %v", indexFile.Name(), err) + } + if indexSize == 0 { + return 0, nil + } } // The deeper-than-tail structural check (every (offset + actual size) // fits inside .dat) lives in volume.load(): it reads @@ -258,14 +276,25 @@ var errTailNotTorn = errors.New("not a torn tail") // A write appends the needle to the .dat, then its entry to the .idx, then // answers the client. A process killed between the first two steps leaves a // needle, or part of one, past the end of the needle at the highest indexed -// offset. After a power loss the order on disk is whatever the page cache +// offset; a crash after the very first append leaves it past an index with +// no entry at all, which tailEntryPos < 0 marks: the walk then starts at the +// first record past the superblock with no anchor needle and no append +// timestamp to compare against. After a power loss the order on disk is +// whatever the page cache // flushed: a complete needle can survive whose .idx entry did not, although // the client was answered. So a complete needle is never dropped: every // complete needle in the tail (its header, body, checksum, timestamp and // padding inside the file, its CRC valid, its append timestamp later than the -// needle before it) gets the .idx entry the volume server would have written -// (a size 0 record as resolveEmptyTailRecords reads it), in .dat order, and -// the .idx is synced. Only what follows the last complete +// needle before it) gets the .idx entry the volume server would have written, +// in .dat order; the .dat is synced first so the rows never become durable +// ahead of the records they name, and the .idx is synced after. A size 0 +// record is indexed at size 0, +// the same live entry the write path publishes for a write with no data. A +// delete appends the identical record, so a recovered size 0 record can also +// be a delete whose key then reads back as an empty object rather than +// not-found -- the two cannot be told apart, and resurrecting an empty +// placeholder beats a wrong tombstone, which drops an acknowledged write's +// object entirely. Only what follows the last complete // needle and is itself incomplete is cut: fewer bytes than a header, a record // running past the end of the file, or a last record whose bytes are damaged // (its CRC fails or its body does not parse); the .dat is synced after the @@ -293,18 +322,45 @@ func repairUnindexedDatTail(v *Volume, indexFile *os.File, indexSize int64, tail // does not name its needle's position on its own return 0, 0, nil } - entry, err := readIndexEntryAtOffset(indexFile, tailEntryPos) - if err != nil { - return 0, 0, fmt.Errorf("read index entry at %d: %v", tailEntryPos, err) + var start int64 + var lastAppendAtNs uint64 + if tailEntryPos < 0 { + // The index holds no row at all, so no entry can anchor the walk: + // it starts at the first record past the superblock. + start = int64(v.SuperBlock.BlockSize()) + } else { + entry, err := readIndexEntryAtOffset(indexFile, tailEntryPos) + if err != nil { + return 0, 0, fmt.Errorf("read index entry at %d: %v", tailEntryPos, err) + } + key, offset, size := idx.IdxFileEntry(entry) + if offset.IsZero() { + return 0, 0, nil + } + start = needleDiskEnd(offset, size, version) + if start >= datSize { + // the .dat ends at the last needle (clean), or the index references + // bytes the .dat does not have: not a torn tail, not ours to fix + return 0, 0, nil + } + + // Anchor: the needle at the highest indexed offset must be the one its + // entry describes, whole and with a valid CRC. + onDiskSize := size + if size.IsDeleted() { + onDiskSize = 0 + } + anchor, err := readTailNeedle(v.DataBackend, version, offset.ToActualOffset(), onDiskSize) + if err != nil { + return 0, 0, fmt.Errorf("last indexed needle %v at %d: %v: %w", key, offset.ToActualOffset(), err, errTailNotTorn) + } + if anchor.Id != key { + return 0, 0, fmt.Errorf("needle at %d is id %v, its index entry says id %v: %w", offset.ToActualOffset(), anchor.Id, key, errTailNotTorn) + } + lastAppendAtNs = anchor.AppendAtNs } - key, offset, size := idx.IdxFileEntry(entry) - if offset.IsZero() { - return 0, 0, nil - } - start := needleDiskEnd(offset, size, version) if start >= datSize { - // the .dat ends at the last needle (clean), or the index references - // bytes the .dat does not have: not a torn tail, not ours to fix + // no record past the superblock: nothing to index return 0, 0, nil } // The backend rounds its size up to the needle alignment; the walk needs @@ -314,28 +370,17 @@ func repairUnindexedDatTail(v *Volume, indexFile *os.File, indexSize int64, tail fileSize = st.Size() } - // Anchor: the needle at the highest indexed offset must be the one its - // entry describes, whole and with a valid CRC. - onDiskSize := size - if size.IsDeleted() { - onDiskSize = 0 - } - anchor, err := readTailNeedle(v.DataBackend, version, offset.ToActualOffset(), onDiskSize) - if err != nil { - return 0, 0, fmt.Errorf("last indexed needle %v at %d: %v: %w", key, offset.ToActualOffset(), err, errTailNotTorn) - } - if anchor.Id != key { - return 0, 0, fmt.Errorf("needle at %d is id %v, its index entry says id %v: %w", offset.ToActualOffset(), anchor.Id, key, errTailNotTorn) - } - - entries, cutAt, err := walkUnindexedDatTail(v.DataBackend, version, start, fileSize, anchor.AppendAtNs) + entries, cutAt, err := walkUnindexedDatTail(v.DataBackend, version, start, fileSize, lastAppendAtNs) if err != nil { return 0, 0, err } reindexed := len(entries) / types.NeedleMapEntrySize if reindexed > 0 { - if err := resolveEmptyTailRecords(indexFile, entries); err != nil { - return 0, 0, err + // The new .idx rows must not become durable before the .dat records + // they name: a power loss between the two would leave the .idx + // pointing past the end of the .dat. + if err := v.DataBackend.Sync(); err != nil { + return 0, 0, fmt.Errorf("sync %s before re-indexing: %v", v.FileName(".dat"), err) } if err := appendIndexEntries(indexFile, entries, indexSize); err != nil { return 0, 0, err @@ -400,48 +445,6 @@ func readTailNeedle(datFile backend.BackendStorageFile, version needle.Version, return n, nil } -// resolveEmptyTailRecords turns the size 0 entries among the re-indexed ones -// into what the volume server indexed them as. A delete and a write with no -// data both append a size 0 record; a delete is indexed as a tombstone, an -// empty write with size 0. A delete is only appended for a needle that is -// live, so a size 0 record of a needle that is not live (absent or deleted in -// the .idx and in the tail before it) is an empty write and keeps size 0. One -// of a live needle is indexed as a tombstone, as weed fix and the backup's -// index regeneration read a size 0 record. -func resolveEmptyTailRecords(indexFile io.ReaderAt, entries []byte) error { - live := make(map[types.NeedleId]bool) - for i := 0; i < len(entries); i += types.NeedleMapEntrySize { - if key, _, size := idx.IdxFileEntry(entries[i : i+types.NeedleMapEntrySize]); size == 0 { - live[key] = false - } - } - if len(live) == 0 { - return nil - } - if err := idx.WalkIndexFile(indexFile, 0, func(key types.NeedleId, offset types.Offset, size types.Size) error { - if _, ok := live[key]; ok { - live[key] = !offset.IsZero() && !size.IsDeleted() - } - return nil - }); err != nil { - return fmt.Errorf("scan index: %v", err) - } - for i := 0; i < len(entries); i += types.NeedleMapEntrySize { - key, offset, size := idx.IdxFileEntry(entries[i : i+types.NeedleMapEntrySize]) - wasLive, tracked := live[key] - if !tracked { - continue - } - if size == 0 && wasLive { - copy(entries[i:], needle_map.ToBytes(key, offset, types.TombstoneFileSize)) - live[key] = false - continue - } - live[key] = true - } - return nil -} - // walkUnindexedDatTail walks the needles from start to fileSize and returns // the .idx entries of the complete ones, and the offset from which the rest is // an incomplete last record (fileSize when there is none). It fails with @@ -480,8 +483,9 @@ func walkUnindexedDatTail(datFile backend.BackendStorageFile, version needle.Ver return nil, 0, fmt.Errorf("record %v at %d was appended at %d, not after %d: %w", n.Id, at, n.AppendAtNs, lastAppendAtNs, errTailNotTorn) } lastAppendAtNs = n.AppendAtNs - // a size 0 record is a delete or an empty write: resolveEmptyTailRecords - // tells them apart + // A size 0 record is a delete or an empty write; the two append the + // identical record, so it is indexed at size 0 the way the write path + // indexes an empty write. entries = append(entries, needle_map.ToBytes(n.Id, types.ToOffset(at), header.Size)...) at = end } diff --git a/weed/storage/volume_checking_tail_repair_test.go b/weed/storage/volume_checking_tail_repair_test.go index 19a5f6a4ea..cbf3e9a40f 100644 --- a/weed/storage/volume_checking_tail_repair_test.go +++ b/weed/storage/volume_checking_tail_repair_test.go @@ -156,6 +156,20 @@ func assertNotReadable(t *testing.T, v *Volume, id uint64) { } } +// A recovered size 0 record indexes as a live empty needle: a delete and a +// write with no data append the identical record, so the repair cannot tell +// them apart and indexes the entry the write path would have written. +func assertEmpty(t *testing.T, v *Volume, id uint64) { + t.Helper() + n := &needle.Needle{Id: types.Uint64ToNeedleId(id)} + if _, err := v.readNeedle(n, nil, nil); err != nil { + t.Fatalf("read needle %d: %v", id, err) + } + if len(n.Data) != 0 { + t.Fatalf("needle %d reads back %d bytes, want an empty needle", id, len(n.Data)) + } +} + func writes(ids ...uint64) []tornOp { ops := make([]tornOp, len(ids)) for i, id := range ids { @@ -173,6 +187,7 @@ func TestVolumeLoadRepairsUnindexedDatTail(t *testing.T) { wantSize func(f *tornFixture) int64 // .dat size after the repair reindexed int readable []uint64 + empty []uint64 // readable as an empty live needle absent []uint64 }{ {name: "one complete unindexed needle is re-indexed", @@ -183,10 +198,14 @@ func TestVolumeLoadRepairsUnindexedDatTail(t *testing.T) { ops: writes(1, 2, 3, 4, 5), unindexed: 2, wantSize: func(f *tornFixture) int64 { return f.end }, reindexed: 2, readable: []uint64{1, 2, 3, 4, 5}}, - {name: "an unindexed delete is re-indexed as a tombstone", + {name: "an unindexed delete is re-indexed as an empty live needle", ops: append(writes(1, 2, 3), tornOp{id: 2, del: true}), unindexed: 1, wantSize: func(f *tornFixture) int64 { return f.end }, reindexed: 1, - readable: []uint64{1, 3}, absent: []uint64{2}}, + readable: []uint64{1, 3}, empty: []uint64{2}}, + {name: "an unindexed empty write of a live id stays an empty needle", + ops: append(writes(1, 2, 3), tornOp{id: 2, empty: true}), unindexed: 1, + wantSize: func(f *tornFixture) int64 { return f.end }, reindexed: 1, + readable: []uint64{1, 3}, empty: []uint64{2}}, {name: "a partial header is cut", ops: writes(1, 2, 3, 4), unindexed: 1, damage: func(t *testing.T, f *tornFixture) { f.truncateDat(t, f.starts[3]+7) }, @@ -223,11 +242,11 @@ func TestVolumeLoadRepairsUnindexedDatTail(t *testing.T) { {name: "an unindexed write and delete of the same id are re-indexed in order", ops: append(writes(1, 2, 3), tornOp{id: 3, del: true}), unindexed: 2, wantSize: func(f *tornFixture) int64 { return f.end }, reindexed: 2, - readable: []uint64{1, 2}, absent: []uint64{3}}, + readable: []uint64{1, 2}, empty: []uint64{3}}, {name: "an unindexed empty write and its delete are re-indexed in order", ops: append(writes(1, 2), tornOp{id: 3, empty: true}, tornOp{id: 3, del: true}), unindexed: 2, wantSize: func(f *tornFixture) int64 { return f.end }, reindexed: 2, - readable: []uint64{1, 2}, absent: []uint64{3}}, + readable: []uint64{1, 2}, empty: []uint64{3}}, {name: "a partial needle after an indexed tombstone is cut", ops: append(append(writes(1, 2, 3), tornOp{id: 2, del: true}), tornOp{id: 4}), unindexed: 1, damage: func(t *testing.T, f *tornFixture) { f.truncateDat(t, f.midRecord(4)) }, @@ -256,6 +275,9 @@ func TestVolumeLoadRepairsUnindexedDatTail(t *testing.T) { for _, id := range tc.readable { assertReadable(t, v, f.written[id]) } + for _, id := range tc.empty { + assertEmpty(t, v, id) + } for _, id := range tc.absent { assertNotReadable(t, v, id) } @@ -280,6 +302,132 @@ func TestVolumeLoadRepairsUnindexedDatTail(t *testing.T) { } } +// An empty index can hide writes a crash stranded in the .dat: the writable +// index loader creates the .idx before the first needle append, so a crash +// after that append leaves a volume whose data nothing indexes. The same +// repair runs from the first record past the superblock. +func TestVolumeLoadRepairsTailPastEmptyIndex(t *testing.T) { + cases := []struct { + name string + ops []tornOp + damage func(t *testing.T, f *tornFixture) + wantSize func(f *tornFixture) int64 // .dat size after the repair + reindexed int + readable []uint64 + empty []uint64 + absent []uint64 + }{ + {name: "the only write is re-indexed", + ops: writes(7), + wantSize: func(f *tornFixture) int64 { return f.end }, reindexed: 1, + readable: []uint64{7}}, + {name: "every write is re-indexed", + ops: writes(1, 2, 3), + wantSize: func(f *tornFixture) int64 { return f.end }, reindexed: 3, + readable: []uint64{1, 2, 3}}, + {name: "an unindexed delete is re-indexed as an empty live needle", + ops: append(writes(1, 2, 3), tornOp{id: 2, del: true}), + wantSize: func(f *tornFixture) int64 { return f.end }, reindexed: 4, + readable: []uint64{1, 3}, empty: []uint64{2}}, + {name: "a complete needle is re-indexed and the partial one after it cut", + ops: writes(1, 2), + damage: func(t *testing.T, f *tornFixture) { f.truncateDat(t, f.midRecord(1)) }, + wantSize: func(f *tornFixture) int64 { return f.starts[1] }, reindexed: 1, + readable: []uint64{1}, absent: []uint64{2}}, + {name: "a last needle with a bad CRC is cut", + ops: writes(1, 2), + damage: func(t *testing.T, f *tornFixture) { f.flipDataByte(t, f.starts[1]) }, + wantSize: func(f *tornFixture) int64 { return f.starts[1] }, reindexed: 1, + readable: []uint64{1}, absent: []uint64{2}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + f := buildTornFixture(t, tc.ops, len(tc.ops)) + if tc.damage != nil { + tc.damage(t, f) + } + + v := newTornTailTestVolume(t, f.dir) + defer v.Close() + if v.noWriteOrDelete || v.noWriteCanDelete { + t.Fatal("volume with an unindexed .dat tail loaded read-only") + } + if got, want := fileSize(t, f.datName), tc.wantSize(f); got != want { + t.Fatalf(".dat is %d bytes after load, want %d", got, want) + } + if got, want := fileSize(t, f.idxName), int64(tc.reindexed)*types.NeedleMapEntrySize; got != want { + t.Fatalf(".idx is %d bytes after load, want %d (%d entries re-indexed)", got, want, tc.reindexed) + } + for _, id := range tc.readable { + assertReadable(t, v, f.written[id]) + } + for _, id := range tc.empty { + assertEmpty(t, v, id) + } + for _, id := range tc.absent { + assertNotReadable(t, v, id) + } + v.Close() + + // the repaired volume reloads clean, and keeps what it indexed + v2 := newTornTailTestVolume(t, f.dir) + defer v2.Close() + if v2.noWriteOrDelete { + t.Fatal("repaired volume loaded read-only on the next load") + } + for _, id := range tc.readable { + assertReadable(t, v2, f.written[id]) + } + for _, id := range tc.empty { + assertEmpty(t, v2, id) + } + }) + } +} + +// An .dat an interrupted append does not explain, over an index with no rows +// at all, loads the volume read-only: the checks that would quarantine it +// otherwise all skip an empty index. +func TestVolumeLoadKeepsUnexplainedTailPastEmptyIndexReadOnly(t *testing.T) { + cases := []struct { + name string + damage func(t *testing.T, f *tornFixture) + }{ + {name: "zero bytes where the first record would be (id 0)", + damage: func(t *testing.T, f *tornFixture) { + file, err := os.OpenFile(f.datName, os.O_RDWR|os.O_APPEND, 0644) + if err != nil { + t.Fatal(err) + } + defer file.Close() + if _, err := file.Write(make([]byte, 64)); err != nil { + t.Fatal(err) + } + }}, + {name: "a damaged record more data follows", + damage: func(t *testing.T, f *tornFixture) { f.flipDataByte(t, f.starts[0]) }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + f := buildTornFixture(t, writes(1, 2), 2) + tc.damage(t, f) + datBefore, idxBefore := fileSize(t, f.datName), fileSize(t, f.idxName) + + v := newTornTailTestVolume(t, f.dir) + defer v.Close() + if !v.noWriteOrDelete { + t.Fatal("volume with an unexplained tail past an empty index loaded writable") + } + if got := fileSize(t, f.datName); got != datBefore { + t.Fatalf(".dat changed from %d to %d bytes", datBefore, got) + } + if got := fileSize(t, f.idxName); got != idxBefore { + t.Fatalf(".idx changed from %d to %d bytes", idxBefore, got) + } + }) + } +} + // Every tail an interrupted append does not explain is left exactly as it is, // and the volume loads read-only as before. func TestVolumeLoadKeepsUnexplainedTailReadOnly(t *testing.T) {