fix: do not stop accepting connections on transient errors

internal/cli/run_proxy.go

@@ -333,7 +333,7 @@ func runProxy(conf *config.Config, version string) error { //nolint: funlen, cyc
 		return fmt.Errorf("cannot create a proxy: %w", err)
 	}
 
-	listener, err := utils.NewListener(conf.BindTo.Get(""), 0)
+	listener, err := utils.NewListener(conf.BindTo.Get(""), 0, logger.Named("listener"))
 	if err != nil {
 		return fmt.Errorf("cannot start proxy: %w", err)
 	}
@@ -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/logthrottle/logthrottle.go

@@ -0,0 +1,53 @@
+// Package logthrottle limits how often a repetitive log line is written.
+//
+// Some log lines are written per connection on paths an attacker controls (a
+// flood of rejected or broken connections). Writing each of them turns a
+// flood into a log flood, while dropping them makes the problem invisible.
+// A Throttle lets one line through per interval and counts the rest, so the
+// next line can report how many were suppressed.
+package logthrottle
+
+import (
+	"sync"
+	"time"
+)
+
+// Throttle allows at most one message per interval. It is safe for
+// concurrent use.
+type Throttle struct {
+	mu         sync.Mutex
+	interval   time.Duration
+	last       time.Time
+	suppressed int
+	now        func() time.Time
+}
+
+// New returns a Throttle that allows one message per interval.
+func New(interval time.Duration) *Throttle {
+	return &Throttle{
+		interval: interval,
+		now:      time.Now,
+	}
+}
+
+// Allow reports whether a message may be written now. If it may, it also
+// returns the number of messages suppressed since the previous allowed one,
+// and resets that counter.
+func (t *Throttle) Allow() (bool, int) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	now := t.now()
+
+	if !t.last.IsZero() && now.Sub(t.last) < t.interval {
+		t.suppressed++
+
+		return false, 0
+	}
+
+	suppressed := t.suppressed
+	t.last = now
+	t.suppressed = 0
+
+	return true, suppressed
+}

internal/logthrottle/logthrottle_internal_test.go

@@ -0,0 +1,66 @@
+package logthrottle
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestThrottleAllowsOnePerIntervalAndCountsTheRest(t *testing.T) {
+	t.Parallel()
+
+	now := time.Unix(1_000_000, 0)
+
+	throttle := New(10 * time.Second)
+	throttle.now = func() time.Time { return now }
+
+	if ok, suppressed := throttle.Allow(); !ok || suppressed != 0 {
+		t.Fatalf("the first message must pass: ok=%v suppressed=%d", ok, suppressed)
+	}
+
+	for range 5 {
+		now = now.Add(time.Second)
+
+		if ok, _ := throttle.Allow(); ok {
+			t.Fatal("a message within the interval must be suppressed")
+		}
+	}
+
+	now = now.Add(5 * time.Second)
+
+	if ok, suppressed := throttle.Allow(); !ok || suppressed != 5 {
+		t.Fatalf("after the interval: want ok with 5 suppressed, got ok=%v suppressed=%d", ok, suppressed)
+	}
+
+	now = now.Add(10 * time.Second)
+
+	if ok, suppressed := throttle.Allow(); !ok || suppressed != 0 {
+		t.Fatalf("the counter must reset: ok=%v suppressed=%d", ok, suppressed)
+	}
+}
+
+func TestThrottleConcurrentLetsOneThrough(t *testing.T) {
+	t.Parallel()
+
+	throttle := New(time.Hour)
+
+	var (
+		allowed atomic.Int32
+		wg      sync.WaitGroup
+	)
+
+	for range 100 {
+		wg.Go(func() {
+			if ok, _ := throttle.Allow(); ok {
+				allowed.Add(1)
+			}
+		})
+	}
+
+	wg.Wait()
+
+	if allowed.Load() != 1 {
+		t.Fatalf("exactly one message must pass, got %d", allowed.Load())
+	}
+}

internal/utils/net_listener.go

@@ -3,36 +3,81 @@ package utils
 import (
 	"fmt"
 	"net"
+	"time"
 
+	"github.com/9seconds/mtg/v2/internal/logthrottle"
+	"github.com/9seconds/mtg/v2/mtglib"
 	"github.com/9seconds/mtg/v2/network"
 )
 
+// socketOptionsLogInterval limits the warning about connections dropped
+// because their socket options cannot be set: one line per interval, with the
+// number of suppressed ones.
+const socketOptionsLogInterval = 10 * time.Second
+
+// Listener applies client socket options to accepted connections.
 type Listener struct {
 	net.Listener
+
+	logger   mtglib.Logger
+	throttle *logthrottle.Throttle
 }
 
-func (l Listener) Accept() (net.Conn, error) {
-	conn, err := l.Listener.Accept()
-	if err != nil {
-		return nil, err //nolint: wrapcheck
+// WrapListener returns a Listener that accepts connections from base.
+func WrapListener(base net.Listener, logger mtglib.Logger) *Listener {
+	return &Listener{
+		Listener: base,
+		logger:   logger,
+		throttle: logthrottle.New(socketOptionsLogInterval),
+	}
+}
+
+// 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.
+//
+// The skip is logged as a warning, rate limited: if setting the options starts
+// failing for every connection (a bad option, an unsupported platform), the
+// proxy must not drop everything silently, nor flood the log under load.
+func (l *Listener) Accept() (net.Conn, error) {
+	for {
+		conn, err := l.Listener.Accept()
+		if err != nil {
+			return nil, err //nolint: wrapcheck
+		}
+
+		if err := network.SetClientSocketOptions(conn, 0); err != nil {
+			l.logSkipped(conn, err)
+			conn.Close() //nolint: errcheck
+
+			continue
+		}
+
+		return conn, nil
 	}
+}
 
-	if err := network.SetClientSocketOptions(conn, 0); err != nil {
-		conn.Close() //nolint: errcheck
+func (l *Listener) logSkipped(conn net.Conn, err error) {
+	ok, suppressed := l.throttle.Allow()
+	if !ok {
+		return
+	}
+
+	logger := l.logger.BindInt("suppressed", suppressed)
 
-		return nil, fmt.Errorf("cannot set TCP options: %w", err)
+	if addr := conn.RemoteAddr(); addr != nil {
+		logger = logger.BindStr("ip", addr.String())
 	}
 
-	return conn, nil
+	logger.WarningError("cannot set client socket options, the connection is dropped", err)
 }
 
-func NewListener(bindTo string, bufferSize int) (net.Listener, error) {
+func NewListener(bindTo string, bufferSize int, logger mtglib.Logger) (net.Listener, error) {
 	base, err := net.Listen("tcp", bindTo)
 	if err != nil {
 		return nil, fmt.Errorf("cannot build a base listener: %w", err)
 	}
 
-	return Listener{
-		Listener: base,
-	}, nil
+	return WrapListener(base, logger), nil
 }