TOOLS-4148 Add tests for SIGPIPE handling, remove equivalent JS tests, and fix handling for other signals

bsondump/bsondump_test.go

@@ -348,6 +348,28 @@ func runBsondumpWithLargeFile(t *testing.T, size int) (string, error) {
 	return runBsondump("--bsonFile", bsonFile, "--outFile", outFile)
 }
 
+// TestBrokenPipe verifies that bsondump handles a broken pipe gracefully
+// (exits with a write error rather than being killed by SIGPIPE).
+func TestBrokenPipe(t *testing.T) {
+	testtype.SkipUnlessTestType(t, testtype.UnitTestType)
+
+	// Write 1000 docs (~1 KB each) so the JSON output exceeds the pipe buffer.
+	dir := t.TempDir()
+	bsonPath := filepath.Join(dir, "large.bson")
+	f, err := os.Create(bsonPath)
+	require.NoError(t, err)
+	for i := range 1000 {
+		doc := bson.D{{"_id", int32(i)}, {"data", strings.Repeat("x", 1000)}}
+		raw, err := bson.Marshal(doc)
+		require.NoError(t, err)
+		_, err = f.Write(raw)
+		require.NoError(t, err)
+	}
+	require.NoError(t, f.Close())
+
+	testutil.AssertBrokenPipeHandled(t, bsondumpCommand(bsonPath))
+}
+
 func runBsondump(args ...string) (string, error) {
 	cmd := []string{"go", "run", filepath.Join("..", "bsondump", "main")}
 	cmd = append(cmd, args...)

common/signals/signals.go

@@ -21,24 +21,23 @@ func Handle() chan struct{} {
 	return HandleWithInterrupt(nil)
 }
 
-// HandleWithInterrupt starts a goroutine which listens for SIGTERM, SIGINT, and
-// SIGKILL and explicitly ignores SIGPIPE. It calls the finalizer function when
-// the first signal is received and forcibly terminates the program after the
-// second. If a nil function is provided, the program will exit after the first
-// signal.
+// HandleWithInterrupt starts a goroutine which listens for SIGTERM, SIGINT, and SIGKILL. It also
+// calles signal.Ignore to explicitly ignore SIGPIPE. It calls the finalizer function when the first
+// signal is received and forcibly terminates the program after the second. If a nil function is
+// provided, the program will exit after the first signal.
 func HandleWithInterrupt(finalizer func()) chan struct{} {
+	// Ignore SIGPIPE synchronously so there is no race between this call and the first write: a
+	// broken pipe must surface as an EPIPE write error, not kill the process.
+	signal.Ignore(syscall.SIGPIPE)
+
 	finishedChan := make(chan struct{})
 	go handleSignals(finalizer, finishedChan)
 	return finishedChan
 }
 
 func handleSignals(finalizer func(), finishedChan chan struct{}) {
-	// explicitly ignore SIGPIPE; the tools should deal with write errors
-	noopChan := make(chan os.Signal)
-	signal.Notify(noopChan, syscall.SIGPIPE)
-
 	log.Logv(log.DebugLow, "will listen for SIGTERM, SIGINT, and SIGKILL")
-	sigChan := make(chan os.Signal, 2)
+	sigChan := make(chan os.Signal, 1)
 	signal.Notify(sigChan, syscall.SIGTERM, syscall.SIGINT)
 	defer signal.Stop(sigChan)
 	if finalizer != nil {

common/signals/signals_test.go

@@ -0,0 +1,46 @@
+// Copyright (C) MongoDB, Inc. 2014-present.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License. You may obtain
+// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+
+package signals
+
+import (
+	"fmt"
+	"os"
+	"os/exec"
+	"testing"
+
+	"github.com/mongodb/mongo-tools/common/testutil"
+)
+
+// TestMain handles the subprocess mode used by TestBrokenPipeHandledAsWriteError.
+func TestMain(m *testing.M) {
+	if os.Getenv("TEST_BROKEN_PIPE_SUBPROCESS") == "1" {
+		// Set up SIGPIPE handling as our tools do.
+		done := HandleWithInterrupt(nil)
+		defer close(done)
+
+		// Write repeatedly to stdout. When the reader closes the pipe,
+		// the write will return an EPIPE error. We exit 0 to signal that
+		// the error was surfaced and handled — not a SIGPIPE death.
+		for {
+			_, err := fmt.Println("data")
+			if err != nil {
+				os.Exit(0)
+			}
+		}
+	}
+	os.Exit(m.Run())
+}
+
+// TestBrokenPipeHandledAsWriteError verifies that when the read end of a pipe
+// is closed, our signal handler causes the write error to surface as an EPIPE
+// (allowing the tool to exit cleanly) rather than the process being killed by
+// SIGPIPE.
+func TestBrokenPipeHandledAsWriteError(t *testing.T) {
+	cmd := exec.Command(os.Args[0], "-test.run=TestBrokenPipeHandledAsWriteError")
+	cmd.Env = append(os.Environ(), "TEST_BROKEN_PIPE_SUBPROCESS=1")
+	testutil.AssertBrokenPipeHandled(t, cmd)
+}

common/testutil/broken_pipe_unix.go

@@ -0,0 +1,60 @@
+// Copyright (C) MongoDB, Inc. 2014-present.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License. You may obtain
+// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+
+//go:build !windows
+
+package testutil
+
+import (
+	"os"
+	"os/exec"
+	"syscall"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// AssertBrokenPipeHandled starts cmd with its stdout connected to a pipe,
+// reads a small amount of output to confirm the process has started writing,
+// then breaks the pipe by closing the read end. It asserts the process was
+// not killed by SIGPIPE — a broken pipe must surface as a write error that
+// the process can handle cleanly.
+//
+// cmd.Stdout must not be set before calling; this function sets it to the
+// write end of a pipe.
+func AssertBrokenPipeHandled(t *testing.T, cmd *exec.Cmd) {
+	t.Helper()
+
+	pr, pw, err := os.Pipe()
+	require.NoError(t, err)
+	cmd.Stdout = pw
+
+	require.NoError(t, cmd.Start())
+	require.NoError(t, pw.Close())
+
+	buf := make([]byte, 64)
+	_, err = pr.Read(buf)
+	require.NoError(t, err, "process should write at least some output before the pipe breaks")
+	require.NoError(t, pr.Close())
+
+	err = cmd.Wait()
+	if err == nil {
+		return
+	}
+
+	var exitErr *exec.ExitError
+	require.ErrorAs(t, err, &exitErr, "non-zero exit from cmd.Wait() should always be an ExitError")
+	status, ok := exitErr.Sys().(syscall.WaitStatus)
+	require.True(t, ok, "exitErr.Sys() should return a syscall.WaitStatus")
+	if status.Signaled() {
+		require.NotEqual(
+			t,
+			syscall.SIGPIPE,
+			status.Signal(),
+			"process should not be killed by SIGPIPE: broken pipe should surface as a write error",
+		)
+	}
+}

common/testutil/broken_pipe_windows.go

@@ -0,0 +1,19 @@
+// Copyright (C) MongoDB, Inc. 2014-present.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License. You may obtain
+// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+
+//go:build windows
+
+package testutil
+
+import (
+	"os/exec"
+	"testing"
+)
+
+// AssertBrokenPipeHandled is a no-op on Windows where SIGPIPE does not apply.
+func AssertBrokenPipeHandled(t *testing.T, cmd *exec.Cmd) {
+	t.Skip("broken pipe handling via SIGPIPE is not applicable on Windows")
+}

mongodump/mongodump_test.go

@@ -9,12 +9,14 @@ package mongodump
 import (
 	"bytes"
 	"compress/gzip"
+	"context"
 	"crypto/sha1"
 	"encoding/base64"
 	"fmt"
 	"io"
 	"math/rand"
 	"os"
+	"os/exec"
 	"path/filepath"
 	"regexp"
 	"strings"
@@ -2517,3 +2519,37 @@ func dumpAndCheckPipelineOrder(t *testing.T, collName string, pipeline bson.A) {
 	os.RemoveAll(dumpDir)
 	metaFile.Close()
 }
+
+// TestBrokenPipe verifies that mongodump handles a broken pipe gracefully
+// (exits with a write error rather than being killed by SIGPIPE).
+func TestBrokenPipe(t *testing.T) {
+	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)
+
+	const (
+		dbName   = "mongodump_broken_pipe_test"
+		collName = "docs"
+	)
+
+	sessionProvider, _, err := testutil.GetBareSessionProvider()
+	require.NoError(t, err)
+	client, err := sessionProvider.GetSession()
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		_ = client.Database(dbName).Drop(context.Background())
+	})
+
+	// Insert 2000 docs so the archive output exceeds the pipe buffer.
+	docs := make([]any, 2000)
+	for i := range 2000 {
+		docs[i] = bson.D{{"_id", int32(i)}, {"data", strings.Repeat("x", 500)}}
+	}
+	_, err = client.Database(dbName).Collection(collName).InsertMany(context.Background(), docs)
+	require.NoError(t, err)
+
+	args := append(
+		[]string{"run", filepath.Join("..", "mongodump", "main")},
+		testutil.GetBareArgs()...,
+	)
+	args = append(args, "--db", dbName, "--archive=-")
+	testutil.AssertBrokenPipeHandled(t, exec.Command("go", args...))
+}

mongoexport/mongoexport_test.go

@@ -8,12 +8,16 @@ package mongoexport
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
 	"os"
+	"os/exec"
+	"path/filepath"
 	"runtime"
+	"strings"
 	"testing"
 
 	"github.com/mongodb/mongo-tools/common/bsonutil"
@@ -312,3 +316,37 @@ func TestBadOptions(t *testing.T) {
 		}
 	}
 }
+
+// TestBrokenPipe verifies that mongoexport handles a broken pipe gracefully
+// (exits with a write error rather than being killed by SIGPIPE).
+func TestBrokenPipe(t *testing.T) {
+	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)
+
+	const (
+		dbName   = "mongoexport_broken_pipe_test"
+		collName = "docs"
+	)
+
+	sessionProvider, _, err := testutil.GetBareSessionProvider()
+	require.NoError(t, err)
+	client, err := sessionProvider.GetSession()
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		_ = client.Database(dbName).Drop(context.Background())
+	})
+
+	// Insert 1000 docs with a large field so JSON output exceeds the pipe buffer.
+	docs := make([]any, 1000)
+	for i := range 1000 {
+		docs[i] = bson.D{{"_id", int32(i)}, {"data", strings.Repeat("x", 1000)}}
+	}
+	_, err = client.Database(dbName).Collection(collName).InsertMany(context.Background(), docs)
+	require.NoError(t, err)
+
+	args := append(
+		[]string{"run", filepath.Join("..", "mongoexport", "main")},
+		testutil.GetBareArgs()...,
+	)
+	args = append(args, "--db", dbName, "--collection", collName)
+	testutil.AssertBrokenPipeHandled(t, exec.Command("go", args...))
+}