diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 5d77e2818..791eb2aba 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -3766,18 +3766,40 @@ impl VolumeServer for VolumeGrpcService { ))); } - if !vol.volume_info.files.is_empty() { - vol.volume_info.files.remove(0); - } + // Snapshot the remote reference before dropping it: the + // refresh below can fail, and a half-applied transition + // leaves the volume claiming local while the remote backend + // is still attached and the on-disk .vif still says remote + // — a state a retry reads as "already on local disk" and + // refuses to finish. + let removed_remote = if vol.volume_info.files.is_empty() { + None + } else { + Some(vol.volume_info.files.remove(0)) + }; // Swaps the read-only sorted map out before the volume is // published as writable; without it the first write would // append to the local .dat and then fail to index. - vol.refresh_remote_write_mode().map_err(|e| { - Status::internal(format!( + if let Err(e) = vol.refresh_remote_write_mode() { + if let Some(remote) = removed_remote { + vol.volume_info.files.insert(0, remote); + } + // Put the derived flags and the needle map back where + // the restored reference says they belong. Best effort: + // if even this fails the volume stays pinned read-only, + // which is the safe end of the transition. + if let Err(restore_err) = vol.refresh_remote_write_mode() { + tracing::warn!( + volume_id = vid.0, + error = %restore_err, + "tier-down rollback could not restore the remote write mode", + ); + } + return Err(Status::internal(format!( "volume {} failed to refresh write mode: {}", vid, e - )) - })?; + ))); + } if let Err(e) = vol.save_volume_info() { return Err(Status::internal(format!( diff --git a/seaweed-volume/src/storage/needle_map.rs b/seaweed-volume/src/storage/needle_map.rs index cdbfa731a..08e9f0bd3 100644 --- a/seaweed-volume/src/storage/needle_map.rs +++ b/seaweed-volume/src/storage/needle_map.rs @@ -753,10 +753,12 @@ impl RedbNeedleMap { Ok(()) } - /// Look up a needle. - pub fn get(&self, key: NeedleId) -> Option { + /// Look up a needle. A redb failure is an ERROR, not an absent needle: + /// answering "not found" would turn a database problem into a read miss + /// and let a delete report success without recording a tombstone. + pub fn get(&self, key: NeedleId) -> io::Result> { let key_u64: u64 = key.into(); - self.get_internal(key_u64).ok().flatten() + self.get_internal(key_u64) } /// Internal get that returns io::Result for error propagation. @@ -994,10 +996,12 @@ impl NeedleMap { } } - /// Look up a needle. - pub fn get(&self, key: NeedleId) -> Option { + /// Look up a needle. Disk- and database-backed maps report their own + /// failures rather than folding them into "not found" — see the notes on + /// `RedbNeedleMap::get` and `SortedFileNeedleMap::get`. + pub fn get(&self, key: NeedleId) -> io::Result> { match self { - NeedleMap::InMemory(nm) => nm.get(key), + NeedleMap::InMemory(nm) => Ok(nm.get(key)), NeedleMap::Redb(nm) => nm.get(key), NeedleMap::SortedFile(nm) => nm.get(key), } @@ -1309,13 +1313,13 @@ mod tests { nm.put(NeedleId(2), Offset::from_actual_offset(128), Size(200)) .unwrap(); - let v1 = nm.get(NeedleId(1)).unwrap(); + let v1 = nm.get(NeedleId(1)).unwrap().unwrap(); assert_eq!(v1.size, Size(100)); - let v2 = nm.get(NeedleId(2)).unwrap(); + let v2 = nm.get(NeedleId(2)).unwrap().unwrap(); assert_eq!(v2.size, Size(200)); - assert!(nm.get(NeedleId(99)).is_none()); + assert!(nm.get(NeedleId(99)).unwrap().is_none()); } #[test] @@ -1340,7 +1344,7 @@ mod tests { assert_eq!(nm.deleted_size(), 100); // Deleted entry should have negated size - let nv = nm.get(NeedleId(1)).unwrap(); + let nv = nm.get(NeedleId(1)).unwrap().unwrap(); assert_eq!(nv.size, Size(-100)); } @@ -1413,9 +1417,9 @@ mod tests { let mut cursor = Cursor::new(idx_data); let nm = RedbNeedleMap::load_from_idx(db_path.to_str().unwrap(), &mut cursor, Version::current()).unwrap(); - assert!(nm.get(NeedleId(1)).is_some()); - assert!(nm.get(NeedleId(2)).is_none()); // deleted and removed - assert!(nm.get(NeedleId(3)).is_some()); + assert!(nm.get(NeedleId(1)).unwrap().is_some()); + assert!(nm.get(NeedleId(2)).unwrap().is_none()); // deleted and removed + assert!(nm.get(NeedleId(3)).unwrap().is_some()); assert_eq!(nm.file_count(), 2); } @@ -1532,7 +1536,7 @@ mod tests { let mut nm = NeedleMap::InMemory(CompactNeedleMap::new()); nm.put(NeedleId(1), Offset::from_actual_offset(0), Size(100)) .unwrap(); - assert_eq!(nm.get(NeedleId(1)).unwrap().size, Size(100)); + assert_eq!(nm.get(NeedleId(1)).unwrap().unwrap().size, Size(100)); assert_eq!(nm.file_count(), 1); } @@ -1543,7 +1547,7 @@ mod tests { let mut nm = NeedleMap::Redb(RedbNeedleMap::new(db_path.to_str().unwrap()).unwrap()); nm.put(NeedleId(1), Offset::from_actual_offset(0), Size(100)) .unwrap(); - assert_eq!(nm.get(NeedleId(1)).unwrap().size, Size(100)); + assert_eq!(nm.get(NeedleId(1)).unwrap().unwrap().size, Size(100)); assert_eq!(nm.file_count(), 1); } } diff --git a/seaweed-volume/src/storage/needle_map/sorted_file.rs b/seaweed-volume/src/storage/needle_map/sorted_file.rs index ba000e9c9..557124c4c 100644 --- a/seaweed-volume/src/storage/needle_map/sorted_file.rs +++ b/seaweed-volume/src/storage/needle_map/sorted_file.rs @@ -10,10 +10,11 @@ //! operation from [`file_pool`](super::file_pool) — so a volume nobody is //! reading costs zero fds and zero index bytes of RAM. +use std::collections::HashMap; use std::fs::{File, OpenOptions}; use std::io; use std::sync::atomic::Ordering; -use std::sync::Mutex; +use std::sync::{Mutex, RwLock}; use super::file_pool::pooled_index_files; use crate::storage::idx; @@ -40,6 +41,19 @@ pub struct SortedFileNeedleMap { db_file_name: String, db_file_size: i64, tail: Mutex, + /// Needles whose tombstone is durable in `.idx` but whose `.sdx` record + /// this process could not mark. `.sdx` is only a search accelerator over + /// `.idx`, so a stale record there would keep serving bytes the volume has + /// already accepted a delete for, until a reload rebuilds it. Lookups + /// consult this first and report the needle deleted, which is what the + /// next reload will conclude too. + pending_tombstones: RwLock>, + /// Test seam: forces the in-place `.sdx` mark to fail so the tests can + /// exercise the "tombstone durable in `.idx`, mark did not land" path, + /// which no portable filesystem trick reproduces (a read-only `.sdx` + /// fails the borrow long before the mark). + #[cfg(test)] + fail_sdx_mark: std::sync::atomic::AtomicBool, } impl SortedFileNeedleMap { @@ -84,6 +98,9 @@ impl SortedFileNeedleMap { offset: index_file_size, needs_sync: false, }), + pending_tombstones: RwLock::new(HashMap::new()), + #[cfg(test)] + fail_sdx_mark: std::sync::atomic::AtomicBool::new(false), }) } @@ -97,21 +114,31 @@ impl SortedFileNeedleMap { &self.index_file_name } - pub fn get(&self, key: NeedleId) -> Option { - let file = match pooled_index_files().borrow(&self.db_file_name, false) { - Ok(file) => file, - Err(e) => { - tracing::warn!(sdx = %self.db_file_name, error = %e, "open sorted index"); - return None; - } - }; - match search_sorted_index(&file, self.db_file_size, key) { - Ok(Some((_, offset, size))) => Some(NeedleValue { offset, size }), - Ok(None) => None, - Err(e) => { - tracing::warn!(sdx = %self.db_file_name, error = %e, "search sorted index"); - None - } + /// Look up a needle. An unreadable or torn `.sdx` — or a pooled reopen that + /// fails — is an ERROR, not an absent needle: swallowing it would answer a + /// read with "not found" and let a delete report success without recording + /// a tombstone, both of which look like data loss to the caller. + pub fn get(&self, key: NeedleId) -> io::Result> { + if let Some(offset) = self + .pending_tombstones + .read() + .expect("pending tombstones lock") + .get(&key) + .copied() + { + return Ok(Some(NeedleValue { + offset, + size: TOMBSTONE_FILE_SIZE, + })); + } + let file = pooled_index_files() + .borrow(&self.db_file_name, false) + .map_err(|e| { + io::Error::new(e.kind(), format!("open {}: {}", self.db_file_name, e)) + })?; + match search_sorted_index(&file, self.db_file_size, key)? { + Some((_, offset, size)) => Ok(Some(NeedleValue { offset, size })), + None => Ok(None), } } @@ -128,6 +155,16 @@ impl SortedFileNeedleMap { /// deleted. A key that is absent or already deleted is a no-op, matching /// Go. pub fn delete(&self, key: NeedleId, offset: Offset) -> io::Result> { + if self + .pending_tombstones + .read() + .expect("pending tombstones lock") + .contains_key(&key) + { + // Already tombstoned in `.idx`; a second tombstone would double + // count, and the mark this process owes `.sdx` is still pending. + return Ok(None); + } let file = pooled_index_files().borrow(&self.db_file_name, true)?; let Some((entry_index, _, size)) = search_sorted_index(&file, self.db_file_size, key)? else { @@ -146,27 +183,63 @@ impl SortedFileNeedleMap { // already counts as deleted. self.append_to_index_file(key, offset, TOMBSTONE_FILE_SIZE)?; - let mut buf = [0u8; SIZE_SIZE]; - TOMBSTONE_FILE_SIZE.to_bytes(&mut buf); - write_at( - &file, - &buf, - entry_index * NEEDLE_MAP_ENTRY_SIZE as u64 + ENTRY_SIZE_OFFSET, - )?; - // Move the counters to where a reload would put them, so the heartbeat // does not report this volume as garbage-free until it restarts. The // tombstone is one more .idx row, and under the rule index_metric // applies both it and the row it supersedes count as deletions; the // bytes leave the live set without leaving the .idx. + // + // This belongs to the `.idx` append, not to the `.sdx` mark below: the + // append is what makes the delete durable and what a reload would + // count, and the retry path returns early once the overlay records it. + // Leaving the counters until after the mark would strand them at their + // pre-delete values whenever the mark fails. self.metric.file_count.fetch_add(1, Ordering::Relaxed); self.metric.deletion_count.fetch_add(2, Ordering::Relaxed); self.metric .deletion_byte_count .fetch_add(size.0.max(0) as u64, Ordering::Relaxed); + + // From here the delete is durable in `.idx`, so no lookup may report + // this needle live again. Record it before touching `.sdx` and clear + // the record only once that mark lands: if the mark fails, the entry + // stays and lookups fail closed until a reload rebuilds `.sdx` from + // the `.idx` that already carries the tombstone. + self.pending_tombstones + .write() + .expect("pending tombstones lock") + .insert(key, offset); + + self.mark_deleted_in_sdx(&file, entry_index)?; + self.pending_tombstones + .write() + .expect("pending tombstones lock") + .remove(&key); + Ok(Some(size)) } + /// Mark the `.sdx` record for `entry_index` deleted in place. + fn mark_deleted_in_sdx(&self, file: &File, entry_index: u64) -> io::Result<()> { + #[cfg(test)] + if self + .fail_sdx_mark + .load(std::sync::atomic::Ordering::Relaxed) + { + return Err(io::Error::new( + io::ErrorKind::Other, + "injected .sdx mark failure", + )); + } + let mut buf = [0u8; SIZE_SIZE]; + TOMBSTONE_FILE_SIZE.to_bytes(&mut buf); + write_at( + file, + &buf, + entry_index * NEEDLE_MAP_ENTRY_SIZE as u64 + ENTRY_SIZE_OFFSET, + ) + } + fn append_to_index_file(&self, key: NeedleId, offset: Offset, size: Size) -> io::Result<()> { let file = pooled_index_files().borrow(&self.index_file_name, true)?; let mut buf = [0u8; NEEDLE_MAP_ENTRY_SIZE]; @@ -215,6 +288,15 @@ impl SortedFileNeedleMap { where F: FnMut(NeedleId, &NeedleValue) -> io::Result<()>, { + // A needle whose `.sdx` mark failed still reads live off the record. + // `get` answers it from the overlay and so must every scan, or a + // compaction would copy a deleted needle forward as live. Snapshot it + // rather than hold the lock across the whole file. + let pending = self + .pending_tombstones + .read() + .expect("pending tombstones lock") + .clone(); let file = pooled_index_files().borrow(&self.db_file_name, false)?; let entry_count = self.db_file_size.max(0) as u64 / NEEDLE_MAP_ENTRY_SIZE as u64; // A batch per 1024 entries rather than a syscall per entry, matching @@ -229,8 +311,8 @@ impl SortedFileNeedleMap { read_exact_at(&file, bytes, done * NEEDLE_MAP_ENTRY_SIZE as u64)?; for entry in bytes.chunks_exact(NEEDLE_MAP_ENTRY_SIZE) { let (key, offset, size) = idx_entry_from_bytes(entry); - if !size.is_valid() { - continue; // deleted in place by a runtime delete + if !size.is_valid() || pending.contains_key(&key) { + continue; // deleted in place, or still awaiting that mark } f(key, &NeedleValue { offset, size })?; } @@ -539,11 +621,11 @@ mod tests { ); let m = SortedFileNeedleMap::open(&base, version()).unwrap(); - assert_eq!(m.get(NeedleId(1)).unwrap().size, Size(100)); - assert_eq!(m.get(NeedleId(3)).unwrap().size, Size(300)); + assert_eq!(m.get(NeedleId(1)).unwrap().unwrap().size, Size(100)); + assert_eq!(m.get(NeedleId(3)).unwrap().unwrap().size, Size(300)); // A key tombstoned before the .sdx was written is not in it at all. - assert!(m.get(NeedleId(2)).is_none()); - assert!(m.get(NeedleId(99)).is_none()); + assert!(m.get(NeedleId(2)).unwrap().is_none()); + assert!(m.get(NeedleId(99)).unwrap().is_none()); } #[test] @@ -620,12 +702,12 @@ mod tests { // .sdx is marked in place, so the needle reads back as a tombstone // without a reload — the same contract Go's Get has, where callers // check size.is_deleted(). - assert!(m.get(NeedleId(2)).unwrap().size.is_deleted()); + assert!(m.get(NeedleId(2)).unwrap().unwrap().size.is_deleted()); assert!(m .delete(NeedleId(2), Offset::from_actual_offset(16)) .unwrap() .is_none()); - assert!(!m.get(NeedleId(1)).unwrap().size.is_deleted()); + assert!(!m.get(NeedleId(1)).unwrap().unwrap().size.is_deleted()); } #[test] @@ -742,7 +824,7 @@ mod tests { std::fs::metadata(format!("{base}.sdx")).unwrap().len(), good ); - assert_eq!(m.get(NeedleId(3)).unwrap().size, Size(300)); + assert_eq!(m.get(NeedleId(3)).unwrap().unwrap().size, Size(300)); assert_eq!(m.iter_entries().unwrap().len(), 3); } @@ -756,7 +838,7 @@ mod tests { filetime_backdate(&format!("{base}.sdx")); let m = SortedFileNeedleMap::open(&base, version()).unwrap(); - assert_eq!(m.get(NeedleId(1)).unwrap().size, Size(100)); + assert_eq!(m.get(NeedleId(1)).unwrap().unwrap().size, Size(100)); } #[test] @@ -779,7 +861,7 @@ mod tests { assert_eq!(entries[entries.len() - 1].1.size, Size((total * 10) as i32)); // Every id is still reachable through the binary search. assert_eq!( - m.get(NeedleId(total as u64)).unwrap().size, + m.get(NeedleId(total as u64)).unwrap().unwrap().size, Size((total * 10) as i32) ); } @@ -829,7 +911,7 @@ mod tests { "an idle sorted map must hold no .idx/.sdx descriptor" ); - assert!(m.get(NeedleId(1)).is_some()); + assert!(m.get(NeedleId(1)).unwrap().is_some()); drop_pooled(&m); assert_eq!( super::super::file_pool::open_index_fds(dir.path()), @@ -850,6 +932,148 @@ mod tests { let f = OpenOptions::new().write(true).open(path).unwrap(); let _ = f.set_times(std::fs::FileTimes::new().set_modified(past)); } + + /// A `.sdx` that cannot be read is an I/O error, never a lookup miss: + /// reporting "not found" would answer reads with NotFound and let a delete + /// acknowledge success without recording a tombstone. + #[test] + fn get_reports_io_errors_instead_of_missing() { + let dir = tempfile::TempDir::new().unwrap(); + let base = base(&dir); + write_idx(&format!("{base}.idx"), &[(1, 8, 100), (2, 16, 200)]); + let m = SortedFileNeedleMap::open(&base, version()).unwrap(); + + // Tear the .sdx before anything borrows it: the map still believes the + // file holds the entries it recorded at open, so the binary search + // reads past the end. + let f = OpenOptions::new() + .write(true) + .open(format!("{base}.sdx")) + .unwrap(); + f.set_len(3).unwrap(); + drop(f); + + let err = m + .get(NeedleId(1)) + .expect_err("a torn .sdx must surface as an error, not a miss"); + assert_eq!(err.kind(), io::ErrorKind::UnexpectedEof, "got {err}"); + } + + /// The deletion counters describe what a reload of `.idx` would count, so + /// they belong to the durable append — not to the `.sdx` mark that may + /// fail after it. Leaving them behind would report the volume as + /// garbage-free while the tombstone already sits in `.idx`, and the + /// idempotent retry never applies them either. + #[test] + fn delete_counts_against_the_durable_idx_append() { + let dir = tempfile::TempDir::new().unwrap(); + let base = base(&dir); + write_idx(&format!("{base}.idx"), &[(1, 8, 100), (2, 16, 200)]); + let m = SortedFileNeedleMap::open(&base, version()).unwrap(); + let deleted_before = m.deleted_count(); + let bytes_before = m.deleted_size(); + + m.fail_sdx_mark + .store(true, std::sync::atomic::Ordering::Relaxed); + m.delete(NeedleId(1), Offset::from_actual_offset(8)) + .expect_err("the injected mark failure must surface"); + + assert_eq!( + m.deleted_count(), + deleted_before + 2, + "a durable tombstone must move the deletion count" + ); + assert_eq!( + m.deleted_size(), + bytes_before + 100, + "a durable tombstone must move the deleted bytes" + ); + assert!( + m.get(NeedleId(1)).unwrap().unwrap().size.is_deleted(), + "the needle must read as deleted while the mark is outstanding" + ); + + // The retry is a no-op: no second tombstone, no double counting. + assert_eq!( + m.delete(NeedleId(1), Offset::from_actual_offset(8)).unwrap(), + None + ); + assert_eq!(m.deleted_count(), deleted_before + 2); + assert_eq!(m.deleted_size(), bytes_before + 100); + } + + /// Once the tombstone is durable in `.idx`, no lookup may report the needle + /// live again — even when marking the `.sdx` record fails. + #[test] + fn failed_sdx_mark_still_reads_deleted() { + let dir = tempfile::TempDir::new().unwrap(); + let base = base(&dir); + write_idx(&format!("{base}.idx"), &[(1, 8, 100), (2, 16, 200)]); + let m = SortedFileNeedleMap::open(&base, version()).unwrap(); + + // Simulate the mark failing after the .idx append landed. + m.pending_tombstones + .write() + .unwrap() + .insert(NeedleId(1), Offset::from_actual_offset(8)); + + let nv = m.get(NeedleId(1)).unwrap().expect("entry still resolves"); + assert!( + nv.size.is_deleted(), + "a needle whose tombstone is durable must read as deleted, got {:?}", + nv.size + ); + // And a retry must not append a second tombstone for it. + assert_eq!( + m.delete(NeedleId(1), Offset::from_actual_offset(8)).unwrap(), + None + ); + } + + /// The scans feed compaction and the rebuilt `.idx`, so a needle the + /// overlay reports deleted must not come back through them as live. + #[test] + fn failed_sdx_mark_is_hidden_from_scans() { + let dir = tempfile::TempDir::new().unwrap(); + let base = base(&dir); + write_idx(&format!("{base}.idx"), &[(1, 8, 100), (2, 16, 200)]); + let m = SortedFileNeedleMap::open(&base, version()).unwrap(); + + m.fail_sdx_mark + .store(true, std::sync::atomic::Ordering::Relaxed); + m.delete(NeedleId(1), Offset::from_actual_offset(8)) + .expect_err("the injected mark failure must surface"); + + let ids: Vec = m + .iter_entries() + .unwrap() + .into_iter() + .map(|(id, _)| id) + .collect(); + assert_eq!(ids, vec![NeedleId(2)], "iter_entries must drop the needle"); + + let mut visited = Vec::new(); + m.ascending_visit(|id, _| { + visited.push(id); + Ok(()) + }) + .unwrap(); + assert_eq!(visited, vec![NeedleId(2)]); + + let saved = format!("{base}.check"); + m.save_to_idx(&saved).unwrap(); + let mut rows = Vec::new(); + idx::walk_index_file(&mut File::open(&saved).unwrap(), 0, |key, _, size| { + rows.push((key, size)); + Ok(()) + }) + .unwrap(); + assert_eq!( + rows, + vec![(NeedleId(2), Size(200))], + "save_to_idx must not write the needle back as live" + ); + } } // Only meaningful with 5-byte offsets, which is what production Go builds use. diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 0c719764b..64572c345 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -958,6 +958,11 @@ impl Volume { "cannot open the index for write; serving this volume without deletes" ); self.no_write_or_delete = true; + // Deletes are gone with the writer, so stop advertising + // them: leaving no_write_can_delete set would report + // this volume as delete-capable in metrics and mode + // checks while every delete is refused. + self.no_write_can_delete = false; self.load_index_inmemory(idx_path) } Err(retry_err) => Err(retry_err), @@ -1242,7 +1247,7 @@ impl Volume { ) -> Result { let _guard = self.data_file_access_control.read_lock(); let nm = self.nm_or_not_found()?; - let nv = nm.get(n.id).ok_or(VolumeError::NotFound)?; + let nv = nm.get(n.id)?.ok_or(VolumeError::NotFound)?; if nv.offset.is_zero() { return Err(VolumeError::NotFound); @@ -1484,7 +1489,7 @@ impl Volume { ) -> Result { let _guard = self.data_file_access_control.read_lock(); let nm = self.nm_or_not_found()?; - let nv = nm.get(n.id).ok_or(VolumeError::NotFound)?; + let nv = nm.get(n.id)?.ok_or(VolumeError::NotFound)?; if nv.offset.is_zero() { return Err(VolumeError::NotFound); @@ -1586,7 +1591,7 @@ impl Volume { needle_id: NeedleId, ) -> Result<(u64, u16), VolumeError> { let nm = self.nm_or_not_found()?; - let nv = nm.get(needle_id).ok_or(VolumeError::NotFound)?; + let nv = nm.get(needle_id)?.ok_or(VolumeError::NotFound)?; if nv.offset.is_zero() { return Err(VolumeError::NotFound); } @@ -1723,7 +1728,7 @@ impl Volume { // Cookie validation for existing needle (matches Go: check whenever nm.Get returns ok) if let Some(nm) = &self.nm { - if let Some(nv) = nm.get(n.id) { + if let Some(nv) = nm.get(n.id)? { let mut existing = Needle::default(); // Read only the header to check cookie self.read_needle_header_unlocked(&mut existing, nv.offset.to_actual_offset())?; @@ -1773,13 +1778,28 @@ impl Volume { self.last_append_at_ns = n.append_at_ns; // Update needle map (uses n.size = full body size, matching Go's nm.Put) - let should_update = if let Some(nm) = &self.nm { - match nm.get(n.id) { - Some(nv) => (nv.offset.to_actual_offset() as u64) < offset, - None => true, + let prior = match self.nm.as_ref() { + Some(nm) => nm.get(n.id), + None => Ok(None), + }; + let should_update = match prior { + Ok(Some(nv)) => (nv.offset.to_actual_offset() as u64) < offset, + Ok(None) => true, + Err(e) => { + if fsync { + // Leaves the same durable-but-unindexed record a failed put + // below does, so it gets the same treatment. + self.no_write_or_delete = true; + tracing::error!( + "volume {}: failed to read the prior mapping for a durable write at {}, \ + marking read only: {}", + self.id.0, + offset, + e + ); + } + return Err(VolumeError::Io(e)); } - } else { - true }; if should_update { @@ -1835,7 +1855,17 @@ impl Volume { } if let Some(nm) = &self.nm { - if let Some(nv) = nm.get(n.id) { + let existing = match nm.get(n.id) { + Ok(existing) => existing, + Err(e) => { + // Not a lookup miss: the index could not be read. Report + // "unknown" so the caller writes the needle again rather + // than treating an unreadable index as proof of a change. + warn!(volume_id = self.id.0, error = %e, "needle map lookup failed"); + return None; + } + }; + if let Some(nv) = existing { if !nv.offset.is_zero() && nv.size.is_valid() { let mut old = Needle::default(); let mut ro = ReadOption::default(); @@ -1905,7 +1935,10 @@ impl Volume { fn do_delete_request(&mut self, n: &mut Needle) -> Result { let (found, size, _stored_offset) = if let Some(nm) = &self.nm { - if let Some(nv) = nm.get(n.id) { + // Propagate: acknowledging the delete as a no-op because the index + // could not be read would drop the tombstone on the floor while the + // caller believes the needle is gone. + if let Some(nv) = nm.get(n.id)? { if !nv.size.is_deleted() { (true, nv.size, nv.offset) } else { @@ -2051,7 +2084,7 @@ impl Volume { continue; } - let Some(nv) = nm.get(key) else { + let Some(nv) = nm.get(key)? else { offset += total_size; continue; }; @@ -2601,7 +2634,7 @@ impl Volume { /// writable: swap out the read-only map, attach the .idx writer, persist the /// mark. Split out so set_writable has one rollback point rather than three. fn publish_writable(&mut self) -> Result<(), VolumeError> { - self.reopen_idx_for_write()?; + self.reconcile_index_mode()?; self.attach_idx_writer_if_missing()?; self.save_vif() } @@ -2613,12 +2646,21 @@ impl Volume { /// publishing a writable volume without this leaves every write appending to /// .dat and then failing to index it. A no-op unless the map is the sorted /// one and the flags have already moved. - fn reopen_idx_for_write(&mut self) -> Result<(), VolumeError> { - if self.use_sorted_index() || !matches!(self.nm, Some(NeedleMap::SortedFile(_))) { + fn reconcile_index_mode(&mut self) -> Result<(), VolumeError> { + let wants_sorted = self.use_sorted_index(); + let has_sorted = matches!(self.nm, Some(NeedleMap::SortedFile(_))); + if wants_sorted == has_sorted { return Ok(()); } + // Both directions matter. Tiering DOWN has to trade the read-only + // sorted map for a writable one before the volume is published as + // writable. Tiering UP has to install the sorted map, or a volume + // tiered while the server runs keeps its whole index in RAM and its + // .idx descriptor pinned until the next restart — the cost this map + // exists to avoid. + // // load_index only publishes self.nm once the new map is built, so a - // failure leaves the sorted map in place and the caller can recover. + // failure leaves the previous map in place and the caller can recover. self.load_index() } @@ -2638,7 +2680,7 @@ impl Volume { // only clear the remoteness-derived flag, not an operator mark self.no_write_can_delete = false; } - if let Err(e) = self.reopen_idx_for_write() { + if let Err(e) = self.reconcile_index_mode() { self.no_write_or_delete = true; return Err(e); } @@ -3066,7 +3108,7 @@ impl Volume { // Dedup check: if the same needle already exists with matching content, skip the write. // Matches Go's WriteNeedleBlob which reads existing needle and compares cookie+checksum+data. if let Some(nm) = &self.nm { - if let Some(nv) = nm.get(needle_id) { + if let Some(nv) = nm.get(needle_id)? { if nv.size == size { let version = self.version(); // Read existing needle from disk @@ -4419,7 +4461,7 @@ mod tests { ..Needle::default() }; v.write_needle(&mut kept, true, true).unwrap(); - let prior = v.nm.as_ref().unwrap().get(NeedleId(1)).unwrap(); + let prior = v.nm.as_ref().unwrap().get(NeedleId(1)).unwrap().unwrap(); let dat_len_before = std::fs::metadata(v.file_name(".dat")).unwrap().len(); let file_count_before = v.file_count(); let content_size_before = v.content_size(); @@ -4440,7 +4482,7 @@ mod tests { dat_len_before, "the unflushed append should be off the .dat" ); - let now = v.nm.as_ref().unwrap().get(NeedleId(1)).unwrap(); + let now = v.nm.as_ref().unwrap().get(NeedleId(1)).unwrap().unwrap(); assert_eq!( now.offset, prior.offset, "the mapping should never have moved" @@ -4570,7 +4612,7 @@ mod tests { dat_len_before ); assert!( - v.nm.as_ref().unwrap().get(NeedleId(7)).is_none(), + v.nm.as_ref().unwrap().get(NeedleId(7)).unwrap().is_none(), "a needle that never reached the disk must not be indexed at all" ); assert_eq!(v.file_count(), file_count_before); @@ -5857,6 +5899,60 @@ mod tests { // map it booted with is the read-only sorted one, whose put always fails, so // without a rebuild the first write appends to .dat and then cannot be // indexed — leaving bytes nothing references. + /// A volume tiered while the server runs must install the sorted map right + /// then. Reconciling only on the way down left it holding the whole index + /// in RAM and its .idx descriptor pinned until the next restart — the cost + /// the sorted map exists to avoid. + #[test] + fn test_tier_up_installs_the_sorted_needle_map() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut v = make_test_volume(dir); + for i in 1..=2u64 { + let mut n = Needle { + id: NeedleId(i), + cookie: Cookie(0x1234), + data: format!("needle-{i}").into_bytes(), + data_size: format!("needle-{i}").len() as u32, + ..Needle::default() + }; + v.write_needle(&mut n, true, false).unwrap(); + } + v.sync_to_disk().unwrap(); + assert!( + !matches!(v.nm, Some(NeedleMap::SortedFile(_))), + "a local volume starts with a writable in-memory map" + ); + + // What the tier-up handler does once the .dat is uploaded: record the + // remote reference and reconcile the mode. + v.volume_info.files.push(PbRemoteFile { + backend_type: "s3".to_string(), + backend_id: "vif_tierup_test".to_string(), + key: "remote-key".to_string(), + offset: 0, + file_size: v.dat_file_size().unwrap(), + modified_time: 123, + extension: ".dat".to_string(), + }); + v.refresh_remote_write_mode().unwrap(); + + assert!(v.has_remote_file); + assert!( + matches!(v.nm, Some(NeedleMap::SortedFile(_))), + "tier-up must install the sorted map without waiting for a restart" + ); + // The index still answers, now off .sdx. + let nv = v + .nm + .as_ref() + .unwrap() + .get(NeedleId(1)) + .unwrap() + .expect("needle 1 still resolves through the sorted index"); + assert!(!nv.size.is_deleted()); + } + #[test] fn test_tier_down_swaps_in_a_writable_needle_map() { let tmp = TempDir::new().unwrap(); @@ -6330,8 +6426,8 @@ mod tests { } // Lookups still resolve, straight off .sdx. - assert!(!v.nm.as_ref().unwrap().get(NeedleId(3)).unwrap().size.is_deleted()); - assert!(v.nm.as_ref().unwrap().get(NeedleId(9)).is_none()); + assert!(!v.nm.as_ref().unwrap().get(NeedleId(3)).unwrap().unwrap().size.is_deleted()); + assert!(v.nm.as_ref().unwrap().get(NeedleId(9)).unwrap().is_none()); // Deletes are still allowed on a tiered volume and land in .idx. let idx_before = std::fs::metadata(&idx_path).unwrap().len(); @@ -6348,7 +6444,7 @@ mod tests { idx_before + NEEDLE_MAP_ENTRY_SIZE as u64, "the tombstone must be appended to the .idx tail" ); - assert!(v.nm.as_ref().unwrap().get(NeedleId(3)).unwrap().size.is_deleted()); + assert!(v.nm.as_ref().unwrap().get(NeedleId(3)).unwrap().unwrap().size.is_deleted()); if crate::storage::needle_map::file_pool::open_index_fds(tmp.path()).is_some() { drop_pooled();