From e660dc1c22ed4d74e643fad93dd2853fed9a80b7 Mon Sep 17 00:00:00 2001 From: Andrey Osipuk <45951478+AndreyOsipuk@users.noreply.github.com> Date: Tue, 6 Oct 2026 15:11:33 +0300 Subject: [PATCH 1/2] fix: do not stop accepting connections on transient errors Proxy.Serve returned on any Accept error and run_proxy started it with 'go proxy.Serve(listener)' ignoring the result, so running out of file descriptors under a connection flood (EMFILE) left the process alive but no longer accepting connections. utils.Listener also returned an error when socket options could not be set on one accepted connection (e.g. reset by the client right away), which stopped the accept loop because of one bad client. Temporary accept errors are now retried with a pause growing from 5ms to 1s, like net/http; a connection with broken socket options is closed and skipped; and if Serve still returns, runProxy shuts down and returns the error so a supervisor restarts the process. --- internal/cli/run_proxy.go | 22 ++++- internal/utils/net_listener.go | 24 +++-- internal/utils/net_listener_test.go | 75 +++++++++++++++ mtglib/accept.go | 78 ++++++++++++++++ mtglib/accept_internal_test.go | 137 ++++++++++++++++++++++++++++ mtglib/proxy.go | 2 +- 6 files changed, 326 insertions(+), 12 deletions(-) create mode 100644 internal/utils/net_listener_test.go create mode 100644 mtglib/accept.go create mode 100644 mtglib/accept_internal_test.go diff --git a/internal/cli/run_proxy.go b/internal/cli/run_proxy.go index f1ce8af8b..ef482f8e6 100644 --- a/internal/cli/run_proxy.go +++ b/internal/cli/run_proxy.go @@ -346,13 +346,31 @@ 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() + }() <-ctx.Done() listener.Close() //nolint: errcheck proxy.Shutdown() + select { + case err := <-serveErr: + if err != nil { + return fmt.Errorf("proxy stopped accepting connections: %w", err) + } + default: + } + return nil } diff --git a/internal/utils/net_listener.go b/internal/utils/net_listener.go index 59b1dc11b..040dc886f 100644 --- a/internal/utils/net_listener.go +++ b/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) { diff --git a/internal/utils/net_listener_test.go b/internal/utils/net_listener_test.go new file mode 100644 index 000000000..411fa53c6 --- /dev/null +++ b/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") + } +} diff --git a/mtglib/accept.go b/mtglib/accept.go new file mode 100644 index 000000000..fd15fe608 --- /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..c835fd30a --- /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(): From 0df3d4d4227713e78b1538581a7b68f479ec9f4b Mon Sep 17 00:00:00 2001 From: Andrey Osipuk <45951478+AndreyOsipuk@users.noreply.github.com> Date: Tue, 6 Oct 2026 16:05:55 +0300 Subject: [PATCH 2/2] fix: an ordinary stop is not an accept-loop failure The listener is closed during shutdown, so Serve returns an accept error and every stop or restart exited with code 1. Only a Serve that failed before shutdown started is reported now. --- internal/cli/run_proxy.go | 28 ++++++++++++---- internal/cli/run_proxy_shutdown_test.go | 43 +++++++++++++++++++++++++ 2 files changed, 65 insertions(+), 6 deletions(-) create mode 100644 internal/cli/run_proxy_shutdown_test.go diff --git a/internal/cli/run_proxy.go b/internal/cli/run_proxy.go index ef482f8e6..058d3abcc 100644 --- a/internal/cli/run_proxy.go +++ b/internal/cli/run_proxy.go @@ -360,17 +360,33 @@ func runProxy(conf *config.Config, version string) error { //nolint: funlen, cyc 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 err := <-serveErr: - if err != nil { - return fmt.Errorf("proxy stopped accepting connections: %w", err) - } + 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) +}