diff --git a/internal/cli/run_proxy.go b/internal/cli/run_proxy.go index f1ce8af8b..bfabc3ff0 100644 --- a/internal/cli/run_proxy.go +++ b/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 } diff --git a/internal/cli/run_proxy_shutdown_test.go b/internal/cli/run_proxy_shutdown_test.go new file mode 100644 index 000000000..7ab195e45 --- /dev/null +++ b/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) +} diff --git a/internal/logthrottle/logthrottle.go b/internal/logthrottle/logthrottle.go new file mode 100644 index 000000000..f8983fefc --- /dev/null +++ b/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 +} diff --git a/internal/logthrottle/logthrottle_internal_test.go b/internal/logthrottle/logthrottle_internal_test.go new file mode 100644 index 000000000..aecb7cdce --- /dev/null +++ b/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()) + } +} diff --git a/internal/utils/net_listener.go b/internal/utils/net_listener.go index 59b1dc11b..77aa64e83 100644 --- a/internal/utils/net_listener.go +++ b/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 } diff --git a/internal/utils/net_listener_test.go b/internal/utils/net_listener_test.go new file mode 100644 index 000000000..0f97523d5 --- /dev/null +++ b/internal/utils/net_listener_test.go @@ -0,0 +1,197 @@ +package utils_test + +import ( + "net" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/9seconds/mtg/v2/internal/utils" + "github.com/9seconds/mtg/v2/mtglib" +) + +// 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) + } + + log := &recordingLogger{} + + listener := utils.WrapListener(&firstConnBroken{Listener: base}, log) + 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") + } + + // The skipped connection must not vanish silently. + if warnings := log.warnings(); len(warnings) != 1 { + t.Fatalf("expected one warning about the skipped connection, got %v", warnings) + } +} + +// brokenConns returns only already closed TCP connections: setting socket +// options fails on every one of them. +type brokenConns struct { + net.Listener + + calls atomic.Int32 +} + +func (l *brokenConns) Accept() (net.Conn, error) { + l.calls.Add(1) + + conn, err := l.Listener.Accept() + if err != nil { + return nil, err //nolint: wrapcheck + } + + conn.Close() //nolint: errcheck + + return conn, nil +} + +// If setting socket options fails for every connection, the warning must be +// rate limited instead of written per connection. +func TestListenerRateLimitsSkipWarning(t *testing.T) { + t.Parallel() + + base, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + + log := &recordingLogger{} + broken := &brokenConns{Listener: base} + + listener := utils.WrapListener(broken, log) + + const conns = 10 + + var dialed sync.WaitGroup + + for range conns { + dialed.Go(func() { + conn, err := net.Dial("tcp", base.Addr().String()) + if err == nil { + conn.Close() //nolint: errcheck + } + }) + } + + done := make(chan struct{}) + + go func() { + defer close(done) + + listener.Accept() //nolint: errcheck + }() + + dialed.Wait() + + // The listener has skipped every connection once it asks for one more. + deadline := time.Now().Add(2 * time.Second) + + for broken.calls.Load() <= conns && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + + listener.Close() //nolint: errcheck + <-done + + if got := broken.calls.Load(); got <= conns { + t.Fatalf("expected %d skipped connections, got %d", conns, got-1) + } + + if warnings := log.warnings(); len(warnings) != 1 { + t.Fatalf("expected one rate-limited warning for %d skipped connections, got %v", conns, warnings) + } +} + +// recordingLogger records warnings. +type recordingLogger struct { + mu sync.Mutex + msgs []string +} + +func (l *recordingLogger) Named(_ string) mtglib.Logger { return l } +func (l *recordingLogger) BindInt(_ string, _ int) mtglib.Logger { return l } +func (l *recordingLogger) BindStr(_, _ string) mtglib.Logger { return l } +func (l *recordingLogger) BindJSON(_, _ string) mtglib.Logger { return l } +func (l *recordingLogger) Printf(_ string, _ ...any) {} +func (l *recordingLogger) Info(_ string) {} +func (l *recordingLogger) InfoError(_ string, _ error) {} +func (l *recordingLogger) Warning(msg string) { l.record(msg) } +func (l *recordingLogger) Debug(_ string) {} +func (l *recordingLogger) DebugError(_ string, _ error) {} + +func (l *recordingLogger) WarningError(msg string, err error) { + l.record(msg + ": " + err.Error()) +} + +func (l *recordingLogger) record(msg string) { + l.mu.Lock() + defer l.mu.Unlock() + + l.msgs = append(l.msgs, msg) +} + +func (l *recordingLogger) warnings() []string { + l.mu.Lock() + defer l.mu.Unlock() + + return append([]string(nil), l.msgs...) +} diff --git a/mtglib/accept.go b/mtglib/accept.go new file mode 100644 index 000000000..64c4d8e03 --- /dev/null +++ b/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: + } + } +} diff --git a/mtglib/accept_internal_test.go b/mtglib/accept_internal_test.go new file mode 100644 index 000000000..d3a669ec6 --- /dev/null +++ b/mtglib/accept_internal_test.go @@ -0,0 +1,137 @@ +package mtglib + +import ( + "context" + "errors" + "fmt" + "net" + "sync" + "syscall" + "testing" + "time" +) + +// scriptedListener returns the scripted results of Accept in order and then +// net.ErrClosed. +type scriptedListener struct { + mu sync.Mutex + results []error + calls int +} + +func (l *scriptedListener) Accept() (net.Conn, error) { + l.mu.Lock() + defer l.mu.Unlock() + + l.calls++ + + if len(l.results) == 0 { + return nil, net.ErrClosed + } + + err := l.results[0] + l.results = l.results[1:] + + if err != nil { + return nil, err + } + + client, server := net.Pipe() + client.Close() //nolint: errcheck + + return server, nil +} + +func (l *scriptedListener) Close() error { return nil } +func (l *scriptedListener) Addr() net.Addr { return &net.TCPAddr{} } + +func TestIsTemporaryAcceptError(t *testing.T) { + t.Parallel() + + temporary := []error{ + syscall.EMFILE, + fmt.Errorf("accept tcp: %w", syscall.ENFILE), + &net.OpError{Op: "accept", Err: syscall.ECONNABORTED}, + syscall.ENOBUFS, + } + for _, err := range temporary { + if !isTemporaryAcceptError(err) { + t.Errorf("%v must be temporary", err) + } + } + + permanent := []error{nil, net.ErrClosed, fmt.Errorf("wrapped: %w", net.ErrClosed), errors.New("boom")} + for _, err := range permanent { + if isTemporaryAcceptError(err) { + t.Errorf("%v must not be temporary", err) + } + } +} + +func TestAcceptRetryDelayGrowsUpToOneSecond(t *testing.T) { + t.Parallel() + + delay := acceptRetryDelay(0) + if delay != acceptRetryDelayMin { + t.Fatalf("first delay: got %v", delay) + } + + for range 20 { + delay = acceptRetryDelay(delay) + } + + if delay != acceptRetryDelayMax { + t.Fatalf("delay must be capped at %v, got %v", acceptRetryDelayMax, delay) + } +} + +// Running out of file descriptors under a connection flood must not stop the +// accept loop: the next connection is accepted once descriptors are available. +func TestAcceptWithRetrySurvivesTemporaryErrors(t *testing.T) { + t.Parallel() + + listener := &scriptedListener{results: []error{syscall.EMFILE, syscall.EMFILE, nil}} + + conn, err := acceptWithRetry(context.Background(), listener, NoopLogger{}) + if err != nil { + t.Fatalf("expected a connection after temporary errors, got %v", err) + } + + conn.Close() //nolint: errcheck + + if listener.calls != 3 { + t.Fatalf("expected 3 Accept calls, got %d", listener.calls) + } +} + +func TestAcceptWithRetryStopsOnClosedListener(t *testing.T) { + t.Parallel() + + _, err := acceptWithRetry(context.Background(), &scriptedListener{}, NoopLogger{}) + if !errors.Is(err, net.ErrClosed) { + t.Fatalf("expected net.ErrClosed, got %v", err) + } +} + +func TestAcceptWithRetryStopsOnShutdownDuringPause(t *testing.T) { + t.Parallel() + + errs := make([]error, 100) + for i := range errs { + errs[i] = syscall.EMFILE + } + + ctx, cancel := context.WithCancel(context.Background()) + time.AfterFunc(50*time.Millisecond, cancel) + + start := time.Now() + + _, err := acceptWithRetry(ctx, &scriptedListener{results: errs}, NoopLogger{}) + if err == nil { + t.Fatal("expected an error after shutdown") + } + + if time.Since(start) > time.Second { + t.Fatalf("shutdown must interrupt the retry pause, took %v", time.Since(start)) + } +} diff --git a/mtglib/proxy.go b/mtglib/proxy.go index 1ea6f2e63..d9fa29ec8 100644 --- a/mtglib/proxy.go +++ b/mtglib/proxy.go @@ -130,7 +130,7 @@ func (p *Proxy) Serve(listener net.Listener) error { defer p.streamWaitGroup.Done() for { - conn, err := listener.Accept() + conn, err := acceptWithRetry(p.ctx, listener, p.logger) if err != nil { select { case <-p.ctx.Done():