Github-Mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-09-08 15:41:15 +02:00
Code Issues Packages Projects Releases Wiki Activity

rust volume: stop a VolumeCopy whose caller has gone (#11188)

Browse Source 
* rust volume: stop a VolumeCopy whose caller has gone

VolumeCopy runs its copy in a detached tokio::spawn and reports progress
with the send error discarded, so nothing observes the client leaving.
When the caller cancels the RPC -- which weed-admin's batch balance does
routinely, starting far more copies than it finishes -- the server
streamed the whole volume from the source, wrote it to disk, and mounted
it. The destination is then left holding a volume nobody took delivery
of: its index cache is never reclaimed, and under replication=000 one
volume id ends up on two servers, both writable, which concurrent writes
can diverge.

Three checks now reach the task:

- Every chunk in copy_file_from_source, via the sender's is_closed().
  This is the one that matters in practice. The first progress report is
  128MB in, so for a smaller volume -- the ordinary balance move -- no
  send ever happens and its result says nothing; only the closed channel
  does. The sender is passed for the .idx and .vif copies too, with
  reporting gated separately, so those phases notice as well.
- The throttle sleep, which for a throttled copy runs for seconds at a
  time, now races the sender's closed() instead of being slept through.
- Immediately before mount_volume, and once at the top of the task.

Cancellation surfaces as an ordinary Err(Status::cancelled), so it lands
in the existing error branch that already removes the partial .dat/.idx/
.vif and the .note. That branch also logs now: the error otherwise went
to a channel nobody was reading, leaving the operator with the
balancer's "delete that copy, then re-run the move" and no cause.

This also clears the stranded read-only sources reported on the issue.
They are downstream of the orphan mount, not a separate defect:
LiveMoveVolume's cleanup probes the target before undoing the freeze
(volume_move.go:95, "the server can finish the copy and mount the target
even when the client loses the stream"), and when it finds a mounted
copy it cannot attribute, or cannot delete, it deliberately keeps the
source readonly rather than risk two writable replicas -- the messages
at volume_move.go:110 and :123. With nothing mounted on the target the
probe reports clean and the freeze is undone.

On Go parity: the progress send result is honoured here too, matching
`return false` in volume_grpc_copy.go. But that report is Go's only
abort signal, and measured against a 120MiB volume -- above the
throttler's activation threshold, below the 128MiB report interval -- a
Go destination mounts an abandoned copy as well. The issue's premise
that Go aborts holds only above the report interval. The Rust side now
stops in both cases; the Go behaviour is worth its own issue.

Tests: the integration test runs against both implementations and is
green on Go, red on Rust before this change. Its 192MiB fixture is sized
for two separate constraints, documented at the fixture: IoBytePerSecond
is a no-op below ~100ms of wall clock (64MiB copies in ~110ms on a tmpfs
loopback cluster), and the payload must exceed the 128MiB report
interval for the Go leg to pass at all. The two Rust unit tests cover
what the integration test cannot reach: cancellation detected with no
progress report at all, and the cleanup of the partial files plus the
.note.

Fixes #11186

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0122W3eqt6gmLUMxmRoZdPAb

* rust volume: surface VolumeCopy cancellation as Status::cancelled

copy_file_from_source returned Result<_, String>, so the per-chunk
cancellation path -- the one that matters in practice for volumes
below the 128MB report interval -- was wrapped to Status::internal at
the call sites. The spawn logging branch then classified it as a
generic failure instead of the intended "abandoned by caller",
defeating the logging change in the same PR for the case that occurs
most often.

Return Result<_, Status> from copy_file_from_source: Status::cancelled
for caller-gone, Status::internal for the existing errors. Drop the
.map_err(|e| Status::internal(e)) at the three call sites.

The unit test now asserts the code is Cancelled, not just the message
text, so the classification is locked in.

* rust volume: close cancellation gaps in VolumeCopy

Address two review findings on the same PR:

1. Roll back a mount that races a departing caller. The pre-mount
is_closed() check cannot close the window between the check and
mount_volume: if the receiver drops in that gap, the volume mounts
and the final tx.send(Ok(...)) fails, but its error was discarded
(let _ =), so the task returned Ok(()) and the error branch never
ran. The destination then held an orphaned mounted replica — the
exact defect this PR prevents.

   Fix: track a mounted flag. The final send now checks its result;
   on failure it returns Status::cancelled, and the error branch
   calls store.delete_volume (which unmounts AND removes the files)
   when mounted is true, instead of only unlinking.

2. Observe cancellation while awaiting the source stream. The
per-chunk is_closed() check only runs after stream.message().await
returns. A stalled source (slow disk, partition, GC pause) never
delivers a chunk, so a caller that has already left cannot preempt
the read: the task, the source connection, and the partial files
(including the .note) all outlive the caller indefinitely.

   Fix: race stream.message() against progress_tx.closed() in a
   tokio::select!, so a departing caller preempts a stalled source.

Adds test_volume_copy_after_mount_cancellation_rolls_back_mount to
cover the after-mount rollback path. cargo test --release green
(497 + 5 + 1 + 28).

* rust volume: keep remote data on after-mount rollback, race RPC startup

Two review findings on the after-mount rollback added in ff51f6a:

1. The rollback called delete_volume(vid, false, false), i.e.
   keep_remote_data=false. A remote-tier copy .vif points at the same
   cloud object the source replica references, so destroying the
   abandoned destination with keep_remote_data=false deletes the
   source remote data via Volume::destroy backend.delete_file_blocking.
   Use keep_remote_data=true, matching the pre-spawn delete_volume at
   the top of volume_copy.

2. client.copy_file(copy_req).await (the initial RPC establishment)
   was not raced against progress_tx.closed(). If the source stalls
   before sending response headers, the per-message select! added in
   ff51f6a is never reached, and the task, source connection, and
   preallocated files outlive a departed caller. Race the RPC
   establishment against progress_tx.closed() the same way.

cargo test --release green (497 + 5 + 1 + 28).

* rust volume: race master-configuration wait against caller cancellation

try_get_master_configuration().await was the last un-raced await in
the VolumeCopy task before copy_file_from_source. A stalled master
(or slow leader election) would hold the task and its .note past a
departing caller, since the per-chunk cancellation checks are never
reached. Race it against tx.closed() the same way the source stream
reads already are.

cargo test --release green (497 + 5 + 1 + 28).

* rust volume: fix after-mount rollback test to actually reach that path

The previous version of
test_volume_copy_after_mount_cancellation_rolls_back_mount dropped the
response immediately after volume_copy returned, so tx.is_closed() was
already true at the spawn first check and the task returned before
mount_volume. The test was green for the wrong reason: the mounted
flag, the rollback, and the keep_remote_data=true line were all
uncovered.

Use the store write lock as a seam: take it immediately after
volume_copy returns so the task runs the copy to completion with the
caller still attached, passes the pre-mount is_closed() check, then
parks entering the mount block. Drop the response (caller gone) and
release the guard: the task mounts, fails the final tx.send, and must
roll back via delete_volume.

Verified by setting the rollback guard to if false: the test fails
with "destination still holds a mounted volume". With the rollback
enabled, probe eprintlns confirmed the full path: about to mount ->
mounted = true -> final send failed -> rolling back mount -> rollback
done.

cargo test --release green (497 + 5 + 1 + 28).

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>
This commit is contained in:
Eliah Rusin
2026-09-06 10:47:18 -07:00
committed by GitHub
co-authored by Claude Opus 5 Chris Lu
parent b82cb05d71
commit 4b41329e12
2 changed files with 530 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
seaweed-volume/src/server/grpc_server.rs
+423 -44
View File
@@ -1396,7 +1396,22 @@ impl VolumeServer for VolumeGrpcService {
let state = self.state.clone();
tokio::spawn(async move {
// Tracks whether mount_volume succeeded so the error branch can
// unmount (not just unlink) an orphaned replica. A mount that
// races a departing caller is the one case where the existing
// file-only cleanup leaves the volume loaded in memory.
let mut mounted = false;
let result = async {
// Nothing below is worth doing for a caller that has already
// gone: the transfer would spend the source's bandwidth and the
// destination's disk on a volume nobody will take delivery of.
if tx.is_closed() {
return Err(Status::cancelled(format!(
"volume {} copy cancelled by caller",
vid
)));
}
let report_interval: i64 = 128 * 1024 * 1024;
let mut next_report_target: i64 = report_interval;
let io_byte_per_second = if req.io_byte_per_second > 0 {
@@ -1410,12 +1425,24 @@ impl VolumeServer for VolumeGrpcService {
let mut preallocate_size: i64 = 0;
if !has_remote_dat {
let grpc_addr = super::heartbeat::to_grpc_address(&state.master_url);
match super::heartbeat::try_get_master_configuration(
// Race the master-configuration RPC against the caller's
// response channel: a stalled master (or a slow leader
// election) would otherwise hold the task and its .note
// past a departing caller, since the per-chunk checks in
// copy_file_from_source are never reached.
let config = tokio::select! {
res = super::heartbeat::try_get_master_configuration(
&grpc_addr,
state.outgoing_grpc_tls.as_ref(),
)
.await
{
) => res,
_ = tx.closed() => {
return Err(Status::cancelled(format!(
"volume {} copy cancelled by caller",
vid
)));
}
};
match config {
Ok(resp) => {
if resp.volume_preallocate {
preallocate_size = resp.volume_size_limit_m_b as i64 * 1024 * 1024;
@@ -1454,13 +1481,13 @@ impl VolumeServer for VolumeGrpcService {
".dat",
false,
true,
Some(&tx),
&tx,
true,
&mut next_report_target,
report_interval,
&mut throttler,
)
.await
.map_err(|e| Status::internal(e))?;
.await?;
if dat_modified_ts_ns > 0 {
let _ = set_file_mtime(&dat_path, dat_modified_ts_ns);
}
@@ -1479,13 +1506,13 @@ impl VolumeServer for VolumeGrpcService {
".idx",
false,
false,
None,
&tx,
false,
&mut next_report_target,
report_interval,
&mut throttler,
)
.await
.map_err(|e| Status::internal(e))?;
.await?;
if idx_modified_ts_ns > 0 {
let _ = set_file_mtime(&idx_path, idx_modified_ts_ns);
}
@@ -1503,13 +1530,13 @@ impl VolumeServer for VolumeGrpcService {
".vif",
false,
true,
None,
&tx,
false,
&mut next_report_target,
report_interval,
&mut throttler,
)
.await
.map_err(|e| Status::internal(e))?;
.await?;
if vif_modified_ts_ns > 0 {
let _ = set_file_mtime(&vif_path, vif_modified_ts_ns);
}
@@ -1546,6 +1573,18 @@ impl VolumeServer for VolumeGrpcService {
vol_info.dat_file_timestamp_seconds * 1_000_000_000
};
// An orphaned mount is how an abandoned copy does lasting
// damage: the destination carries that volume's index cache for
// the lifetime of the process, and under replication=000 the
// cluster is left holding one volume id on two servers, both
// writable, which concurrent writes can diverge.
if tx.is_closed() {
return Err(Status::cancelled(format!(
"volume {} copy cancelled by caller before mount",
vid
)));
}
// Mount the volume
{
let mut store = state.store.write().unwrap();
@@ -1554,23 +1593,69 @@ impl VolumeServer for VolumeGrpcService {
.map_err(|e| {
Status::internal(format!("failed to mount volume {}: {}", vid, e))
})?;
mounted = true;
}
state.volume_state_notify.notify_one();
// Send final response with last_append_at_ns
let _ = tx
// Send final response with last_append_at_ns. A failed send
// means the caller is gone: the mount above raced a departing
// receiver (the pre-mount is_closed() check cannot close that
// window), and leaving the volume mounted is exactly the orphan
// this PR prevents. Surface it as Cancelled so the error branch
// unmounts and deletes the replica it just created.
if tx
.send(Ok(volume_server_pb::VolumeCopyResponse {
last_append_at_ns: last_append_at_ns,
processed_bytes: 0,
}))
.await;
.await
.is_err()
{
return Err(Status::cancelled(format!(
"volume {} copy cancelled by caller after mount",
vid
)));
}
Ok::<(), Status>(())
}
.await;
if let Err(e) = result {
// Clean up on error
// An abandoned copy is otherwise invisible here: the error goes
// to a channel nobody is reading. Logging it gives the operator
// the cause behind the balancer's "delete that copy, then re-run
// the move".
if e.code() == tonic::Code::Cancelled {
tracing::info!(
"volume {} copy from {} abandoned by its caller, discarding the partial copy",
vid,
req.source_data_node
);
} else {
tracing::warn!(
"volume {} copy from {} failed: {}",
vid,
req.source_data_node,
e
);
}
// Clean up on error. If the volume was mounted (the after-mount
// race), delete_volume unmounts it from the store AND removes
// the .dat/.idx/.vif in one step; the remove_file calls below
// cover the never-mounted partial-file case and are harmless
// no-ops when delete_volume already removed the files.
//
// keep_remote_data=true matches the pre-spawn delete_volume at
// the top of volume_copy: a remote-tier copy's .vif points at
// the same cloud object the source replica references, so
// destroying the abandoned destination with keep_remote_data=
// false would delete the source's remote data.
if mounted {
let mut store = state.store.write().unwrap();
let _ = store.delete_volume(vid, false, true);
state.volume_state_notify.notify_one();
}
let _ = std::fs::remove_file(format!("{}.dat", data_base_name));
let _ = std::fs::remove_file(format!("{}.idx", idx_base_name));
let _ = std::fs::remove_file(format!("{}.vif", data_base_name));
@@ -5009,13 +5094,12 @@ async fn copy_file_from_source<T>(
ext: &str,
is_append: bool,
ignore_source_not_found: bool,
progress_tx: Option<
&tokio::sync::mpsc::Sender<Result<volume_server_pb::VolumeCopyResponse, Status>>,
>,
progress_tx: &tokio::sync::mpsc::Sender<Result<volume_server_pb::VolumeCopyResponse, Status>>,
report_progress: bool,
next_report_target: &mut i64,
report_interval: i64,
throttler: &mut WriteThrottler,
) -> Result<i64, String>
) -> Result<i64, Status>
where
T: tonic::client::GrpcService<tonic::body::BoxBody>,
T::Error: Into<tonic::codegen::StdError>,
@@ -5032,62 +5116,107 @@ where
ignore_source_file_not_found: ignore_source_not_found,
};
let mut stream = client
.copy_file(copy_req)
.await
.map_err(|e| {
format!(
// Cancellation surfaces as Status::cancelled so the spawn's error branch
// can distinguish it from ordinary failures (logging + future handling).
// The other errors here stay Status::internal, matching the prior
// `.map_err(|e| Status::internal(e))` at the call sites.
let cancelled = || {
Status::cancelled(format!(
"volume {} {} copy cancelled by caller",
volume_id, ext
))
};
// Race the initial RPC establishment against the caller's response channel:
// if the source stalls before sending response headers, the per-message
// select! below is never reached, and without this race the task, the
// source connection, and any preallocated files would outlive a caller
// that has already gone.
let mut stream = tokio::select! {
res = client.copy_file(copy_req) => {
res.map_err(|e| {
Status::internal(format!(
"failed to start copying volume {} {} file: {}",
volume_id, ext, e
)
))
})?
.into_inner();
.into_inner()
}
_ = progress_tx.closed() => return Err(cancelled()),
};
let mut file = if is_append {
std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(dest_path)
.map_err(|e| format!("open file {}: {}", dest_path, e))?
.map_err(|e| Status::internal(format!("open file {}: {}", dest_path, e)))?
} else {
std::fs::OpenOptions::new()
.write(true)
.create(true)
.truncate(true)
.open(dest_path)
.map_err(|e| format!("open file {}: {}", dest_path, e))?
.map_err(|e| Status::internal(format!("open file {}: {}", dest_path, e)))?
};
let mut progressed_bytes: i64 = 0;
let mut modified_ts_ns: i64 = 0;
while let Some(resp) = stream
.message()
.await
.map_err(|e| format!("receiving {}: {}", dest_path, e))?
{
// The source stream's message() future is the one await in this loop that
// can hang indefinitely: a stalled source (slow disk, network partition,
// GC pause on the source host) never returns a chunk, and without racing
// it against the caller's response channel the detached task, the source
// connection, and the partial files (including the .note) all outlive the
// caller. select! lets a departing caller preempt the source read.
loop {
let resp = tokio::select! {
msg = stream.message() => {
msg.map_err(|e| Status::internal(format!("receiving {}: {}", dest_path, e)))?
}
_ = progress_tx.closed() => return Err(cancelled()),
};
let resp = match resp {
Some(r) => r,
None => break,
};
if resp.modified_ts_ns != 0 {
modified_ts_ns = resp.modified_ts_ns;
}
if !resp.file_content.is_empty() {
// The caller's response stream is the only thing that makes this
// copy worth finishing. Checked every chunk rather than only where
// progress is reported: the first report lands 128MB in, so a
// smaller volume would otherwise run to completion — and mount —
// long after its caller stopped listening.
if progress_tx.is_closed() {
return Err(cancelled());
}
use std::io::Write;
file.write_all(&resp.file_content)
.map_err(|e| format!("write file {}: {}", dest_path, e))?;
.map_err(|e| Status::internal(format!("write file {}: {}", dest_path, e)))?;
progressed_bytes += resp.file_content.len() as i64;
throttler
.maybe_slowdown(resp.file_content.len() as i64)
.await;
// A throttled copy sleeps seconds at a time; wake for a departing
// caller instead of finishing the nap first.
tokio::select! {
_ = throttler.maybe_slowdown(resp.file_content.len() as i64) => {}
_ = progress_tx.closed() => return Err(cancelled()),
}
if let Some(tx) = progress_tx {
if progressed_bytes > *next_report_target {
let _ = tx
if report_progress && progressed_bytes > *next_report_target {
// Go aborts the transfer when this send fails
// (volume_grpc_copy.go: `return false`); so do we.
if progress_tx
.send(Ok(volume_server_pb::VolumeCopyResponse {
last_append_at_ns: 0,
processed_bytes: progressed_bytes,
}))
.await;
*next_report_target = progressed_bytes + report_interval;
.await
.is_err()
{
return Err(cancelled());
}
*next_report_target = progressed_bytes + report_interval;
}
}
}
@@ -5906,6 +6035,256 @@ mod tests {
(service, tmp, dat_bytes)
}
// Serve a VolumeGrpcService over a real gRPC endpoint. VolumeCopy dials its
// source_data_node rather than calling in process, so a cancellation test
// needs a source that is reachable over the wire.
async fn serve_source(
service: VolumeGrpcService,
) -> (u16, tokio::sync::oneshot::Sender<()>) {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
tokio::spawn(async move {
let _ = tonic::transport::Server::builder()
.add_service(
crate::pb::volume_server_pb::volume_server_server::VolumeServerServer::new(
service,
),
)
.serve_with_incoming_shutdown(
tokio_stream::wrappers::TcpListenerStream::new(listener),
async {
let _ = shutdown_rx.await;
},
)
.await;
});
(port, shutdown_tx)
}
// The transfer itself must notice a caller that has gone, with no progress
// report to carry the news. The first report is 128MB in, so for anything
// smaller -- the ordinary case for a balance move -- the send result says
// nothing and only the closed channel does.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_copy_file_from_source_stops_when_the_caller_is_gone() {
let (source_service, _source_tmp, dat_bytes) = make_local_service_with_large_volume();
assert!(
dat_bytes.len() > 2 * 1024 * 1024,
"fixture must span several 2MB chunks"
);
let (port, _shutdown) = serve_source(source_service).await;
let channel = tonic::transport::Endpoint::from_shared(format!("http://127.0.0.1:{}", port))
.unwrap()
.connect()
.await
.unwrap();
let mut client =
volume_server_pb::volume_server_client::VolumeServerClient::new(channel)
.max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
.max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE);
let dest_tmp = TempDir::new().unwrap();
let dest_path = format!("{}/copied.dat", dest_tmp.path().to_str().unwrap());
let (tx, rx) = tokio::sync::mpsc::channel::<
Result<volume_server_pb::VolumeCopyResponse, Status>,
>(16);
drop(rx); // exactly what a client hanging up does to the sender
let mut next_report_target: i64 = 128 * 1024 * 1024;
let mut throttler = WriteThrottler::new(0);
let err = copy_file_from_source(
&mut client,
false,
"",
1,
u32::MAX,
dat_bytes.len() as u64,
&dest_path,
".dat",
false,
true,
&tx,
true,
&mut next_report_target,
128 * 1024 * 1024,
&mut throttler,
)
.await
.expect_err("a copy whose caller is gone must not run to completion");
// Cancellation must surface as Status::cancelled, not Status::internal:
// the spawn's error branch classifies on the code, so an Internal here
// would be logged as a generic failure instead of "abandoned by caller".
assert_eq!(
err.code(),
tonic::Code::Cancelled,
"expected Cancelled, got {:?}: {}",
err.code(),
err
);
assert!(
err.message().contains("cancelled by caller"),
"unexpected error: {}",
err
);
let copied = std::fs::metadata(&dest_path).map(|m| m.len()).unwrap_or(0);
assert!(
copied < dat_bytes.len() as u64,
"copy should have stopped short, wrote {} of {} bytes",
copied,
dat_bytes.len()
);
}
// An abandoned VolumeCopy must leave nothing behind: no mounted volume (its
// index cache would be held for the life of the process, and under
// replication=000 the cluster would carry one volume id on two writable
// servers), and none of the partial files or the .note, which fails the
// volume load on the next restart.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_volume_copy_cancelled_by_caller_mounts_nothing_and_cleans_up() {
let (source_service, _source_tmp, _dat_bytes) = make_local_service_with_large_volume();
let (port, _shutdown) = serve_source(source_service).await;
let (dest_service, dest_tmp) = make_local_service_with_volume("", None);
{
let mut store = dest_service.state.store.write().unwrap();
store.delete_volume(VolumeId(1), false, false).unwrap();
// available_space is filled in by the periodic disk check, which
// does not run in a unit test; without it VolumeCopy finds no
// location with room and never gets as far as copying.
for loc in &store.locations {
loc.check_disk_space();
}
}
let dest_dir = dest_tmp.path().to_str().unwrap().to_string();
let dest_file = |ext: &str| format!("{}/1{}", dest_dir, ext);
assert!(
!std::path::Path::new(&dest_file(".dat")).exists(),
"destination must start without the volume for this test to mean anything"
);
let response = dest_service
.volume_copy(Request::new(volume_server_pb::VolumeCopyRequest {
volume_id: 1,
collection: String::new(),
source_data_node: format!("127.0.0.1:1.{}", port),
disk_type: String::new(),
io_byte_per_second: 0,
replication: String::new(),
ttl: String::new(),
}))
.await
.unwrap();
// The caller hangs up. Dropping the response drops the receiving half of
// the channel, which is all a cancelled RPC amounts to on this side.
drop(response);
// Give the copy longer than it would need to finish, and hold it to the
// assertion throughout rather than only at the end.
for _ in 0..100 {
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
let store = dest_service.state.store.read().unwrap();
assert!(
store.find_volume(VolumeId(1)).is_none(),
"destination mounted a volume whose copy was abandoned"
);
}
for ext in [".dat", ".idx", ".vif", ".note"] {
assert!(
!std::path::Path::new(&dest_file(ext)).exists(),
"abandoned copy left {} behind",
dest_file(ext)
);
}
}
// If the caller departs in the narrow window between the pre-mount
// is_closed() check and the final tx.send, the volume is already mounted
// when the send fails. The task must roll back that mount — not just unlink
// the files — or the destination carries the volume's index cache for the
// life of the process.
//
// Reaching that window black-box needs a seam. The one the code already
// offers is the store write lock: the test takes it immediately after
// volume_copy returns, so the spawned task runs the copy to completion with
// the caller still attached, passes the pre-mount is_closed() check, then
// parks entering the mount block (state.store.write() blocks on the test's
// guard). The test then drops the response — the caller is gone — and
// releases the guard. The task acquires the lock, mounts, fails the final
// tx.send (receiver dropped), and must roll back via the error branch's
// delete_volume. Without the lock seam the task sees is_closed() at its
// very first check and returns before mount_volume, exercising the wrong
// path — the test would be green for the wrong reason.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_volume_copy_after_mount_cancellation_rolls_back_mount() {
let (source_service, _source_tmp, _dat_bytes) = make_local_service_with_large_volume();
let (port, _shutdown) = serve_source(source_service).await;
let (dest_service, dest_tmp) = make_local_service_with_volume("", None);
{
let mut store = dest_service.state.store.write().unwrap();
store.delete_volume(VolumeId(1), false, false).unwrap();
for loc in &store.locations {
loc.check_disk_space();
}
}
let dest_dir = dest_tmp.path().to_str().unwrap().to_string();
let dest_file = |ext: &str| format!("{}/1{}", dest_dir, ext);
let response = dest_service
.volume_copy(Request::new(volume_server_pb::VolumeCopyRequest {
volume_id: 1,
collection: String::new(),
source_data_node: format!("127.0.0.1:1.{}", port),
disk_type: String::new(),
io_byte_per_second: 0,
replication: String::new(),
ttl: String::new(),
}))
.await
.unwrap();
// Hold the store write lock so the task parks at the mount block after
// completing the copy and passing the pre-mount is_closed() check.
let store_guard = dest_service.state.store.write().unwrap();
// Give the copy time to complete and the task to reach the mount block.
// The 5 MB fixture copies in well under a second on a local loopback
// gRPC connection; 2 s is generous and keeps the test fast.
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
// The caller hangs up while the task is parked at the mount block.
drop(response);
// Release the store lock so the task can enter the mount block, mount,
// fail the final send, and roll back.
drop(store_guard);
// Wait for the rollback to finish.
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
let store = dest_service.state.store.read().unwrap();
assert!(
store.find_volume(VolumeId(1)).is_none(),
"destination still holds a mounted volume after after-mount cancellation"
);
drop(store);
for ext in [".dat", ".idx", ".vif", ".note"] {
assert!(
!std::path::Path::new(&dest_file(ext)).exists(),
"abandoned copy left {} behind",
dest_file(ext)
);
}
}
// copy_file must stream the whole .dat in 2MB chunks (not buffer it) and
// reassemble byte-for-byte, with the mtime carried only on the first message.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
test/volume_server/grpc/copy_sync_test.go
+98
View File
@@ -3,6 +3,7 @@ package volume_server_grpc_test
import (
"context"
"io"
"math/rand"
"net/http"
"strings"
"testing"
@@ -295,3 +296,100 @@ func TestVolumeCopyOverwritesExistingDestinationVolume(t *testing.T) {
t.Fatalf("destination post-copy payload mismatch: got %q want %q", string(destReadAfterBody), string(sourcePayload))
}
}
// A VolumeCopy whose caller goes away must not leave a mounted volume behind on
// the destination. weed-admin's batch balance routinely starts far more copies
// than it finishes, and each abandoned copy that still mounts costs the
// destination a volume it was never asked to hold: a per-volume index cache that
// is never reclaimed, and — under replication=000 — a second writable copy of a
// volume id that two writers can diverge.
//
// The copy is throttled so it is demonstrably still in flight when the caller
// cancels; the assertion before the cancel keeps the test from passing
// vacuously if the throttle ever stops biting.
func TestVolumeCopyCancelledByCallerDoesNotMountDestination(t *testing.T) {
if testing.Short() {
t.Skip("skipping integration test in short mode")
}
clusterHarness := framework.StartDualVolumeCluster(t, matrix.P1())
sourceConn, sourceClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress(0))
defer sourceConn.Close()
destConn, destClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress(1))
defer destConn.Close()
const volumeID = uint32(44)
const payloadMiB = 192
framework.AllocateVolume(t, sourceClient, volumeID, "")
// The fixture size is load-bearing twice over; do not shrink it to save CI
// time without reading both reasons.
//
// First, the copy has to still be running when the caller cancels, and the
// only lever for that is IoBytePerSecond. It is a no-op below ~100ms of wall
// clock — the throttler never re-checks within its first window — and on a
// tmpfs loopback cluster 64 MiB copies in ~110ms.
//
// Second, 192 MiB is above the 128 MiB progress-report interval, which is
// what lets this test be green on Go: a failing stream.Send from that report
// is Go's only abort signal. Measured below the interval (120 MiB), a Go
// destination mounts the abandoned copy too, so a smaller fixture would fail
// the Go leg without any Rust change.
//
// The bytes are pseudo-random because the volume server gzips compressible
// uploads; a repeating payload lands on disk as a few KB and the throttle
// never bites.
httpClient := framework.NewHTTPClient()
chunk := make([]byte, 1024*1024)
if _, err := rand.New(rand.NewSource(11186)).Read(chunk); err != nil {
t.Fatalf("generate payload: %v", err)
}
for i := 0; i < payloadMiB; i++ {
fid := framework.NewFileID(volumeID, uint64(880100+i), 0x3456789a)
uploadResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(0), fid, chunk)
_ = framework.ReadAllAndClose(t, uploadResp)
if uploadResp.StatusCode != http.StatusCreated {
t.Fatalf("upload %d to source expected 201, got %d", i, uploadResp.StatusCode)
}
}
// 192 MiB at 16 MiB/s is about twelve seconds of copying.
copyCtx, cancelCopy := context.WithCancel(context.Background())
defer cancelCopy()
_, err := destClient.VolumeCopy(copyCtx, &volume_server_pb.VolumeCopyRequest{
VolumeId: volumeID,
Collection: "",
SourceDataNode: clusterHarness.VolumeAdminAddress(0) + "." + strings.Split(clusterHarness.VolumeGRPCAddress(0), ":")[1],
IoBytePerSecond: 16 * 1024 * 1024,
})
if err != nil {
t.Fatalf("VolumeCopy start failed: %v", err)
}
time.Sleep(2 * time.Second)
// Premise: the copy must still be running when we cancel. If the
// destination has already mounted, the throttle no longer bites and the
// rest of this test would prove nothing.
premiseCtx, cancelPremise := context.WithTimeout(context.Background(), 5*time.Second)
_, premiseErr := destClient.ReadVolumeFileStatus(premiseCtx, &volume_server_pb.ReadVolumeFileStatusRequest{VolumeId: volumeID})
cancelPremise()
if premiseErr == nil {
t.Fatalf("copy of volume %d already completed before the cancel; raise payloadMiB or lower IoBytePerSecond", volumeID)
}
cancelCopy()
// Well past the ~10s the unabandoned copy would have needed: the volume
// must never appear on the destination.
deadline := time.Now().Add(25 * time.Second)
for time.Now().Before(deadline) {
pollCtx, cancelPoll := context.WithTimeout(context.Background(), 5*time.Second)
_, statusErr := destClient.ReadVolumeFileStatus(pollCtx, &volume_server_pb.ReadVolumeFileStatusRequest{VolumeId: volumeID})
cancelPoll()
if statusErr == nil {
t.Fatalf("destination mounted volume %d after its VolumeCopy caller cancelled", volumeID)
}
time.Sleep(500 * time.Millisecond)
}
}