mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-08-19 13:46:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(volume): make tier-down crash-safe and serve from local (Rust) (#10113)

Browse Source 
* fix(volume): fsync .vif and downloaded tier .dat (Rust)

save_volume_info wrote the .vif with a plain write and no fsync, and the
tier download never synced the .dat it wrote. Either could be lost on a
crash before the tier-down path acts on them. fsync both, matching the Go
volume server's util.WriteFile and DownloadFile.

* fix(volume): swap to local before deleting remote on tier-down (Rust)

The tier-down path deleted the shared remote object before trimming the
.vif, so a crash in between left the volume's .vif pointing at a deleted
object. It also dropped the remote backend only on the delete path and
never opened the downloaded local .dat, so reads broke until reload and a
keep-remote download kept serving from the slow remote object.

Trim the .vif and swap to the local .dat on both paths, bracketed by
directory fsyncs, before removing the remote object; gate only the object
removal on keep_remote_dat_file. Matches the Go volume server's crash-safe
ordering.
This commit is contained in:
Chris Lu
2026-06-25 12:29:21 -07:00
committed by GitHub
parent 66620a1ab8
commit c2668fbc64
3 changed files with 229 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
seaweed-volume/src/remote_storage/s3_tier.rs
+12
View File
@@ -364,6 +364,18 @@ impl S3TierBackend {
.map_err(|e| format!("download task panicked: {}", e))??;
}
// fsync the file so its content is durable before the caller trims the .vif
// and deletes the remote object (matches Go's DownloadFile f.Sync()).
let synced = tokio::fs::OpenOptions::new()
.write(true)
.open(dest_path)
.await
.map_err(|e| format!("failed to open {} for fsync: {}", dest_path, e))?;
synced
.sync_all()
.await
.map_err(|e| format!("failed to fsync {}: {}", dest_path, e))?;
Ok(file_size)
}
seaweed-volume/src/server/grpc_server.rs
+199 -46
View File
@@ -3471,41 +3471,88 @@ impl VolumeServer for VolumeGrpcService {
let modified_ts_ns = (remote_modified_secs as i64).saturating_mul(1_000_000_000);
if let Err(e) = set_file_mtime(&dat_path, modified_ts_ns) {
tracing::warn!("volume {} restore data file {} modified time: {}", vid, dat_path, e);
}
}
if !keep_remote {
// Delete remote file
backend.delete_file(&storage_key).await.map_err(|e| {
Status::internal(format!(
"volume {} failed to delete remote file {}: {}",
vid, storage_key, e
))
})?;
// Update volume info: remove remote file reference
{
let mut store = state.store.write().unwrap();
if let Some((_, vol)) = store.find_volume_mut(vid) {
if !vol.volume_info.files.is_empty() {
vol.volume_info.files.remove(0);
}
vol.refresh_remote_write_mode();
if let Err(e) = vol.save_volume_info() {
return Err(Status::internal(format!(
"volume {} failed to save remote file info: {}",
vid, e
)));
}
// Close old remote backend (matches Go: v.DataBackend.Close(); v.DataBackend = nil)
// This forces the next read to discover and open the local .dat file.
vol.close_remote_dat_backend();
} else if let Ok(dat_file) = tokio::fs::File::open(&dat_path).await {
// Persist the restored mtime past the download's content fsync;
// best-effort, only TTL accuracy depends on it surviving a crash.
if let Err(e) = dat_file.sync_all().await {
tracing::warn!("volume {} fsync data file {} after mtime restore: {}", vid, dat_path, e);
}
}
}
// fsync the directory so the freshly downloaded .dat is durably linked
// before we rewrite the .vif and (later) delete the shared remote object.
if let Err(e) = crate::storage::volume::fsync_dir(&dat_path) {
return Err(Status::internal(format!(
"volume {} fsync dir for {}: {}",
vid, dat_path, e
)));
}
// Trim the remote reference, persist the .vif, and swap to the local
// .dat on BOTH paths BEFORE deleting the remote object. After this the
// volume serves from local disk (has_remote_file = false), so a crash
// before the delete only leaks the remote object; the .vif must never
// reference an object that has already been deleted.
{
let mut store = state.store.write().unwrap();
let (_, vol) = store.find_volume_mut(vid).ok_or_else(|| {
Status::not_found(format!("volume {} disappeared during tier-down", vid))
})?;
// The volume could have been deleted/recreated or re-tiered while the
// download ran. Abort rather than trim the .vif or delete a remote
// object this volume no longer points at.
let (current_name, current_key) = vol.remote_storage_name_key();
if current_name != storage_name || current_key != storage_key {
return Err(Status::failed_precondition(format!(
"volume {} remote reference changed during tier-down",
vid
)));
}
if !vol.volume_info.files.is_empty() {
vol.volume_info.files.remove(0);
}
vol.refresh_remote_write_mode();
if let Err(e) = vol.save_volume_info() {
return Err(Status::internal(format!(
"volume {} failed to save remote file info: {}",
vid, e
)));
}
if let Err(e) = vol.open_local_dat_backend() {
return Err(Status::internal(format!(
"volume {} failed to open local dat file {}: {}",
vid, dat_path, e
)));
}
}
// fsync the directory again so the rewritten .vif is durable before the
// remote object is removed.
if let Err(e) = crate::storage::volume::fsync_dir(&dat_path) {
return Err(Status::internal(format!(
"volume {} fsync dir for {} after saving volume info: {}",
vid, dat_path, e
)));
}
if keep_remote {
// Surviving replicas still reference this object; keep it intact.
return Ok(());
}
// Only the last replica to download deletes the shared remote object.
backend.delete_file(&storage_key).await.map_err(|e| {
Status::internal(format!(
"volume {} failed to delete remote file {}: {}",
vid, storage_key, e
))
})?;
// Go does NOT send a final 100% progress message after download completion
Ok(())
}
@@ -4648,12 +4695,21 @@ mod tests {
assert!(!volume_is_remote_only(dat_path.to_str().unwrap(), false));
}
fn spawn_fake_s3_server(body: Vec<u8>) -> (String, tokio::sync::oneshot::Sender<()>) {
use axum::http::{header, HeaderMap, HeaderValue, StatusCode};
fn spawn_fake_s3_server(
body: Vec<u8>,
) -> (
String,
tokio::sync::oneshot::Sender<()>,
std::sync::Arc<std::sync::atomic::AtomicUsize>,
) {
use axum::http::{header, HeaderMap, HeaderValue, Method, StatusCode};
use axum::routing::any;
use axum::Router;
use std::sync::atomic::{AtomicUsize, Ordering};
let body = Arc::new(body);
let delete_count = Arc::new(AtomicUsize::new(0));
let delete_count_handler = delete_count.clone();
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap();
listener.set_nonblocking(true).unwrap();
@@ -4666,9 +4722,13 @@ mod tests {
.build()
.unwrap();
runtime.block_on(async move {
let app = Router::new().fallback(any(move |headers: HeaderMap| {
let app = Router::new().fallback(any(move |method: Method, headers: HeaderMap| {
let body = body.clone();
let delete_count = delete_count_handler.clone();
async move {
if method == Method::DELETE {
delete_count.fetch_add(1, Ordering::SeqCst);
}
let bytes = body.as_ref();
if let Some(range) = headers
.get(header::RANGE)
@@ -4732,15 +4792,16 @@ mod tests {
// Wait for the server thread to be ready before returning.
ready_rx.recv().unwrap();
(format!("http://{}", addr), shutdown_tx)
(format!("http://{}", addr), shutdown_tx, delete_count)
}
fn make_remote_only_service() -> (
fn make_remote_only_service(backend_id: &str) -> (
VolumeGrpcService,
TempDir,
tokio::sync::oneshot::Sender<()>,
Vec<u8>,
u64,
std::sync::Arc<std::sync::atomic::AtomicUsize>,
) {
let tmp = TempDir::new().unwrap();
let dir = tmp.path().to_str().unwrap();
@@ -4776,7 +4837,7 @@ mod tests {
let dat_path = format!("{}/1.dat", dir);
std::fs::remove_file(&dat_path).unwrap();
let (endpoint, shutdown_tx) = spawn_fake_s3_server(dat_bytes.clone());
let (endpoint, shutdown_tx, delete_count) = spawn_fake_s3_server(dat_bytes.clone());
// Use a test-specific backend_id to avoid racing with other tests
// that share the global registry. Never call clear() — only
// register/remove our own entries.
@@ -4791,16 +4852,13 @@ mod tests {
};
{
let mut registry = global_s3_tier_registry().write().unwrap();
registry.register(
"s3.incr_copy_test".to_string(),
S3TierBackend::new(&tier_config),
);
registry.register(format!("s3.{}", backend_id), S3TierBackend::new(&tier_config));
}
let vif = crate::storage::volume::VifVolumeInfo {
files: vec![crate::storage::volume::VifRemoteFile {
backend_type: "s3".to_string(),
backend_id: "incr_copy_test".to_string(),
backend_id: backend_id.to_string(),
key: "remote-key".to_string(),
offset: 0,
file_size: dat_bytes.len() as u64,
@@ -4859,9 +4917,13 @@ mod tests {
pre_stop_seconds: 0,
volume_state_notify: tokio::sync::Notify::new(),
write_queue: std::sync::OnceLock::new(),
s3_tier_registry: std::sync::RwLock::new(
crate::remote_storage::s3_tier::S3TierRegistry::new(),
),
s3_tier_registry: std::sync::RwLock::new({
// The tier-down handler resolves the backend from the per-server
// registry, so register it here too (reads use the global one).
let mut reg = crate::remote_storage::s3_tier::S3TierRegistry::new();
reg.register(format!("s3.{}", backend_id), S3TierBackend::new(&tier_config));
reg
}),
read_mode: crate::config::ReadMode::Local,
allow_untrusted_remote_endpoints: false,
master_url: String::new(),
@@ -4890,6 +4952,7 @@ mod tests {
shutdown_tx,
dat_bytes,
super_block_size,
delete_count,
)
}
@@ -4994,7 +5057,8 @@ mod tests {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_volume_incremental_copy_streams_remote_only_volume_data() {
let (service, _tmp, shutdown_tx, dat_bytes, super_block_size) = make_remote_only_service();
let (service, _tmp, shutdown_tx, dat_bytes, super_block_size, _delete_count) =
make_remote_only_service("incr_copy_test");
let response = service
.volume_incremental_copy(Request::new(
@@ -5021,6 +5085,95 @@ mod tests {
.remove("s3.incr_copy_test");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_tier_move_from_remote_swaps_to_local_and_deletes_object() {
let (service, tmp, shutdown_tx, dat_bytes, _super_block_size, delete_count) =
make_remote_only_service("tier_down_delete");
let dat_path = format!("{}/1.dat", tmp.path().to_str().unwrap());
assert!(!std::path::Path::new(&dat_path).exists());
let mut stream = service
.volume_tier_move_dat_from_remote(Request::new(
volume_server_pb::VolumeTierMoveDatFromRemoteRequest {
volume_id: 1,
collection: String::new(),
keep_remote_dat_file: false,
},
))
.await
.unwrap()
.into_inner();
while let Some(message) = stream.next().await {
message.unwrap();
}
// The downloaded .dat is on local disk and matches the remote bytes.
assert!(std::path::Path::new(&dat_path).exists());
assert_eq!(std::fs::read(&dat_path).unwrap(), dat_bytes);
// The volume now serves from local disk with no remote reference.
{
let store = service.state.store.read().unwrap();
let (_, vol) = store.find_volume(VolumeId(1)).unwrap();
assert!(!vol.has_remote_file);
assert!(vol.volume_info.files.is_empty());
assert!(vol.has_data_backend());
}
// The shared remote object was deleted exactly once.
assert_eq!(delete_count.load(std::sync::atomic::Ordering::SeqCst), 1);
let _ = shutdown_tx.send(());
global_s3_tier_registry()
.write()
.unwrap()
.remove("s3.tier_down_delete");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_tier_move_from_remote_keep_remote_still_swaps_to_local() {
let (service, tmp, shutdown_tx, dat_bytes, _super_block_size, delete_count) =
make_remote_only_service("tier_down_keep");
let dat_path = format!("{}/1.dat", tmp.path().to_str().unwrap());
let mut stream = service
.volume_tier_move_dat_from_remote(Request::new(
volume_server_pb::VolumeTierMoveDatFromRemoteRequest {
volume_id: 1,
collection: String::new(),
keep_remote_dat_file: true,
},
))
.await
.unwrap()
.into_inner();
while let Some(message) = stream.next().await {
message.unwrap();
}
// Even when the shared remote object is kept for other replicas, this
// replica swaps to its freshly downloaded local .dat and drops its
// remote reference.
assert!(std::path::Path::new(&dat_path).exists());
assert_eq!(std::fs::read(&dat_path).unwrap(), dat_bytes);
{
let store = service.state.store.read().unwrap();
let (_, vol) = store.find_volume(VolumeId(1)).unwrap();
assert!(!vol.has_remote_file);
assert!(vol.volume_info.files.is_empty());
assert!(vol.has_data_backend());
}
// The shared remote object is kept for the surviving replicas.
assert_eq!(delete_count.load(std::sync::atomic::Ordering::SeqCst), 0);
let _ = shutdown_tx.send(());
global_s3_tier_registry()
.write()
.unwrap()
.remove("s3.tier_down_keep");
}
/// Build a local service whose volume has a `.dat` large enough to span
/// several 2MB copy chunks, so the streaming copy paths are exercised
/// across multiple messages rather than a single buffer.
seaweed-volume/src/storage/volume.rs
+18 -4
View File
@@ -2225,10 +2225,16 @@ impl Volume {
}
}
/// Close the remote dat file backend (matches Go's v.DataBackend.Close(); v.DataBackend = nil).
/// Called after tier-download when the remote backend is being replaced by local storage.
pub fn close_remote_dat_backend(&mut self) {
/// Open the local .dat as the data backend, dropping any remote backend, so reads
/// are served from local disk. Mirrors Go's swapToLocalDatBackend after a tier-down.
pub(crate) fn open_local_dat_backend(&mut self) -> Result<(), VolumeError> {
let file = OpenOptions::new()
.read(true)
.write(true)
.open(self.dat_path())?;
self.remote_dat_file = None;
self.dat_file = Some(file);
Ok(())
}
/// Path to .vif file.
@@ -2366,7 +2372,15 @@ impl Volume {
let vif = VifVolumeInfo::from_pb(&self.volume_info);
let content = serde_json::to_string_pretty(&vif)
.map_err(|e| VolumeError::Io(io::Error::new(io::ErrorKind::Other, e.to_string())))?;
fs::write(&self.vif_path(), content)?;
// fsync the .vif so a tiered volume's remote reference is durable before the
// caller acts on it, e.g. deletes the remote object (matches Go util.WriteFile).
let mut f = OpenOptions::new()
.write(true)
.create(true)
.truncate(true)
.open(self.vif_path())?;
f.write_all(content.as_bytes())?;
f.sync_all()?;
Ok(())
}