fix: do not stop accepting connections on transient errors

internal/cli/run_proxy.go

@@ -346,13 +346,47 @@ func runProxy(conf *config.Config, version string) error { //nolint: funlen, cyc
 		}
 	}
 
-	ctx := utils.RootContext()
+	ctx, cancel := context.WithCancel(utils.RootContext())
+	defer cancel()
 
-	go proxy.Serve(listener) //nolint: errcheck
+	// If the accept loop stops for any reason other than shutdown, the process
+	// must not stay alive without accepting connections: stop and report the
+	// error, so a supervisor (systemd, docker) restarts it.
+	serveErr := make(chan error, 1)
 
+	go func() {
+		serveErr <- proxy.Serve(listener)
+
+		cancel()
+	}()
+
+	return waitAndShutdown(ctx, serveErr, func() {
+		listener.Close() //nolint: errcheck
+		proxy.Shutdown()
+	})
+}
+
+// waitAndShutdown waits for shutdown, stops the proxy and decides whether the
+// accept loop has failed. A Serve that failed on its own puts its error into
+// the channel before calling cancel(), so the error is already there when ctx
+// is done. On an ordinary stop (SIGTERM) Serve is still running; the error it
+// returns because stop() closes the listener is not a failure, otherwise every
+// stop or restart would exit with code 1.
+func waitAndShutdown(ctx context.Context, serveErr <-chan error, stop func()) error {
 	<-ctx.Done()
-	listener.Close() //nolint: errcheck
-	proxy.Shutdown()
+
+	var serveFailure error
+
+	select {
+	case serveFailure = <-serveErr:
+	default:
+	}
+
+	stop()
+
+	if serveFailure != nil {
+		return fmt.Errorf("proxy stopped accepting connections: %w", serveFailure)
+	}
 
 	return nil
 }

internal/cli/run_proxy_shutdown_test.go

@@ -0,0 +1,43 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// An ordinary stop closes the listener, which wakes Serve up with an accept
+// error. That must not turn into exit code 1.
+func TestWaitAndShutdownOrdinaryStopIsClean(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	serveErr := make(chan error, 1)
+
+	cancel() // SIGTERM
+
+	err := waitAndShutdown(ctx, serveErr, func() {
+		serveErr <- errors.New("use of closed network connection")
+	})
+	require.NoError(t, err)
+}
+
+func TestWaitAndShutdownReportsServeFailure(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	serveErr := make(chan error, 1)
+	stopped := false
+
+	// Serve failed on its own: the error is sent first, then cancel().
+	serveErr <- errors.New("too many open files")
+	cancel()
+
+	err := waitAndShutdown(ctx, serveErr, func() { stopped = true })
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "too many open files")
+	assert.True(t, stopped)
+}

internal/utils/net_listener.go

@@ -11,19 +11,25 @@ type Listener struct {
 	net.Listener
 }
 
+// Accept returns the next connection with client socket options applied. A
+// connection whose options cannot be set (typically it was reset by the client
+// right after the handshake) is closed and skipped: returning an error here
+// would stop the whole accept loop because of one bad client.
 func (l Listener) Accept() (net.Conn, error) {
-	conn, err := l.Listener.Accept()
-	if err != nil {
-		return nil, err //nolint: wrapcheck
-	}
+	for {
+		conn, err := l.Listener.Accept()
+		if err != nil {
+			return nil, err //nolint: wrapcheck
+		}
 
-	if err := network.SetClientSocketOptions(conn, 0); err != nil {
-		conn.Close() //nolint: errcheck
+		if err := network.SetClientSocketOptions(conn, 0); err != nil {
+			conn.Close() //nolint: errcheck
 
-		return nil, fmt.Errorf("cannot set TCP options: %w", err)
-	}
+			continue
+		}
 
-	return conn, nil
+		return conn, nil
+	}
 }
 
 func NewListener(bindTo string, bufferSize int) (net.Listener, error) {

internal/utils/net_listener_test.go

@@ -0,0 +1,75 @@
+package utils_test
+
+import (
+	"net"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/9seconds/mtg/v2/internal/utils"
+)
+
+// firstConnBroken returns an already closed TCP connection first (setting
+// socket options on it fails) and then real connections.
+type firstConnBroken struct {
+	net.Listener
+
+	done atomic.Bool
+}
+
+func (l *firstConnBroken) Accept() (net.Conn, error) {
+	conn, err := l.Listener.Accept()
+	if err != nil {
+		return nil, err
+	}
+
+	if l.done.CompareAndSwap(false, true) {
+		conn.Close() //nolint: errcheck
+	}
+
+	return conn, nil
+}
+
+// One connection whose socket options cannot be set (for example, reset by
+// the client right away) must be skipped, not stop the whole accept loop.
+func TestListenerSkipsConnectionWithBrokenSocket(t *testing.T) {
+	t.Parallel()
+
+	base, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	listener := utils.Listener{Listener: &firstConnBroken{Listener: base}}
+	defer listener.Close() //nolint: errcheck
+
+	for range 2 {
+		go func() {
+			conn, err := net.Dial("tcp", base.Addr().String())
+			if err == nil {
+				time.Sleep(200 * time.Millisecond)
+				conn.Close() //nolint: errcheck
+			}
+		}()
+	}
+
+	result := make(chan error, 1)
+
+	go func() {
+		conn, err := listener.Accept()
+		if err == nil {
+			conn.Close() //nolint: errcheck
+		}
+
+		result <- err
+	}()
+
+	select {
+	case err := <-result:
+		if err != nil {
+			t.Fatalf("expected the second connection, got error %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("Accept did not return the next connection")
+	}
+}

mtglib/accept.go

@@ -0,0 +1,78 @@
+package mtglib
+
+import (
+	"context"
+	"errors"
+	"net"
+	"syscall"
+	"time"
+)
+
+const (
+	acceptRetryDelayMin = 5 * time.Millisecond
+	acceptRetryDelayMax = time.Second
+)
+
+// IsTemporaryAcceptError reports whether an Accept error is transient: the
+// listener is still usable and accepting should be retried after a pause.
+// Running out of file descriptors under a connection flood (EMFILE/ENFILE) is
+// the important case: returning from the accept loop on such an error leaves
+// the process alive but deaf until it is restarted.
+func IsTemporaryAcceptError(err error) bool {
+	if err == nil || errors.Is(err, net.ErrClosed) {
+		return false
+	}
+
+	var netErr net.Error
+	if errors.As(err, &netErr) && netErr.Timeout() {
+		return true
+	}
+
+	return errors.Is(err, syscall.EMFILE) ||
+		errors.Is(err, syscall.ENFILE) ||
+		errors.Is(err, syscall.ENOBUFS) ||
+		errors.Is(err, syscall.ENOMEM) ||
+		errors.Is(err, syscall.ECONNABORTED) ||
+		errors.Is(err, syscall.ECONNRESET)
+}
+
+// AcceptRetryDelay returns the pause before the next Accept attempt after a
+// temporary error: it starts at 5ms and doubles up to 1s, like net/http.
+func AcceptRetryDelay(previous time.Duration) time.Duration {
+	if previous == 0 {
+		return acceptRetryDelayMin
+	}
+
+	return min(previous*2, acceptRetryDelayMax)
+}
+
+// acceptWithRetry accepts the next connection, retrying temporary errors with
+// a growing pause. It returns an error only when the listener is unusable or
+// ctx is done.
+func acceptWithRetry(ctx context.Context, listener net.Listener, logger Logger) (net.Conn, error) {
+	var delay time.Duration
+
+	for {
+		conn, err := listener.Accept()
+		if err == nil {
+			return conn, nil
+		}
+
+		if ctx.Err() != nil || !IsTemporaryAcceptError(err) {
+			return nil, err
+		}
+
+		delay = AcceptRetryDelay(delay)
+		logger.BindStr("retry_in", delay.String()).WarningError("temporary accept error", err)
+
+		timer := time.NewTimer(delay)
+
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+
+			return nil, ctx.Err()
+		case <-timer.C:
+		}
+	}
+}