Add a new test for our SIGPIPE handling in the common code

This test makes sure that we ignore SIGPIPE. Instead, writes to a broken pipe will cause the write call to return an error. The tools should always check errors on write calls, so these will be surfaced as normal "failure to write to an fd" errors, as opposed to the signal aborting the process.

Author: autarch

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