diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index bacd83185..5f5ccacef 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -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( - &grpc_addr, - state.outgoing_grpc_tls.as_ref(), - ) - .await - { + // 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(), + ) => 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( ext: &str, is_append: bool, ignore_source_not_found: bool, - progress_tx: Option< - &tokio::sync::mpsc::Sender>, - >, + progress_tx: &tokio::sync::mpsc::Sender>, + report_progress: bool, next_report_target: &mut i64, report_interval: i64, throttler: &mut WriteThrottler, -) -> Result +) -> Result where T: tonic::client::GrpcService, T::Error: Into, @@ -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!( - "failed to start copying volume {} {} file: {}", - volume_id, ext, e - ) - })? - .into_inner(); + // 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() + } + _ = 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 - .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 +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, + >(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)] diff --git a/test/volume_server/grpc/copy_sync_test.go b/test/volume_server/grpc/copy_sync_test.go index 810395fb6..f47fa186b 100644 --- a/test/volume_server/grpc/copy_sync_test.go +++ b/test/volume_server/grpc/copy_sync_test.go @@ -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) + } +}