port/builtin: retry accept after fd exhaustion

pkg/port/builtin/parent/tcp/tcp.go

@@ -1,12 +1,15 @@
 package tcp
 
 import (
+	"errors"
 	"fmt"
 	"io"
 	"net"
 	"os"
 	"strconv"
 	"sync"
+	"syscall"
+	"time"
 
 	"github.com/rootless-containers/rootlesskit/v3/pkg/port"
 	"github.com/rootless-containers/rootlesskit/v3/pkg/port/builtin/msg"
@@ -19,17 +22,7 @@ func Run(socketPath string, spec port.Spec, stopCh <-chan struct{}, stoppedCh ch
 		return err
 	}
 	newConns := make(chan net.Conn)
-	go func() {
-		for {
-			c, err := ln.Accept()
-			if err != nil {
-				fmt.Fprintf(logWriter, "accept: %v\n", err)
-				close(newConns)
-				return
-			}
-			newConns <- c
-		}
-	}()
+	go acceptConnections(ln, newConns, stopCh, logWriter)
 	go func() {
 		defer func() {
 			stoppedCh <- ln.Close()
@@ -56,6 +49,63 @@ func Run(socketPath string, spec port.Spec, stopCh <-chan struct{}, stoppedCh ch
 	return nil
 }
 
+const (
+	minAcceptRetryDelay = 5 * time.Millisecond
+	maxAcceptRetryDelay = time.Second
+)
+
+func acceptConnections(ln net.Listener, newConns chan<- net.Conn, stopCh <-chan struct{}, logWriter io.Writer) {
+	defer close(newConns)
+
+	var retryDelay time.Duration
+	for {
+		c, err := ln.Accept()
+		if err != nil {
+			select {
+			case <-stopCh:
+				return
+			default:
+			}
+
+			if errors.Is(err, syscall.EMFILE) || errors.Is(err, syscall.ENFILE) {
+				if retryDelay == 0 {
+					retryDelay = minAcceptRetryDelay
+				} else {
+					retryDelay *= 2
+					if retryDelay > maxAcceptRetryDelay {
+						retryDelay = maxAcceptRetryDelay
+					}
+				}
+				fmt.Fprintf(logWriter, "accept: %v; retrying in %s\n", err, retryDelay)
+				timer := time.NewTimer(retryDelay)
+				select {
+				case <-timer.C:
+				case <-stopCh:
+					if !timer.Stop() {
+						select {
+						case <-timer.C:
+						default:
+						}
+					}
+					return
+				}
+				continue
+			}
+
+			fmt.Fprintf(logWriter, "accept: %v\n", err)
+			return
+		}
+		retryDelay = 0
+
+		select {
+		case newConns <- c:
+		case <-stopCh:
+			c.Close()
+			return
+		}
+	}
+}
+
 func copyConnToChild(c net.Conn, socketPath string, spec port.Spec, stopCh <-chan struct{}) error {
 	defer c.Close()
 	// get fd from the child as an SCM_RIGHTS cmsg

pkg/port/builtin/parent/tcp/tcp_test.go

@@ -0,0 +1,157 @@
+package tcp
+
+import (
+	"bytes"
+	"errors"
+	"net"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+)
+
+type acceptResult struct {
+	conn net.Conn
+	err  error
+}
+
+type testListener struct {
+	results chan acceptResult
+	closed  chan struct{}
+	once    sync.Once
+}
+
+func newTestListener() *testListener {
+	return &testListener{
+		results: make(chan acceptResult),
+		closed:  make(chan struct{}),
+	}
+}
+
+func (l *testListener) Accept() (net.Conn, error) {
+	select {
+	case result := <-l.results:
+		return result.conn, result.err
+	case <-l.closed:
+		return nil, net.ErrClosed
+	}
+}
+
+func (l *testListener) Close() error {
+	l.once.Do(func() { close(l.closed) })
+	return nil
+}
+
+func (l *testListener) Addr() net.Addr {
+	return &net.TCPAddr{}
+}
+
+type signalWriter struct {
+	wrote chan struct{}
+}
+
+func (w signalWriter) Write(p []byte) (int, error) {
+	select {
+	case w.wrote <- struct{}{}:
+	default:
+	}
+	return len(p), nil
+}
+
+func TestAcceptConnectionsRetriesFileDescriptorExhaustion(t *testing.T) {
+	for _, acceptErr := range []error{syscall.EMFILE, syscall.ENFILE} {
+		t.Run(acceptErr.Error(), func(t *testing.T) {
+			listener := newTestListener()
+			stopCh := make(chan struct{})
+			newConns := make(chan net.Conn)
+			done := make(chan struct{})
+			var log bytes.Buffer
+
+			go func() {
+				acceptConnections(listener, newConns, stopCh, &log)
+				close(done)
+			}()
+
+			listener.results <- acceptResult{err: acceptErr}
+			serverConn, clientConn := net.Pipe()
+			defer serverConn.Close()
+			defer clientConn.Close()
+			listener.results <- acceptResult{conn: serverConn}
+
+			select {
+			case conn := <-newConns:
+				if conn != serverConn {
+					t.Fatalf("received unexpected connection %v", conn)
+				}
+			case <-time.After(time.Second):
+				t.Fatal("accept loop did not retry after descriptor exhaustion")
+			}
+
+			close(stopCh)
+			listener.Close()
+			select {
+			case <-done:
+			case <-time.After(time.Second):
+				t.Fatal("accept loop did not stop")
+			}
+		})
+	}
+}
+
+func TestAcceptConnectionsStopsDuringRetryDelay(t *testing.T) {
+	listener := newTestListener()
+	stopCh := make(chan struct{})
+	newConns := make(chan net.Conn)
+	done := make(chan struct{})
+	wrote := make(chan struct{}, 1)
+
+	go func() {
+		acceptConnections(listener, newConns, stopCh, signalWriter{wrote: wrote})
+		close(done)
+	}()
+
+	listener.results <- acceptResult{err: syscall.EMFILE}
+	select {
+	case <-wrote:
+	case <-time.After(time.Second):
+		t.Fatal("accept loop did not report descriptor exhaustion")
+	}
+
+	close(stopCh)
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("accept loop did not stop during retry delay")
+	}
+	if conn, ok := <-newConns; ok {
+		t.Fatalf("expected closed connection channel, got %v", conn)
+	}
+}
+
+func TestAcceptConnectionsStopsOnPermanentError(t *testing.T) {
+	listener := newTestListener()
+	newConns := make(chan net.Conn)
+	var log bytes.Buffer
+	done := make(chan struct{})
+
+	go func() {
+		acceptConnections(listener, newConns, make(chan struct{}), &log)
+		close(done)
+	}()
+
+	wantErr := errors.New("permanent accept error")
+	listener.results <- acceptResult{err: wantErr}
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("accept loop did not stop after a permanent error")
+	}
+
+	if got := <-newConns; got != nil {
+		t.Fatalf("expected closed connection channel, got %v", got)
+	}
+	if !bytes.Contains(log.Bytes(), []byte(wantErr.Error())) {
+		t.Fatalf("expected error in log, got %q", log.String())
+	}
+}