mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-11 17:10:40 +02:00
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
This commit is contained in:
franchb
@@ -1397,6 +1397,16 @@ impl VolumeServer for VolumeGrpcService {
|
||||
|
||||
tokio::spawn(async move {
|
||||
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 {
|
||||
@@ -1454,7 +1464,8 @@ impl VolumeServer for VolumeGrpcService {
|
||||
".dat",
|
||||
false,
|
||||
true,
|
||||
Some(&tx),
|
||||
&tx,
|
||||
true,
|
||||
&mut next_report_target,
|
||||
report_interval,
|
||||
&mut throttler,
|
||||
@@ -1479,7 +1490,8 @@ impl VolumeServer for VolumeGrpcService {
|
||||
".idx",
|
||||
false,
|
||||
false,
|
||||
None,
|
||||
&tx,
|
||||
false,
|
||||
&mut next_report_target,
|
||||
report_interval,
|
||||
&mut throttler,
|
||||
@@ -1503,7 +1515,8 @@ impl VolumeServer for VolumeGrpcService {
|
||||
".vif",
|
||||
false,
|
||||
true,
|
||||
None,
|
||||
&tx,
|
||||
false,
|
||||
&mut next_report_target,
|
||||
report_interval,
|
||||
&mut throttler,
|
||||
@@ -1546,6 +1559,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();
|
||||
@@ -1570,6 +1595,24 @@ impl VolumeServer for VolumeGrpcService {
|
||||
.await;
|
||||
|
||||
if let Err(e) = result {
|
||||
// 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
|
||||
let _ = std::fs::remove_file(format!("{}.dat", data_base_name));
|
||||
let _ = std::fs::remove_file(format!("{}.idx", idx_base_name));
|
||||
@@ -5009,9 +5052,8 @@ 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,
|
||||
@@ -5060,6 +5102,7 @@ where
|
||||
|
||||
let mut progressed_bytes: i64 = 0;
|
||||
let mut modified_ts_ns: i64 = 0;
|
||||
let cancelled = || format!("volume {} {} copy cancelled by caller", volume_id, ext);
|
||||
|
||||
while let Some(resp) = stream
|
||||
.message()
|
||||
@@ -5070,24 +5113,39 @@ where
|
||||
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))?;
|
||||
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
|
||||
.send(Ok(volume_server_pb::VolumeCopyResponse {
|
||||
last_append_at_ns: 0,
|
||||
processed_bytes: progressed_bytes,
|
||||
}))
|
||||
.await;
|
||||
*next_report_target = progressed_bytes + report_interval;
|
||||
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
|
||||
.is_err()
|
||||
{
|
||||
return Err(cancelled());
|
||||
}
|
||||
*next_report_target = progressed_bytes + report_interval;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -5906,6 +5964,165 @@ 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");
|
||||
assert!(
|
||||
err.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)
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// 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)]
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user