volume: expire TTL volumes whose only traffic is deletes (#11167)

* volume: count a TTL volume's age from its last write, not the .dat mtime

A delete appends a tombstone needle and vacuum rewrites the .dat wholesale,
so the file's mtime moves without any write ever landing. The loader read
lastModifiedTsSeconds back from that mtime, so every restart of a volume
taking delete traffic re-armed expired() for another full TTL: an
overwrite-heavy collection kept growing until it hit the max-volume cap.

Recover the clock from the newest .idx entry that is not a tombstone and
read that needle's append timestamp, falling back to the mtime when no
write is recoverable. Only TTL volumes pay for the scan.

Fixes #11160

* volume: count the .vif destroy time from the last write too

ExpireAtSec is what an EC volume is reclaimed on, and it was recomputed as
now+TTL every time the .vif was written. A read-only mark, a tier upload or
an EC encode therefore handed an already expiring volume another full TTL,
the same way the .dat mtime did.

Derive it from the volume's last write, falling back to now for a volume
that has not taken one yet so a fresh volume is not born expired.

* volume: mirror the last-write TTL clock in the Rust volume server

Same recovery as the Go loader: scan the .idx backwards for the newest
entry that is not a tombstone and take that needle's append timestamp,
leaving the clock on the .dat mtime when no write is recoverable.

* volume: mirror the last-write destroy time in the Rust volume server

Both .vif writers and the EC encode computed ExpireAtSec as now+TTL, the
same way Go did, so the destroy time moved every time the sidecar was
rewritten. Route all three through the volume's last write.

* volume: report the .dat mtime in the Rust heartbeat, like Go does

The Rust server reported its TTL clock as ModifiedAtSecond while Go
reports the .dat mtime. The shell's quiet-period gates (volume.tier.move,
volume.delete_empty) read that field as "last touched", which a delete
has to count towards even though the TTL clock deliberately ignores it --
and with the clock now recovered from the last write, the two drift
further apart.

* volume: take the newest write by timestamp on a vacuumed volume

The reverse .idx scan trusted position, which holds only while the .dat is
append ordered. Vacuum rewrites it in key order, and since an overwrite
keeps its original key, the highest-key survivor is not necessarily the
newest write -- the recovered clock could land up to a TTL early and take
the volume with data still inside its TTL.

A volume that has been vacuumed (CompactionRevision > 0) now takes the
maximum append timestamp over a bounded window of write entries instead.
An append-ordered volume still answers in one read.

* volume: never guess a vacuumed volume's last write, and resolve wrapped offsets

Two holes in the reverse scan, both from review:

A vacuumed volume's writes are ordered by key, so any of them can hold the
newest timestamp. Reading a capped window sampled the highest keys, which
could still miss a recently overwritten low-key needle and expire data
inside its TTL. The scan now covers every write a vacuumed volume indexes,
and a volume too large to scan keeps the .dat mtime rather than report a
partial maximum -- late is recoverable, early is not.

A .dat past MaxPossibleVolumeSize wraps the offsets in its .idx, so reading
a timestamp at the unwrapped offset picks up an unrelated needle. Resolve
the entry against the needle header first and retry one volume size in,
the way doCheckAndFixVolumeData already does.

* volume: drop GitHub issue references from TTL comments

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(),

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,

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()`.

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<u64, VolumeError> {
+        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<i64> {
+        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<u64, VolumeError> {
+        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 {