diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 051b47119..bacd83185 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -2426,18 +2426,7 @@ impl VolumeServer for VolumeGrpcService { } let version = vol.version().0 as u32; let dat_size = vol.dat_file_size().unwrap_or(0) as i64; - let expire_at_sec = { - let ttl_seconds = vol.super_block.ttl.to_seconds(); - if ttl_seconds > 0 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs() - + ttl_seconds - } else { - 0 - } - }; + let expire_at_sec = vol.expire_at_sec(); ( store.locations[loc_idx].directory.clone(), store.locations[loc_idx].idx_directory.clone(), diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index d7a2e1667..d5b1fef50 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/seaweed-volume/src/server/heartbeat.rs @@ -961,7 +961,10 @@ fn build_heartbeat_with_ec_status( version: vol.super_block.version.0 as u32, ttl: vol.super_block.ttl.to_u32(), compact_revision: vol.super_block.compaction_revision as u32, - modified_at_second: vol.last_modified_ts() as i64, + // The .dat mtime, as Go reports: the shell's quiet-period + // gates read this as "last touched", which a delete has to + // count towards even though the TTL clock ignores it. + modified_at_second: vol.dat_file_mod_time() as i64, disk_type: loc.disk_type.to_string(), disk_id: disk_id as u32, remote_storage_name, diff --git a/seaweed-volume/src/storage/idx/mod.rs b/seaweed-volume/src/storage/idx/mod.rs index 15392cb39..2cae9bec6 100644 --- a/seaweed-volume/src/storage/idx/mod.rs +++ b/seaweed-volume/src/storage/idx/mod.rs @@ -6,7 +6,7 @@ use crate::storage::needle::needle::get_actual_size; use crate::storage::types::*; use std::io::{self, Read, Seek, SeekFrom}; -const ROWS_TO_READ: usize = 1024; +pub(crate) const ROWS_TO_READ: usize = 1024; /// Walk all entries in an .idx file, calling `f` for each. /// Mirrors Go's `WalkIndexFile()`. diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 943307cea..4e2581958 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -19,7 +19,6 @@ use std::time::{SystemTime, UNIX_EPOCH}; use tracing::{error, info, warn}; -#[cfg(test)] use crate::storage::idx; use crate::storage::needle::needle::{self, get_actual_size, Needle, NeedleError}; use crate::storage::needle_map::sorted_file::SortedFileNeedleMap; @@ -855,6 +854,7 @@ impl Volume { "volumeDataIntegrityChecking failed" ); } + self.recover_last_modified_ts(); // Structural check: no .idx entry may reference bytes past the // end of .dat. The needle map's load walk above already @@ -2069,6 +2069,26 @@ impl Volume { (ttl_minutes as u64) < lived_minutes } + /// When this volume's data becomes garbage, counted from its last write. + /// Counting from the current time instead let every .vif rewrite — a + /// read-only mark, a tier upload, an EC encode — hand an already expiring + /// volume another full TTL. Zero when the volume has no TTL. Mirrors Go's + /// ExpireAtSec. + pub fn expire_at_sec(&self) -> u64 { + let ttl_seconds = self.super_block.ttl.to_seconds(); + if ttl_seconds == 0 { + return 0; + } + let mut last_write_sec = self.last_modified_ts_seconds; + if last_write_sec == 0 { + last_write_sec = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + } + last_write_sec + ttl_seconds + } + pub fn is_expired_long_enough(&self, max_delay_minutes: u32) -> bool { let ttl_minutes = self.super_block.ttl.minutes(); if ttl_minutes == 0 { @@ -2198,6 +2218,128 @@ impl Volume { } } + /// Bounds the work a vacuumed volume's recovery does, where key order makes + /// every write a candidate for the newest one. A volume with more live + /// needles than this keeps the .dat mtime: reading a subset could recover a + /// timestamp older than the newest write and expire data still inside its + /// TTL, so a scan that will not fit declines instead of guessing. + const VACUUMED_LAST_WRITE_SCAN_ENTRIES: usize = 1 << 16; + + /// Point the TTL clock at the newest write recorded in the volume, replacing + /// the .dat mtime the loader starts from. A delete appends a tombstone and + /// vacuum rewrites the .dat wholesale, so the mtime moves without any write + /// ever landing: every restart of a volume taking delete traffic re-armed + /// is_expired() for another full TTL and the volume was never reclaimed. + /// Mirrors Go's recoverLastModifiedTs. + fn recover_last_modified_ts(&mut self) { + if self.super_block.ttl.minutes() == 0 { + return; + } + match self.find_last_write_append_at_ns() { + Ok(0) => {} + Ok(append_at_ns) => self.last_modified_ts_seconds = append_at_ns / 1_000_000_000, + Err(e) => warn!( + volume_id = self.id.0, + error = %e, + "recover the last write from the index" + ), + } + } + + /// Scan the .idx backwards for the newest write — an entry that is not a + /// deletion tombstone — and return that needle's append timestamp. The .idx + /// and the .dat share an order, so an append-ordered volume answers with the + /// first write the scan reaches. Vacuum rewrites both in key order, which + /// tracks write order only because the master issues keys increasing: an + /// overwrite keeps its original, lower key, so a vacuumed volume has to take + /// the maximum over every write it indexes. Returns 0 when the .idx holds + /// nothing but tombstones, when a vacuumed volume holds more needles than + /// the scan budget, or for a volume older than version 3, whose needles + /// carry no append timestamp. Mirrors Go's findLastWriteAppendAtNs. + fn find_last_write_append_at_ns(&self) -> Result { + let version = self.version(); + if version != VERSION_3 { + return Ok(0); + } + let idx_path = self.file_name(".idx"); + let idx_size = fs::metadata(&idx_path).map(|m| m.len()).unwrap_or(0) as i64; + if idx_size == 0 || idx_size % NEEDLE_MAP_ENTRY_SIZE as i64 != 0 { + return Ok(0); + } + let scan_every_write = self.super_block.compaction_revision > 0; + let mut entry_budget = Self::VACUUMED_LAST_WRITE_SCAN_ENTRIES; + let mut last_write_append_at_ns = 0u64; + let mut idx_file = File::open(&idx_path)?; + let mut block = vec![0u8; NEEDLE_MAP_ENTRY_SIZE * idx::ROWS_TO_READ]; + let mut end = idx_size; + while end > 0 { + let start = (end - block.len() as i64).max(0); + let entries = &mut block[..(end - start) as usize]; + idx_file.seek(SeekFrom::Start(start as u64))?; + idx_file.read_exact(entries)?; + for entry in entries.chunks_exact(NEEDLE_MAP_ENTRY_SIZE).rev() { + let (key, offset, size) = idx_entry_from_bytes(entry); + if offset.is_zero() || size.is_deleted() { + continue; + } + let Some(needle_offset) = + self.find_needle_offset(offset.to_actual_offset(), key, size) + else { + continue; + }; + last_write_append_at_ns = last_write_append_at_ns + .max(self.read_needle_append_at_ns(needle_offset, size)?); + if !scan_every_write { + return Ok(last_write_append_at_ns); + } + entry_budget -= 1; + if entry_budget == 0 { + warn!( + volume_id = self.id.0, + budget = Self::VACUUMED_LAST_WRITE_SCAN_ENTRIES, + "too many needles to scan for the last write, keeping the .dat mtime" + ); + return Ok(0); + } + } + end = start; + } + Ok(last_write_append_at_ns) + } + + /// The .dat offset holding the needle an .idx entry describes, or None when + /// no needle there matches it. A .dat past MAX_POSSIBLE_VOLUME_SIZE wraps the + /// offsets in its .idx, so the needle can sit one volume size further in; + /// verify_needle_integrity retries the same way. Mirrors Go's + /// findNeedleOffset. + fn find_needle_offset(&self, actual_offset: i64, key: NeedleId, size: Size) -> Option { + for at in [ + actual_offset, + actual_offset + MAX_POSSIBLE_VOLUME_SIZE as i64, + ] { + let mut header = [0u8; NEEDLE_HEADER_SIZE]; + if self.read_exact_at_backend(&mut header, at as u64).is_err() { + continue; + } + let (_, needle_id, needle_size) = Needle::parse_header(&header); + if needle_id == key && needle_size == size { + return Some(at); + } + } + None + } + + /// Read the append timestamp a version 3 needle carries past its checksum. + fn read_needle_append_at_ns(&self, actual_offset: i64, size: Size) -> Result { + let ts_offset = actual_offset as u64 + + NEEDLE_HEADER_SIZE as u64 + + size.0 as u64 + + NEEDLE_CHECKSUM_SIZE as u64; + let mut ts = [0u8; TIMESTAMP_SIZE]; + self.read_exact_at_backend(&mut ts, ts_offset)?; + Ok(u64::from_be_bytes(ts)) + } + /// .idx file position of the entry whose needle is physically last in the /// .dat (highest offset). The common case — an append-ordered .idx — is /// resolved in O(1): the last entry's on-disk end equals the .dat size. A @@ -2297,10 +2439,7 @@ impl Volume { return Ok(()); } - let ts_offset = checked_offset as u64 + NEEDLE_HEADER_SIZE as u64 + size.0 as u64 + 4; // skip checksum - let mut ts_buf = [0u8; 8]; - self.read_exact_at_backend(&mut ts_buf, ts_offset)?; - let ts = u64::from_be_bytes(ts_buf); + let ts = self.read_needle_append_at_ns(checked_offset, size)?; if ts > 0 { self.last_append_at_ns = ts; } @@ -2835,13 +2974,9 @@ impl Volume { let mut vif = VifVolumeInfo::from_pb(&self.volume_info); // Match Go's SaveVolumeInfo: compute ExpireAtSec from TTL - let ttl_seconds = self.super_block.ttl.to_seconds(); - if ttl_seconds > 0 { - let now = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_secs(); - vif.expire_at_sec = now + ttl_seconds; + let expire_at_sec = self.expire_at_sec(); + if expire_at_sec > 0 { + vif.expire_at_sec = expire_at_sec; } let content = serde_json::to_string_pretty(&vif) @@ -2859,13 +2994,9 @@ impl Volume { self.volume_info.read_only_can_delete = marked_can_delete; // Compute ExpireAtSec from TTL (matches Go's SaveVolumeInfo) - let ttl_seconds = self.super_block.ttl.to_seconds(); - if ttl_seconds > 0 { - let now = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_secs(); - self.volume_info.expire_at_sec = now + ttl_seconds; + let expire_at_sec = self.expire_at_sec(); + if expire_at_sec > 0 { + self.volume_info.expire_at_sec = expire_at_sec; } let vif = VifVolumeInfo::from_pb(&self.volume_info); @@ -4775,6 +4906,165 @@ mod tests { ); } + // Reproduce the delete-traffic TTL bug: deletes append a tombstone to the + // .dat, which moves the file's mtime, and the loader read the TTL clock + // back from that mtime. A volume taking delete traffic therefore had + // is_expired() re-armed for another full TTL on every restart and was + // never reclaimed. + #[test] + fn test_ttl_clock_survives_deletes() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let ttl = crate::storage::needle::ttl::TTL::read("5m").unwrap(); + let last_write_ns = (SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs() + - 2 * 60 * 60) + * 1_000_000_000; + + let content_size = { + let mut v = make_ttl_volume(dir, ttl); + let mut written = Vec::new(); + for i in 1..=3u64 { + let data = format!("data {}", i); + let mut n = Needle { + id: NeedleId(i), + cookie: Cookie(i as u32), + data: data.as_bytes().to_vec(), + data_size: data.len() as u32, + ..Needle::default() + }; + let (offset, _, _) = v.write_needle(&mut n, true, false).unwrap(); + written.push((offset, n.size)); + } + // More than one tombstone: the scan has to walk back over the whole + // run of them to reach a write. + for i in [2u64, 3] { + v.delete_needle(&mut Needle { + id: NeedleId(i), + cookie: Cookie(i as u32), + ..Needle::default() + }) + .unwrap(); + } + v.sync_to_disk().unwrap(); + // Backdate the writes on disk so the last one sits well outside the + // TTL, while the tombstones leave the .dat mtime at now. + for (offset, size) in written { + backdate_append_at_ns(&v.dat_path(), offset, size, last_write_ns); + } + v.content_size() + }; + + let v = make_ttl_volume(dir, ttl); + assert_eq!(v.last_modified_ts(), last_write_ns / 1_000_000_000); + assert!( + v.is_expired(content_size, 1024 * 1024), + "a TTL volume whose last write is 2h old must be expired after a reload" + ); + } + + // The one layout where a .dat's order does not track its write order: + // vacuum rewrites it by key, and an overwrite keeps its original, lower key. + // Reading the position rather than the timestamps would recover the + // highest-key needle's older write time and expire the volume before the + // overwrite has lived out its TTL. + #[test] + fn test_ttl_clock_after_vacuum_takes_newest_write() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let ttl = crate::storage::needle::ttl::TTL::read("5m").unwrap(); + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs(); + let old_write_ns = (now - 2 * 60 * 60) * 1_000_000_000; + let new_write_ns = (now - 60) * 1_000_000_000; + + let content_size = { + let mut v = make_ttl_volume(dir, ttl); + // Needle 1 is overwritten last but sorts first, so vacuum leaves it + // at the head of the .dat with the newest timestamp of the three. + for (id, append_at_ns) in [(2u64, old_write_ns), (3, old_write_ns), (1, new_write_ns)] { + let data = format!("data {}", id); + let mut n = Needle { + id: NeedleId(id), + cookie: Cookie(id as u32), + data: data.as_bytes().to_vec(), + data_size: data.len() as u32, + ..Needle::default() + }; + let (offset, _, _) = v.write_needle(&mut n, true, false).unwrap(); + v.sync_to_disk().unwrap(); + backdate_append_at_ns(&v.dat_path(), offset, n.size, append_at_ns); + } + v.compact_by_index(0, 0, |_| true).unwrap(); + v.commit_compact().unwrap(); + v.content_size() + }; + + let v = make_ttl_volume(dir, ttl); + assert_eq!(v.last_modified_ts(), new_write_ns / 1_000_000_000); + assert!( + !v.is_expired(content_size, 1024 * 1024), + "a volume overwritten a minute ago must not be expired after a vacuum and reload" + ); + } + + // Guard the destroy time an EC volume is reclaimed on: it was recomputed as + // now+TTL every time the .vif was written, so a read-only mark, a tier + // upload or an EC encode handed an already expiring volume another full TTL. + #[test] + fn test_expire_at_sec_counts_from_last_write() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let ttl = crate::storage::needle::ttl::TTL::read("5m").unwrap(); + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs(); + + let mut v = make_ttl_volume(dir, ttl); + // A volume with nothing written yet has no last write to count from, and + // must not land in 1970 with its data due for destruction on sight. + assert!(v.expire_at_sec() >= now); + + v.set_last_modified_ts_for_test(now - 3600); + let want = now - 3600 + ttl.to_seconds(); + for pass in 0..2 { + v.save_volume_info().unwrap(); + assert_eq!( + v.volume_info.expire_at_sec, want, + ".vif save {} moved the destroy time off the last write", + pass + ); + } + } + + fn make_ttl_volume(dir: &str, ttl: crate::storage::needle::ttl::TTL) -> Volume { + Volume::new( + dir, + dir, + "", + VolumeId(1), + NeedleMapKind::InMemory, + None, + Some(ttl), + 0, + Version::current(), + ) + .unwrap() + } + + fn backdate_append_at_ns(dat_path: &str, offset: u64, size: Size, append_at_ns: u64) { + let ts_offset = + offset + NEEDLE_HEADER_SIZE as u64 + size.0 as u64 + NEEDLE_CHECKSUM_SIZE as u64; + let mut f = OpenOptions::new().write(true).open(dat_path).unwrap(); + f.seek(SeekFrom::Start(ts_offset)).unwrap(); + f.write_all(&append_at_ns.to_be_bytes()).unwrap(); + } + fn write_three_needles(dir: &str) { let mut v = make_test_volume(dir); for i in 1..=3u64 { diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index 5a0a12fe0..64085cfb6 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/weed/server/volume_grpc_erasure_coding.go @@ -139,15 +139,8 @@ func (vs *VolumeServer) VolumeEcShardsGenerate(ctx context.Context, req *volume_ } // write .vif files - var expireAtSec uint64 - if v.Ttl != nil { - ttlSecond := v.Ttl.ToSeconds() - if ttlSecond > 0 { - expireAtSec = uint64(time.Now().Unix()) + ttlSecond //calculated expiration time - } - } volumeInfo := &volume_server_pb.VolumeInfo{Version: uint32(v.Version())} - volumeInfo.ExpireAtSec = expireAtSec + volumeInfo.ExpireAtSec = v.ExpireAtSec() // The size the encode actually read, not a separate stat: a replica-sync // write can land between two stats of a live .dat, and the .vif would then // record a DatFileSize and a BlockSize describing different files. diff --git a/weed/storage/volume.go b/weed/storage/volume.go index 1550a5f4a..32a0ae817 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -442,6 +442,25 @@ func (v *Volume) expired(contentSize uint64, volumeSizeLimit uint64) bool { return false } +// ExpireAtSec is when this volume's data becomes garbage, counted from its last +// write. Counting from the current time instead let every .vif rewrite -- a +// read-only mark, a tier upload, an EC encode -- hand an already expiring volume +// another full TTL. Zero when the volume has no TTL. +func (v *Volume) ExpireAtSec() uint64 { + if v.Ttl == nil { + return 0 + } + ttlSeconds := v.Ttl.ToSeconds() + if ttlSeconds == 0 { + return 0 + } + lastWriteSec := v.lastModifiedTsSeconds + if lastWriteSec == 0 { + lastWriteSec = uint64(time.Now().Unix()) + } + return lastWriteSec + ttlSeconds +} + // wait either maxDelayMinutes or 10% of ttl minutes func (v *Volume) expiredLongEnough(maxDelayMinutes uint32) bool { if v.Ttl == nil || v.Ttl.Minutes() == 0 { diff --git a/weed/storage/volume_checking.go b/weed/storage/volume_checking.go index bb8b2efb4..5961fb118 100644 --- a/weed/storage/volume_checking.go +++ b/weed/storage/volume_checking.go @@ -4,6 +4,7 @@ import ( "fmt" "io" "os" + "time" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/storage/backend" @@ -258,6 +259,124 @@ func doCheckAndFixVolumeData(v *Volume, indexFile *os.File, indexOffset int64) ( return lastAppendAtNs, nil } +// recoverLastModifiedTs points the TTL clock at the newest write recorded in +// the volume, replacing the .dat mtime the loader starts from. A delete appends +// a tombstone and vacuum rewrites the .dat wholesale, so the mtime moves +// without any write: every restart of a volume taking delete traffic re-armed +// expired() for another full TTL and the volume was never reclaimed. Left on +// the mtime when no write is recoverable. +func (v *Volume) recoverLastModifiedTs(indexFile *os.File) { + if v.Ttl == nil || v.Ttl.Minutes() == 0 { + return + } + indexSize, err := verifyIndexFileIntegrity(indexFile) + if err != nil || indexSize == 0 { + return + } + appendAtNs, err := findLastWriteAppendAtNs(v, indexFile, indexSize) + if err != nil { + glog.Warningf("volume %d recover last write from %s: %v", v.Id, indexFile.Name(), err) + return + } + if appendAtNs == 0 { + return + } + v.lastModifiedTsSeconds = appendAtNs / uint64(time.Second) +} + +// vacuumedLastWriteScanEntries bounds the work a vacuumed volume's recovery +// does, where key order makes every write a candidate for the newest one. A +// volume with more live needles than this keeps the .dat mtime: reading a +// subset could recover a timestamp older than the newest write and expire data +// still inside its TTL, so a scan that will not fit declines instead of +// guessing. A variable so tests can exercise that path. +var vacuumedLastWriteScanEntries = 1 << 16 + +// findLastWriteAppendAtNs scans the .idx backwards for the newest write -- an +// entry that is not a deletion tombstone -- and returns that needle's append +// timestamp. The .idx and the .dat share an order, so an append-ordered volume +// answers with the first write the scan reaches. Vacuum rewrites both in key +// order, which tracks write order only because the master issues keys +// increasing: an overwrite keeps its original, lower key, so a vacuumed volume +// has to take the maximum over every write it indexes. Returns 0 when the .idx +// holds nothing but tombstones, when a vacuumed volume holds more needles than +// the scan budget, or for a volume older than version 3, whose needles carry no +// append timestamp. +func findLastWriteAppendAtNs(v *Volume, indexFile *os.File, indexSize int64) (uint64, error) { + version := v.Version() + if version != needle.Version3 { + return 0, nil + } + scanEveryWrite := v.SuperBlock.CompactionRevision > 0 + entryBudget := vacuumedLastWriteScanEntries + var lastWriteAppendAtNs uint64 + block := make([]byte, types.NeedleMapEntrySize*idx.RowsToRead) + for end := indexSize; end > 0; { + start := max(end-int64(len(block)), 0) + entries := block[:end-start] + readCount, err := indexFile.ReadAt(entries, start) + if err == io.EOF && readCount == len(entries) { + err = nil + } + if err != nil { + return 0, fmt.Errorf("read %s at %d: %v", indexFile.Name(), start, err) + } + for i := len(entries) - types.NeedleMapEntrySize; i >= 0; i -= types.NeedleMapEntrySize { + key, offset, size := idx.IdxFileEntry(entries[i : i+types.NeedleMapEntrySize]) + if offset.IsZero() || size.IsDeleted() { + continue + } + needleOffset := findNeedleOffset(v.DataBackend, version, offset.ToActualOffset(), key, size) + if needleOffset < 0 { + continue + } + appendAtNs, err := readNeedleAppendAtNs(v.DataBackend, needleOffset, size) + if err != nil { + return 0, err + } + lastWriteAppendAtNs = max(lastWriteAppendAtNs, appendAtNs) + if !scanEveryWrite { + return lastWriteAppendAtNs, nil + } + if entryBudget--; entryBudget == 0 { + glog.V(0).Infof("volume %d: more than %d needles to scan for its last write, keeping the %s mtime", + v.Id, vacuumedLastWriteScanEntries, v.FileName(".dat")) + return 0, nil + } + } + end = start + } + return lastWriteAppendAtNs, nil +} + +// findNeedleOffset returns the .dat offset holding the needle an .idx entry +// describes, or -1 when no needle there matches it. A .dat past +// MaxPossibleVolumeSize wraps the 4-byte offsets in its .idx, so the needle can +// sit one volume size further in; doCheckAndFixVolumeData retries the same way. +func findNeedleOffset(datFile backend.BackendStorageFile, version needle.Version, offset int64, key types.NeedleId, size types.Size) int64 { + for _, at := range []int64{offset, offset + int64(types.MaxPossibleVolumeSize)} { + n, _, _, err := needle.ReadNeedleHeader(datFile, version, at) + if err == nil && n.Id == key && n.Size == size { + return at + } + } + return -1 +} + +// readNeedleAppendAtNs reads the append timestamp a version 3 needle carries +// past its checksum. +func readNeedleAppendAtNs(datFile backend.BackendStorageFile, offset int64, size types.Size) (uint64, error) { + bytes := make([]byte, types.TimestampSize) + readCount, err := datFile.ReadAt(bytes, offset+types.NeedleHeaderSize+int64(size)+needle.NeedleChecksumSize) + if err == io.EOF && readCount == types.TimestampSize { + err = nil + } + if err != nil { + return 0, err + } + return util.BytesToUint64(bytes), nil +} + func verifyIndexFileIntegrity(indexFile *os.File) (indexSize int64, err error) { if indexSize, err = util.GetFileSize(indexFile); err == nil { if indexSize%types.NeedleMapEntrySize != 0 { @@ -293,19 +412,13 @@ func verifyNeedleIntegrity(datFile backend.BackendStorageFile, v needle.Version, return 0, ErrorSizeMismatch } if v == needle.Version3 { - bytes := make([]byte, types.TimestampSize) - var readCount int - readCount, err = datFile.ReadAt(bytes, offset+types.NeedleHeaderSize+int64(size)+needle.NeedleChecksumSize) - if err == io.EOF && readCount == types.TimestampSize { - err = nil - } + n.AppendAtNs, err = readNeedleAppendAtNs(datFile, offset, size) if err == io.EOF { return 0, err } if err != nil { return 0, fmt.Errorf("verifyNeedleIntegrity check %s entry offset %d size %d: %v", datFile.Name(), offset, size, err) } - n.AppendAtNs = util.BytesToUint64(bytes) fileTailOffset := offset + needle.GetActualSize(size, v) fileSize, _, err := datFile.GetStat() if err != nil { diff --git a/weed/storage/volume_loading.go b/weed/storage/volume_loading.go index e78d048f0..5a120f371 100644 --- a/weed/storage/volume_loading.go +++ b/weed/storage/volume_loading.go @@ -293,6 +293,7 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind v.noWriteOrDelete = true glog.V(0).Infof("volumeDataIntegrityChecking failed %v", err) } + v.recoverLastModifiedTs(indexFile) } // The post-load structural check below uses the in-memory needle map diff --git a/weed/storage/volume_tier.go b/weed/storage/volume_tier.go index 204bb2076..e5985a9f7 100644 --- a/weed/storage/volume_tier.go +++ b/weed/storage/volume_tier.go @@ -2,7 +2,6 @@ package storage import ( "fmt" - "time" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" @@ -109,11 +108,8 @@ func (v *Volume) loadRemoteFileLocked() error { func (v *Volume) SaveVolumeInfo() error { tierFileName := v.FileName(".vif") - if v.Ttl != nil { - ttlSeconds := v.Ttl.ToSeconds() - if ttlSeconds > 0 { - v.volumeInfo.ExpireAtSec = uint64(time.Now().Unix()) + ttlSeconds //calculated destroy time from the ec volume was created - } + if expireAtSec := v.ExpireAtSec(); expireAtSec > 0 { + v.volumeInfo.ExpireAtSec = expireAtSec } return volume_info.SaveVolumeInfo(tierFileName, v.volumeInfo) diff --git a/weed/storage/volume_ttl_expiry_test.go b/weed/storage/volume_ttl_expiry_test.go new file mode 100644 index 000000000..3fd689c99 --- /dev/null +++ b/weed/storage/volume_ttl_expiry_test.go @@ -0,0 +1,249 @@ +package storage + +import ( + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// TestVolumeTtlClockSurvivesDeletes reproduces the delete-traffic TTL bug: +// deletes append a tombstone to the .dat, which moves the file's mtime, and +// the loader read the TTL clock back from that mtime. A volume taking delete +// traffic therefore had expired() re-armed for another full TTL on every +// restart and was never reclaimed. The clock has to come from the newest +// write instead. +func TestVolumeTtlClockSurvivesDeletes(t *testing.T) { + dir := t.TempDir() + ttl, err := needle.ReadTTL("5m") + if err != nil { + t.Fatalf("read ttl: %v", err) + } + + v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, ttl, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("volume creation: %v", err) + } + + // Backdate the writes on disk so the last one sits well outside the TTL, + // while the tombstones below leave the .dat mtime at now. + lastWriteNs := uint64(time.Now().Add(-2 * time.Hour).UnixNano()) + for i := 1; i <= 3; i++ { + n := newRandomNeedle(uint64(i)) + offset, _, _, err := v.writeNeedle2(n, true, false, false) + if err != nil { + t.Fatalf("write needle %d: %v", i, err) + } + backdateAppendAtNs(t, v, int64(offset), n.Size, lastWriteNs) + } + // More than one tombstone: the scan has to walk back over the whole run of + // them to reach a write. + for _, id := range []uint64{2, 3} { + if _, err := v.doDeleteRequest(newEmptyNeedle(id)); err != nil { + t.Fatalf("delete needle %d: %v", id, err) + } + } + contentSize := v.ContentSize() + v.Close() + + reloaded, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, ttl, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("reload: %v", err) + } + defer reloaded.Close() + + if got, want := reloaded.lastModifiedTsSeconds, lastWriteNs/uint64(time.Second); got != want { + t.Errorf("TTL clock recovered as %d, want the last write at %d", got, want) + } + if !reloaded.expired(contentSize, 1024*1024) { + t.Error("a TTL volume whose last write is 2h old must be expired after a reload") + } +} + +// TestVolumeTtlClockKeepsMtimeWithoutRecoverableWrite covers a TTL volume whose +// needles carry no append timestamp: the loader must stay on the .dat mtime +// rather than treat the volume as written at the epoch and drop it on sight. +func TestVolumeTtlClockKeepsMtimeWithoutRecoverableWrite(t *testing.T) { + dir := t.TempDir() + ttl, err := needle.ReadTTL("5m") + if err != nil { + t.Fatalf("read ttl: %v", err) + } + + v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, ttl, 0, needle.Version2, 0, 0) + if err != nil { + t.Fatalf("volume creation: %v", err) + } + if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false, false); err != nil { + t.Fatalf("write needle: %v", err) + } + contentSize := v.ContentSize() + v.Close() + + reloaded, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, ttl, 0, needle.Version2, 0, 0) + if err != nil { + t.Fatalf("reload: %v", err) + } + defer reloaded.Close() + + if reloaded.expired(contentSize, 1024*1024) { + t.Error("a just-written volume must not be expired after a reload") + } +} + +// TestVolumeTtlClockAfterVacuumTakesNewestWrite covers the one layout where a +// .dat's order does not track its write order: vacuum rewrites it by key, and +// an overwrite keeps its original, lower key. Reading the position rather than +// the timestamps would recover the highest-key needle's older write time and +// expire the volume before the overwrite has lived out its TTL. +func TestVolumeTtlClockAfterVacuumTakesNewestWrite(t *testing.T) { + dir := t.TempDir() + ttl, err := needle.ReadTTL("5m") + if err != nil { + t.Fatalf("read ttl: %v", err) + } + + v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, ttl, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("volume creation: %v", err) + } + + // Needle 1 is overwritten last but sorts first, so vacuum leaves it at the + // head of the .dat with the newest timestamp of the three. + oldWriteNs := uint64(time.Now().Add(-2 * time.Hour).UnixNano()) + newWriteNs := uint64(time.Now().Add(-time.Minute).UnixNano()) + for _, w := range []struct { + id uint64 + ns uint64 + }{{2, oldWriteNs}, {3, oldWriteNs}, {1, newWriteNs}} { + n := newRandomNeedle(w.id) + offset, _, _, err := v.writeNeedle2(n, true, false, false) + if err != nil { + t.Fatalf("write needle %d: %v", w.id, err) + } + backdateAppendAtNs(t, v, int64(offset), n.Size, w.ns) + } + if err := v.CompactByIndex(nil); err != nil { + t.Fatalf("compact: %v", err) + } + if err := v.CommitCompact(); err != nil { + t.Fatalf("commit compact: %v", err) + } + contentSize := v.ContentSize() + v.Close() + + reloaded, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, ttl, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("reload: %v", err) + } + defer reloaded.Close() + + if got, want := reloaded.lastModifiedTsSeconds, newWriteNs/uint64(time.Second); got != want { + t.Errorf("TTL clock recovered as %d, want the newest write at %d", got, want) + } + if reloaded.expired(contentSize, 1024*1024) { + t.Error("a volume overwritten a minute ago must not be expired after a vacuum and reload") + } +} + +// TestVolumeTtlClockDeclinesUnaffordableScan covers the budget the vacuumed +// path runs under. Reading a subset of a key-ordered volume's writes could +// recover a timestamp older than the newest write and expire live data, so a +// scan that does not fit has to leave the clock on the mtime instead. +func TestVolumeTtlClockDeclinesUnaffordableScan(t *testing.T) { + dir := t.TempDir() + ttl, err := needle.ReadTTL("5m") + if err != nil { + t.Fatalf("read ttl: %v", err) + } + + v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, ttl, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("volume creation: %v", err) + } + oldWriteNs := uint64(time.Now().Add(-2 * time.Hour).UnixNano()) + for i := 1; i <= 3; i++ { + n := newRandomNeedle(uint64(i)) + offset, _, _, err := v.writeNeedle2(n, true, false, false) + if err != nil { + t.Fatalf("write needle %d: %v", i, err) + } + backdateAppendAtNs(t, v, int64(offset), n.Size, oldWriteNs) + } + if err := v.CompactByIndex(nil); err != nil { + t.Fatalf("compact: %v", err) + } + if err := v.CommitCompact(); err != nil { + t.Fatalf("commit compact: %v", err) + } + contentSize := v.ContentSize() + v.Close() + + defer func(budget int) { vacuumedLastWriteScanEntries = budget }(vacuumedLastWriteScanEntries) + vacuumedLastWriteScanEntries = 2 + + reloaded, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, ttl, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("reload: %v", err) + } + defer reloaded.Close() + + if reloaded.lastModifiedTsSeconds == oldWriteNs/uint64(time.Second) { + t.Error("a scan that ran out of budget must not report a partial maximum as the last write") + } + if reloaded.expired(contentSize, 1024*1024) { + t.Error("declining the scan must leave the volume on its mtime, not expire it") + } +} + +// TestVolumeExpireAtSecCountsFromLastWrite guards the destroy time an EC volume +// is reclaimed on (erasure_coding.EcVolume.IsTimeToDestroy). It was recomputed +// as now+TTL on every .vif write, so a read-only mark, a tier upload or an EC +// encode handed an already expiring volume another full TTL. +func TestVolumeExpireAtSecCountsFromLastWrite(t *testing.T) { + dir := t.TempDir() + ttl, err := needle.ReadTTL("5m") + if err != nil { + t.Fatalf("read ttl: %v", err) + } + + v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, ttl, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("volume creation: %v", err) + } + defer v.Close() + + // A volume with nothing written yet has no last write to count from, and + // must not land in 1970 with its data due for destruction on sight. + if got := v.GetVolumeInfo().ExpireAtSec; got < uint64(time.Now().Unix()) { + t.Errorf("a fresh volume expires at %d, already in the past", got) + } + + if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false, false); err != nil { + t.Fatalf("write needle: %v", err) + } + v.lastModifiedTsSeconds = uint64(time.Now().Add(-time.Hour).Unix()) + want := v.lastModifiedTsSeconds + ttl.ToSeconds() + + for i := 0; i < 2; i++ { + if err := v.SaveVolumeInfo(); err != nil { + t.Fatalf("save .vif: %v", err) + } + if got := v.GetVolumeInfo().ExpireAtSec; got != want { + t.Fatalf(".vif save %d put ExpireAtSec at %d, want %d counted from the last write", i, got, want) + } + } +} + +func backdateAppendAtNs(t *testing.T, v *Volume, offset int64, size types.Size, appendAtNs uint64) { + t.Helper() + stamp := make([]byte, types.TimestampSize) + util.Uint64toBytes(stamp, appendAtNs) + tsOffset := offset + types.NeedleHeaderSize + int64(size) + needle.NeedleChecksumSize + if _, err := v.DataBackend.WriteAt(stamp, tsOffset); err != nil { + t.Fatalf("backdate the needle at offset %d: %v", offset, err) + } +}