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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
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 <noreply@anthropic.com>

* rust volume: report an HTTP task panic as a server error

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>

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<String> = 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;
     }

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)
+		}
 	}
 }
 

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]),

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]),

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 != "" {

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 != "" {

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)

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{