From 3b7706bad8fb0deeac09e2eee48b10043163d05c Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 3 Nov 2026 10:44:01 -0800 Subject: [PATCH] rust volume: exit cleanly on SIGTERM, and make the harness notice when a server does not (#12370) * rust volume: exit cleanly on SIGTERM and SIGINT The shutdown path raced the HTTP and gRPC tasks with select!, then awaited both handles again; tokio panics when a JoinHandle is polled after it completed, so every signal ended in "JoinHandle polled after completion" with exit status 101 and the volumes never closed. Waiting for both tasks with join! is enough: both stop on the shutdown signal, and a gRPC serve error sends it. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * test/volume_server: fail when a stopped server does not exit cleanly The harness ignored the exit status after SIGINT, which is how the Rust volume server's shutdown panic went unseen. Every server it stops must now exit 0 within 10s. The Go volume servers get -preStopSeconds=0, which the Rust ones already had, so a stop does not spend the default 10s grace inside that window. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * test/volume_server: let assertArgPair find flags after a bare flag rustVolumeArgs carries --volume.allowUntrustedRemoteEndpoints without a value, so pairs no longer sit at even indexes and TestRustVolumeArgsIncludeReadMode reported --readMode missing although it was there. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * rust volume: stop both servers when either one exits A gRPC task that panicked left the HTTP server and heartbeat running until a signal arrived. Whichever server task ends first now sends the shutdown and the other is awaited once, matching Go where a serve failure is fatal. Co-Authored-By: Claude Opus 5.5 * rust volume: report an HTTP task panic as a server error Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Fable 5.1 --- seaweed-volume/src/main.rs | 35 +++++++++---------- test/volume_server/framework/cluster.go | 20 +++++++---- test/volume_server/framework/cluster_mixed.go | 5 +-- test/volume_server/framework/cluster_multi.go | 5 +-- .../framework/cluster_multi_rust.go | 4 +-- test/volume_server/framework/cluster_rust.go | 4 +-- .../framework/cluster_rust_test.go | 4 ++- .../framework/cluster_with_filer.go | 4 +-- 8 files changed, 46 insertions(+), 35 deletions(-) diff --git a/seaweed-volume/src/main.rs b/seaweed-volume/src/main.rs index 4e884fe0dd..df515ebcd3 100644 --- a/seaweed-volume/src/main.rs +++ b/seaweed-volume/src/main.rs @@ -845,40 +845,39 @@ async fn run( })) }; - // Wait for servers. Use select! with &mut so the losing handle is not - // dropped, then await it explicitly afterward. + // Whichever server stops first takes the other down, as a serve failure + // is fatal for the Go volume server. let mut server_err: Option = None; - let mut http_handle = http_handle; let mut grpc_handle = grpc_handle; - let grpc_finished_first = tokio::select! { - _ = &mut http_handle => false, - _ = &mut grpc_handle => true, - }; - // Inspect the gRPC result (already resolved if it finished first, - // otherwise await it now). - let grpc_result = if grpc_finished_first { - grpc_handle.await - } else { - // HTTP finished first; gRPC is still running. Await it. - grpc_handle.await + let mut http_handle = http_handle; + let (http_result, grpc_result) = tokio::select! { + r = &mut grpc_handle => { + let _ = shutdown_tx.send(()); + (http_handle.await, r) + } + r = &mut http_handle => { + let _ = shutdown_tx.send(()); + (r, grpc_handle.await) + } }; + if let Err(e) = http_result { + let msg = format!("HTTP task panicked: {}", e); + error!("{}", msg); + server_err = Some(msg); + } match grpc_result { Ok(Ok(())) => {} Ok(Err(e)) => { let msg = format!("gRPC server exited with error: {}", e); error!("{}", msg); server_err = Some(msg); - // serve error already sent shutdown inside the task } Err(e) => { let msg = format!("gRPC task panicked: {}", e); error!("{}", msg); server_err = Some(msg); - let _ = shutdown_tx.send(()); } } - // Ensure the HTTP handle completes too. - let _ = http_handle.await; if let Some(h) = public_handle { let _ = h.await; } diff --git a/test/volume_server/framework/cluster.go b/test/volume_server/framework/cluster.go index a134250f5f..6b5baa512b 100644 --- a/test/volume_server/framework/cluster.go +++ b/test/volume_server/framework/cluster.go @@ -186,8 +186,8 @@ func (c *Cluster) Stop() { return } c.cleanupOnce.Do(func() { - stopProcess(c.volumeCmd) - stopProcess(c.masterCmd) + stopProcess(c.testingTB, "volume server", c.volumeCmd) + stopProcess(c.testingTB, "master", c.masterCmd) if !c.keepLogs && !c.testingTB.Failed() { _ = os.RemoveAll(c.baseDir) } else if c.baseDir != "" { @@ -202,7 +202,7 @@ func (c *Cluster) Stop() { // a second-run startup failure. func (c *Cluster) RestartVolumeServer() { c.testingTB.Helper() - stopProcess(c.volumeCmd) + stopProcess(c.testingTB, "volume server", c.volumeCmd) c.volumeCmd = nil // Rotate the log; absent on the first call after a clean start, which // is fine — startVolume will create it. Any real filesystem failure @@ -224,7 +224,7 @@ func (c *Cluster) RestartVolumeServer() { // dirs alone. Pair with RestartVolumeServer or Stop. func (c *Cluster) StopVolumeServer() { c.testingTB.Helper() - stopProcess(c.volumeCmd) + stopProcess(c.testingTB, "volume server", c.volumeCmd) c.volumeCmd = nil } @@ -267,6 +267,7 @@ func (c *Cluster) startVolume(dataDirs []string) error { "-config_dir=" + c.configDir, "volume", "-ip=127.0.0.1", + "-preStopSeconds=0", "-port=" + strconv.Itoa(c.volumePort), "-port.grpc=" + strconv.Itoa(c.volumeGrpcPort), "-port.public=" + strconv.Itoa(c.volumePubPort), @@ -325,7 +326,10 @@ func (c *Cluster) waitForTCP(addr string) error { return fmt.Errorf("timed out waiting for tcp %s", addr) } -func stopProcess(cmd *exec.Cmd) { +// stopProcess interrupts cmd and waits for it to exit, killing it after 10s. +// A server that does not exit cleanly on SIGINT fails the test: a panic or a +// hang on shutdown is a bug the tests would otherwise never see. +func stopProcess(tb testing.TB, name string, cmd *exec.Cmd) { if cmd == nil || cmd.Process == nil { return } @@ -340,7 +344,11 @@ func stopProcess(cmd *exec.Cmd) { case <-time.After(10 * time.Second): _ = cmd.Process.Kill() <-done - case <-done: + tb.Errorf("%s did not exit within 10s of SIGINT", name) + case err := <-done: + if err != nil { + tb.Errorf("%s did not exit cleanly on SIGINT: %v", name, err) + } } } diff --git a/test/volume_server/framework/cluster_mixed.go b/test/volume_server/framework/cluster_mixed.go index 30a850223c..6da9dde4aa 100644 --- a/test/volume_server/framework/cluster_mixed.go +++ b/test/volume_server/framework/cluster_mixed.go @@ -195,9 +195,9 @@ func (c *MixedVolumeCluster) Stop() { } c.cleanupOnce.Do(func() { for i := len(c.volumeCmds) - 1; i >= 0; i-- { - stopProcess(c.volumeCmds[i]) + stopProcess(c.testingTB, fmt.Sprintf("volume server %d", i), c.volumeCmds[i]) } - stopProcess(c.masterCmd) + stopProcess(c.testingTB, "master", c.masterCmd) if !c.keepLogs && !c.testingTB.Failed() { _ = os.RemoveAll(c.baseDir) } else if c.baseDir != "" { @@ -247,6 +247,7 @@ func (c *MixedVolumeCluster) startGoVolume(index int, dataDir string) error { "-config_dir=" + c.configDir, "volume", "-ip=127.0.0.1", + "-preStopSeconds=0", "-port=" + strconv.Itoa(c.volumePorts[index]), "-port.grpc=" + strconv.Itoa(c.volumeGrpcPorts[index]), "-port.public=" + strconv.Itoa(c.volumePubPorts[index]), diff --git a/test/volume_server/framework/cluster_multi.go b/test/volume_server/framework/cluster_multi.go index 9076262d83..5956d686f6 100644 --- a/test/volume_server/framework/cluster_multi.go +++ b/test/volume_server/framework/cluster_multi.go @@ -198,9 +198,9 @@ func (c *MultiVolumeCluster) Stop() { c.cleanupOnce.Do(func() { // Stop volume servers in reverse order for i := len(c.volumeCmds) - 1; i >= 0; i-- { - stopProcess(c.volumeCmds[i]) + stopProcess(c.testingTB, fmt.Sprintf("volume server %d", i), c.volumeCmds[i]) } - stopProcess(c.masterCmd) + stopProcess(c.testingTB, "master", c.masterCmd) if !c.keepLogs && !c.testingTB.Failed() { _ = os.RemoveAll(c.baseDir) } else if c.baseDir != "" { @@ -249,6 +249,7 @@ func (c *MultiVolumeCluster) startVolume(index int, dataDirs []string) error { "-config_dir=" + c.configDir, "volume", "-ip=127.0.0.1", + "-preStopSeconds=0", "-port=" + strconv.Itoa(c.volumePorts[index]), "-port.grpc=" + strconv.Itoa(c.volumeGrpcPorts[index]), "-port.public=" + strconv.Itoa(c.volumePubPorts[index]), diff --git a/test/volume_server/framework/cluster_multi_rust.go b/test/volume_server/framework/cluster_multi_rust.go index dc2515ebdd..79f773b8b1 100644 --- a/test/volume_server/framework/cluster_multi_rust.go +++ b/test/volume_server/framework/cluster_multi_rust.go @@ -178,9 +178,9 @@ func (c *RustMultiVolumeCluster) Stop() { c.cleanupOnce.Do(func() { // Stop volume servers in reverse order for i := len(c.volumeCmds) - 1; i >= 0; i-- { - stopProcess(c.volumeCmds[i]) + stopProcess(c.testingTB, fmt.Sprintf("volume server %d", i), c.volumeCmds[i]) } - stopProcess(c.masterCmd) + stopProcess(c.testingTB, "master", c.masterCmd) if !c.keepLogs && !c.testingTB.Failed() { _ = os.RemoveAll(c.baseDir) } else if c.baseDir != "" { diff --git a/test/volume_server/framework/cluster_rust.go b/test/volume_server/framework/cluster_rust.go index 88d594d739..291b82ae88 100644 --- a/test/volume_server/framework/cluster_rust.go +++ b/test/volume_server/framework/cluster_rust.go @@ -148,8 +148,8 @@ func (rc *RustCluster) Stop() { return } rc.cleanupOnce.Do(func() { - stopProcess(rc.volumeCmd) - stopProcess(rc.masterCmd) + stopProcess(rc.testingTB, "rust volume server", rc.volumeCmd) + stopProcess(rc.testingTB, "master", rc.masterCmd) if !rc.keepLogs && !rc.testingTB.Failed() { _ = os.RemoveAll(rc.baseDir) } else if rc.baseDir != "" { diff --git a/test/volume_server/framework/cluster_rust_test.go b/test/volume_server/framework/cluster_rust_test.go index f2558753a8..1d3f05a768 100644 --- a/test/volume_server/framework/cluster_rust_test.go +++ b/test/volume_server/framework/cluster_rust_test.go @@ -26,7 +26,9 @@ func TestRustVolumeArgsIncludeReadMode(t *testing.T) { func assertArgPair(t *testing.T, args []string, flag string, want string) { t.Helper() - for i := 0; i+1 < len(args); i += 2 { + // Not every flag takes a value (--volume.allowUntrustedRemoteEndpoints), + // so pairs do not line up at even indexes. + for i := 0; i+1 < len(args); i++ { if args[i] == flag { if args[i+1] != want { t.Fatalf("%s value mismatch: got %q want %q", flag, args[i+1], want) diff --git a/test/volume_server/framework/cluster_with_filer.go b/test/volume_server/framework/cluster_with_filer.go index eaaa4a20f4..6dd58229ed 100644 --- a/test/volume_server/framework/cluster_with_filer.go +++ b/test/volume_server/framework/cluster_with_filer.go @@ -63,12 +63,12 @@ func StartSingleVolumeClusterWithFiler(t testing.TB, profile matrix.Profile) *Cl if err = baseCluster.waitForTCP(net.JoinHostPort("127.0.0.1", strconv.Itoa(filerGrpcPort))); err != nil { filerLogTail := baseCluster.tailLog("filer.log") - stopProcess(filerCmd) + stopProcess(t, "filer", filerCmd) t.Fatalf("wait for filer grpc readiness: %v\nfiler log tail:\n%s", err, filerLogTail) } t.Cleanup(func() { - stopProcess(filerCmd) + stopProcess(t, "filer", filerCmd) }) return &ClusterWithFiler{