git » chasquid » commit 6b64c9c

diff --git a/docs/monitoring.md b/docs/monitoring.md
index d9597ac..41e15b3 100644
--- a/docs/monitoring.md
+++ b/docs/monitoring.md
@@ -50,9 +50,16 @@ List of exported variables:
   count of items the queue wrote to disk.
 - **chasquid/queue/putCount** (counter)  
   number of envelopes put in the queue.
+- **chasquid/smtpIn/acceptErrors** (counter)  
+  count of errors accepting incoming connections.
+- **chasquid/smtpIn/activeConnections** (gauge)  
+  number of active incoming connections.
 - **chasquid/smtpIn/commandCount** (map of command -> count)  
   count of SMTP commands received, by command. Note that for unknown commands
   we use `unknown<COMMAND>`.
+- **chasquid/smtpIn/connectionsOverLimit** (counter)  
+  count of incoming connections closed due to the connection limit, which is
+  half of the file descriptor limit.
 - **chasquid/smtpIn/dkimSignErrors** (counter)  
   count of DKIM sign errors
 - **chasquid/smtpIn/dkimSigned** (counter)  
diff --git a/internal/smtpsrv/server.go b/internal/smtpsrv/server.go
index 04a6445..00917ee 100644
--- a/internal/smtpsrv/server.go
+++ b/internal/smtpsrv/server.go
@@ -11,12 +11,15 @@ import (
 	"errors"
 	"flag"
 	"fmt"
+	"math"
 	"net"
 	"net/http"
 	"net/url"
 	"os"
 	"path"
 	"strings"
+	"sync/atomic"
+	"syscall"
 	"time"
 
 	"blitiri.com.ar/go/chasquid/internal/aliases"
@@ -52,6 +55,11 @@ const (
 var (
 	acceptErrors = expvarom.NewInt("chasquid/smtpIn/acceptErrors",
 		"count of errors accepting incoming connections")
+	activeConnections = expvarom.NewInt("chasquid/smtpIn/activeConnections",
+		"number of active incoming connections")
+	connectionsOverLimit = expvarom.NewInt(
+		"chasquid/smtpIn/connectionsOverLimit",
+		"count of incoming connections closed due to the connection limit")
 )
 
 // Server represents an SMTP server instance.
@@ -101,6 +109,14 @@ type Server struct {
 
 	// Path to the hooks.
 	HookPath string
+
+	// Maximum number of concurrent incoming connections (0 = no limit), and
+	// how many there are right now.
+	maxConns    int64
+	activeConns atomic.Int64
+
+	// When we last logged about reaching the connection limit (Unix time).
+	connLimitLogged atomic.Int64
 }
 
 // NewServer returns a new empty Server.
@@ -132,6 +148,7 @@ func NewServer() *Server {
 
 		connTimeout:    20 * time.Minute,
 		commandTimeout: 1 * time.Minute,
+		maxConns:       connLimit(),
 		localDomains:   &set.String{},
 		authr:          authr,
 		aliasesR:       aliasesR,
@@ -331,6 +348,12 @@ func (s *Server) ListenAndServe() {
 
 	go s.periodicallyReload()
 
+	if s.maxConns > 0 {
+		log.Infof("Incoming connection limit: %d", s.maxConns)
+	} else {
+		log.Infof("No incoming connection limit")
+	}
+
 	for m, addrs := range s.addrs {
 		for _, addr := range addrs {
 			l, err := net.Listen("tcp", addr)
@@ -394,6 +417,23 @@ func (s *Server) serve(l net.Listener, mode SocketMode) {
 		}
 		backoff = acceptMinBackoff
 
+		// Close the connection if we're over the limit. Log at most once a
+		// minute (across all listeners) to avoid flooding the logs.
+		if n := s.activeConns.Add(1); s.maxConns > 0 && n > s.maxConns {
+			s.activeConns.Add(-1)
+			connectionsOverLimit.Add(1)
+			conn.Close()
+
+			now, last := time.Now().Unix(), s.connLimitLogged.Load()
+			if now-last >= 60 && s.connLimitLogged.CompareAndSwap(last, now) {
+				log.Errorf("Incoming connection limit (%d) reached, "+
+					"closing new connections (%d closed so far)",
+					s.maxConns, connectionsOverLimit.Value())
+			}
+			continue
+		}
+		activeConnections.Add(1)
+
 		sc := &Conn{
 			hostname:       s.Hostname,
 			maxDataSize:    s.MaxDataSize,
@@ -412,6 +452,32 @@ func (s *Server) serve(l net.Listener, mode SocketMode) {
 			commandTimeout: s.commandTimeout,
 			queue:          s.queue,
 		}
-		go sc.Handle()
+		go func() {
+			sc.Handle()
+			s.activeConns.Add(-1)
+			activeConnections.Add(-1)
+		}()
+	}
+}
+
+// connLimit returns the maximum number of concurrent incoming connections we
+// should allow, or 0 for no limit.
+//
+// It's half of the file descriptor limit, to leave room for everything else
+// that needs them (queue files, outgoing connections, hooks, etc.), so a large
+// number of incoming connections doesn't prevent the rest of the server from
+// working.
+func connLimit() int64 {
+	var lim syscall.Rlimit
+	if err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &lim); err != nil {
+		return 0
+	}
+
+	// The type of lim.Cur varies across platforms (int64 or uint64). Values
+	// out of range (e.g. RLIM_INFINITY) mean there's no meaningful limit.
+	cur := int64(lim.Cur)
+	if cur <= 0 || cur > math.MaxInt32 {
+		return 0
 	}
+	return cur / 2
 }
diff --git a/internal/smtpsrv/server_test.go b/internal/smtpsrv/server_test.go
index b874ab3..7fd3d84 100644
--- a/internal/smtpsrv/server_test.go
+++ b/internal/smtpsrv/server_test.go
@@ -1,11 +1,11 @@
 package smtpsrv
 
 import (
-	"bufio"
 	"crypto/tls"
 	"errors"
 	"flag"
 	"fmt"
+	"io"
 	"io/fs"
 	"net"
 	"net/smtp"
@@ -575,75 +575,139 @@ MC4CAQAwBQYDK2VwBCIEID6bjSoiW6g6NJA67RNl0SZ7zpylVOq9w/VGAXF5whnS
 	}
 }
 
-// scriptedListener is a net.Listener that returns a predefined sequence of
-// results from Accept, and net.ErrClosed once they run out.
-type scriptedListener struct {
-	results []acceptResult
-}
+// chanListener is a net.Listener that returns the results sent to it from
+// Accept, and net.ErrClosed once the channel is closed.
+type chanListener chan acceptResult
 
 type acceptResult struct {
 	conn net.Conn
 	err  error
 }
 
-func (l *scriptedListener) Accept() (net.Conn, error) {
-	if len(l.results) == 0 {
+func (l chanListener) Accept() (net.Conn, error) {
+	r, ok := <-l
+	if !ok {
 		return nil, net.ErrClosed
 	}
-	r := l.results[0]
-	l.results = l.results[1:]
 	return r.conn, r.err
 }
 
-func (l *scriptedListener) Close() error   { return nil }
-func (l *scriptedListener) Addr() net.Addr { return &net.TCPAddr{} }
+func (l chanListener) Close() error   { return nil }
+func (l chanListener) Addr() net.Addr { return &net.TCPAddr{} }
 
-func TestAcceptErrors(t *testing.T) {
+// serveInBackground runs s.serve(l) in a goroutine, and returns a channel
+// that is closed once it returns.
+func serveInBackground(s *Server, l net.Listener) chan struct{} {
+	done := make(chan struct{})
+	go func() {
+		s.serve(l, ModeSMTP)
+		close(done)
+	}()
+	return done
+}
+
+func waitForServe(t *testing.T, done chan struct{}) {
+	t.Helper()
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatalf("serve did not return after net.ErrClosed")
+	}
+}
+
+// newPipeConn sends a new connection to the listener, and returns the client
+// side of it.
+func newPipeConn(l chanListener) *textproto.Conn {
 	srvConn, cliConn := net.Pipe()
-	defer cliConn.Close()
+	l <- acceptResult{conn: srvConn}
+	return textproto.NewConn(cliConn)
+}
 
-	emfile := &net.OpError{Op: "accept", Net: "tcp",
-		Err: os.NewSyscallError("accept4", syscall.EMFILE)}
-	l := &scriptedListener{results: []acceptResult{
-		{err: emfile},
-		{err: emfile},
-		{conn: srvConn},
-		{err: emfile},
-	}}
+func expectGreeting(t *testing.T, c *textproto.Conn, hostname string) {
+	t.Helper()
+	greeting, err := c.ReadLine()
+	if err != nil {
+		t.Fatalf("error reading greeting: %v", err)
+	}
+	if want := "220 " + hostname + " ESMTP chasquid"; greeting != want {
+		t.Errorf("got greeting %q, expected %q", greeting, want)
+	}
+}
 
+func TestAcceptErrors(t *testing.T) {
 	s := NewServer()
 	s.Hostname = "acceptsrv"
 	prevErrors := acceptErrors.Value()
 
+	emfile := &net.OpError{Op: "accept", Net: "tcp",
+		Err: os.NewSyscallError("accept4", syscall.EMFILE)}
+	srvConn, cliConn := net.Pipe()
+	defer cliConn.Close()
+
+	l := make(chanListener, 4)
+	l <- acceptResult{err: emfile}
+	l <- acceptResult{err: emfile}
+	l <- acceptResult{conn: srvConn}
+	l <- acceptResult{err: emfile}
+	close(l)
+
 	// serve should retry on errors, and return once the listener reports
 	// net.ErrClosed.
-	done := make(chan struct{})
-	go func() {
-		s.serve(l, ModeSMTP)
-		close(done)
-	}()
+	done := serveInBackground(s, l)
 
 	// The connection we got in between the errors must have been handled.
-	// net.Pipe is synchronous, so this also unblocks the server's write.
-	greeting, err := textproto.NewReader(bufio.NewReader(cliConn)).ReadLine()
-	if err != nil {
-		t.Fatalf("error reading greeting: %v", err)
-	}
-	if greeting != "220 acceptsrv ESMTP chasquid" {
-		t.Errorf("unexpected greeting: %q", greeting)
-	}
+	expectGreeting(t, textproto.NewConn(cliConn), "acceptsrv")
 
-	select {
-	case <-done:
-	case <-time.After(5 * time.Second):
-		t.Fatalf("serve did not return after net.ErrClosed")
-	}
+	waitForServe(t, done)
 
 	if n := acceptErrors.Value() - prevErrors; n != 3 {
 		t.Errorf("acceptErrors increased by %d, expected 3", n)
 	}
 }
 
+func TestConnectionLimit(t *testing.T) {
+	s := NewServer()
+	s.Hostname = "limitsrv"
+	s.maxConns = 2
+	prevOverLimit := connectionsOverLimit.Value()
+
+	l := make(chanListener)
+	done := serveInBackground(s, l)
+
+	// The first two connections are within the limit.
+	c1 := newPipeConn(l)
+	expectGreeting(t, c1, "limitsrv")
+	c2 := newPipeConn(l)
+	defer c2.Close()
+	expectGreeting(t, c2, "limitsrv")
+
+	// The third one is over the limit, and gets closed without a greeting.
+	c3 := newPipeConn(l)
+	if line, err := c3.ReadLine(); err != io.EOF {
+		t.Errorf("over the limit: got %q, %v; expected EOF", line, err)
+	}
+	if n := connectionsOverLimit.Value() - prevOverLimit; n != 1 {
+		t.Errorf("connectionsOverLimit increased by %d, expected 1", n)
+	}
+
+	// Once a connection finishes, we can accept new ones again.
+	c1.Close()
+	deadline := time.Now().Add(5 * time.Second)
+	for s.activeConns.Load() != 1 {
+		if time.Now().After(deadline) {
+			t.Fatalf("activeConns = %d, expected 1", s.activeConns.Load())
+		}
+		time.Sleep(time.Millisecond)
+	}
+
+	c4 := newPipeConn(l)
+	defer c4.Close()
+	expectGreeting(t, c4, "limitsrv")
+
+	close(l)
+	waitForServe(t, done)
+}
+
 //
 // === Benchmarks ===
 //
diff --git a/test/stress-02-connections/run.sh b/test/stress-02-connections/run.sh
index a237988..0623720 100755
--- a/test/stress-02-connections/run.sh
+++ b/test/stress-02-connections/run.sh
@@ -16,11 +16,12 @@ wait_until_ready 1025
 
 echo "Peak RAM: $(chasquid_ram_peak)"
 
-# Set connection count to (max open files) - (leeway).
-# We set the leeway to account for file descriptors opened by the runtime and
-# listeners; 20 should be enough for now.
+# Set connection count to (connection limit) - (leeway).
+# chasquid limits incoming connections to half of its file descriptor limit.
+# Go raises the soft limit to (almost) the hard limit at startup, so we use
+# the hard limit, with some leeway; 20 should be enough for now.
 # Cap it to 2000, as otherwise it can be problematic due to port availability.
-COUNT=$(( $(ulimit -n) - 20 ))
+COUNT=$(( $(ulimit -Hn) / 2 - 20 ))
 if [ $COUNT -gt 2000 ]; then
 	COUNT=2000
 fi
diff --git a/test/t-24-connection_limit/config/chasquid.conf b/test/t-24-connection_limit/config/chasquid.conf
new file mode 100644
index 0000000..cf76e8a
--- /dev/null
+++ b/test/t-24-connection_limit/config/chasquid.conf
@@ -0,0 +1,10 @@
+smtp_address: ":1025"
+submission_address: ":1587"
+submission_over_tls_address: ":1465"
+monitoring_address: ":1099"
+
+mail_delivery_agent_bin: "test-mda"
+mail_delivery_agent_args: "%to%"
+
+data_dir: "../.data"
+mail_log_path: "../.logs/mail_log"
diff --git a/test/t-24-connection_limit/content b/test/t-24-connection_limit/content
new file mode 100644
index 0000000..76a8b16
--- /dev/null
+++ b/test/t-24-connection_limit/content
@@ -0,0 +1,4 @@
+Subject: Prueba desde el test
+
+Crece desde el test el futuro
+Crece desde el test
diff --git a/test/t-24-connection_limit/run.sh b/test/t-24-connection_limit/run.sh
new file mode 100755
index 0000000..5f2abd5
--- /dev/null
+++ b/test/t-24-connection_limit/run.sh
@@ -0,0 +1,64 @@
+#!/bin/bash
+
+set -e
+. "$(dirname "$0")/../util/lib.sh"
+
+init
+
+generate_certs_for testserver
+add_user user@testserver secretpassword
+
+# Build in advance, so the build doesn't run under the low fd limit below.
+go-build-cached "${TBASE}/../../"
+
+# Run chasquid with a low file descriptor limit, which is easy to exhaust.
+# Reload very often: reloading needs to open files, so if we ran out of file
+# descriptors, chasquid would fail to reload and exit.
+mkdir -p .logs
+(
+	ulimit -n 64
+	chasquid -v=2 --logfile=.logs/chasquid.log --config_dir=config \
+		--testing__reload_every=50ms
+) &
+CHASQUID_PID=$!
+wait_until_ready 1025
+
+# The connection limit should be half the file descriptor limit.
+if ! grep -q "Incoming connection limit: 32$" .logs/chasquid.log; then
+	fail "unexpected connection limit"
+fi
+
+# Open and hold as many connections as the limit allows.
+conngen -addr=localhost:1025 -count=32 -wait > .logs/conngen.log 2>&1 &
+CONNGEN_PID=$!
+wait_until "grep -q 'done, created' .logs/conngen.log"
+
+# Open and hold more connections than there are file descriptors left. The
+# server should close them right away (so we read EOF, not time out).
+for _ in $(seq 1 40); do
+	exec {fd}<>/dev/tcp/localhost/1025
+	if read -r -t 5 line <&"$fd"; then
+		fail "connection over the limit got: $line"
+	elif [ $? -gt 128 ]; then
+		fail "connection over the limit was not closed"
+	fi
+done
+
+if ! grep -q "Incoming connection limit (32) reached" .logs/chasquid.log; then
+	fail "connection limit not logged"
+fi
+
+# Give it time to reload a few times, and check it's still running.
+sleep 0.5
+if ! kill -0 $CHASQUID_PID; then
+	fail "chasquid is not running"
+fi
+
+# Once the connections are closed, it should accept new ones, and work
+# normally. The server notices the connections were closed asynchronously, so
+# we retry until it works.
+kill $CONNGEN_PID
+wait_until "smtpc user@testserver < content 2>/dev/null"
+wait_for_file .mail/user@testserver
+
+success
diff --git a/test/t-24-connection_limit/smtpc.conf b/test/t-24-connection_limit/smtpc.conf
new file mode 100644
index 0000000..c912b55
--- /dev/null
+++ b/test/t-24-connection_limit/smtpc.conf
@@ -0,0 +1,4 @@
+addr localhost:1465
+server_cert config/certs/testserver/fullchain.pem
+user user@testserver
+password secretpassword