mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-08-29 12:17:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

rust volume: fail closed on sorted-index failures and reconcile tier-up (#10956)

Browse Source 
* rust volume: fail closed on sorted-index failures and reconcile tier-up

Follow-ups to the .sdx sorted needle map (#10951):

- get() folded open/read failures into None, so an EIO, a torn .sdx, or a
  failed pooled reopen answered reads with NotFound and let do_delete_request
  acknowledge the delete as Ok(0) without writing a tombstone. It now returns
  io::Result and every caller propagates; redb's get() had the same shape and
  is fixed with it. is_file_unchanged cannot propagate, so it reports unknown
  and logs rather than treating an unreadable index as proof of a change.
- A delete whose .idx append landed but whose .sdx mark failed left the map
  still resolving the old live entry, so deleted content stayed readable until
  a reload. The map now records the tombstone before touching .sdx and only
  clears it once the mark lands; lookups consult that first and report the
  needle deleted, which is what the next reload concludes anyway.
- Mode reconciliation ran one way. Entering remote mode made use_sorted_index()
  true, which returned early, so a volume tiered while the server runs kept its
  in-memory map and pinned .idx descriptor until restart — the RAM and fd win
  never applied. It now reconciles in both directions.
- Tier-down dropped the remote reference before the fallible refresh, so a
  failure left volume_info local, the remote backend attached, the .vif still
  remote, and a retry reporting "already on local disk". The transition is
  snapshotted and rolled back.
- The read-only fallback set no_write_or_delete but left no_write_can_delete,
  so metrics and mode checks called the volume delete-capable while every
  delete was refused.

* rust volume: count a sorted-map delete against the durable .idx append

The deletion counters sat after the in-place .sdx mark, so a mark that
failed left them at their pre-delete values while the tombstone was already
durable in .idx — and with retries now idempotent, nothing applied them
later either. Heartbeats, status responses, and the garbage calculation
would report the volume as free of that garbage until a reload.

Move them to the append that makes the delete durable, which is also what a
reload of .idx would count. Covered by a test that injects a mark failure
through a cfg(test) seam: no portable filesystem trick reproduces it, since
a read-only .sdx fails the borrow long before the mark.

* rust volume: hide a pending tombstone from the sorted-map scans too

The overlay that keeps a needle deleted after a failed .sdx mark was only
consulted by get(). visit_live_entries still read the stale valid record
straight off .sdx, so ascending_visit, iter_entries and save_to_idx all
reported the needle live — and compaction takes iter_entries for the
complete live set, so it would copy the deleted content forward and
save_to_idx would write it back into the rebuilt .idx as live.

Snapshot the overlay once per scan and skip its keys, which is the same
conclusion the next reload reaches from the .idx tombstone.

* rust volume: quarantine a durable write whose index lookup fails

The prior-mapping lookup that decides whether to index a fresh append runs
after the record is already down and flushed, so a failing lookup leaves
exactly the state a failing put leaves: a durable .dat record nothing
indexes. The put path marks the volume read only for it; this one returned
the error and kept taking writes, and the next append would bury the
orphan mid-file where the .dat tail check on reload cannot see it.

Give it the same treatment.
This commit is contained in:
Chris Lu
2026-08-25 23:14:30 -07:00
committed by GitHub
parent 7658305c76
commit b77d954f55
4 changed files with 429 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
seaweed-volume/src/server/grpc_server.rs
+29 -7
View File
@@ -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!(
seaweed-volume/src/storage/needle_map.rs
+19 -15
View File
@@ -753,10 +753,12 @@ impl RedbNeedleMap {
Ok(())
}
/// Look up a needle.
pub fn get(&self, key: NeedleId) -> Option<NeedleValue> {
/// 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<Option<NeedleValue>> {
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<NeedleValue> {
/// 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<Option<NeedleValue>> {
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);
}
}
seaweed-volume/src/storage/needle_map/sorted_file.rs
+260 -36
View File
@@ -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<IndexTail>,
/// 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<HashMap<NeedleId, Offset>>,
/// 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<NeedleValue> {
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<Option<NeedleValue>> {
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<Option<Size>> {
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<NeedleId> = 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.
seaweed-volume/src/storage/volume.rs
+121 -25
View File
@@ -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<i32, VolumeError> {
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<NeedleStreamInfo, VolumeError> {
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<Size, VolumeError> {
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();