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{