fix(dnsfwd): bound TCP connections and in-flight queries per listener, absolute TCP deadline, drop saturated UDP

commit: af04fdc8c3abe0592df27988c36ea772352a6000
author: brice · 10d ago · 2026-10-12 15:54 UTC
parent: a706491
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +218 -35

M internal/dnsfwd/dnsfwd.go +73 -35


@@ -6,9 +6,12 @@
6	6		// policy lets guests reach nothing else on the host. It does not parse queries
7	7		// beyond what framing needs, so it supports every record type and EDNS.
8	8		//
9		-	// Abuse limits: at most MaxInflight concurrent upstream exchanges across all
10		-	// listeners, a deadline on every upstream exchange, idle and total timeouts on
11		-	// TCP connections, and a cap on TCP queries per connection.
	9	+	// Abuse limits, per listener (one listener per guest, so one guest cannot
	10	+	// starve another): at most MaxInflight concurrent upstream exchanges (UDP
	11	+	// queries beyond that are dropped, never queued), at most MaxTCPConns open TCP
	12	+	// connections, an absolute lifetime of TCPLifetime per TCP connection (a client
	13	+	// cannot hold a slot by trickling bytes), a deadline on every upstream
	14	+	// exchange, and a cap on queries per TCP connection.
12	15		package dnsfwd
13	16
14	17		import (

@@ -21,18 +24,24 @@ import (
21	24		"net/netip"
22	25		"strconv"
23	26		"sync"
	27	+	"sync/atomic"
24	28		"time"
25	29
26	30		"github.com/bnema/zerowrap"
27	31		)
28	32
29	33		const (
30		-	// MaxInflight bounds concurrent upstream exchanges.
31		-	MaxInflight = 256
	34	+	// TCPLifetime is the absolute lifetime of a TCP connection, set at accept.
	35	+	TCPLifetime = 30 * time.Second
32	36		// UpstreamTimeout is the deadline of one upstream attempt.
33	37		UpstreamTimeout = 3 * time.Second
34	38		// TCPIdleTimeout closes TCP client connections that stay silent.
35	39		TCPIdleTimeout = 10 * time.Second
	40	+
	41	+	// MaxInflight bounds concurrent upstream exchanges per listener.
	42	+	MaxInflight = 32
	43	+	// MaxTCPConns bounds open TCP connections per listener.
	44	+	MaxTCPConns = 16
36	45		// TCPMaxQueries caps queries per TCP connection.
37	46		TCPMaxQueries = 64
38	47

@@ -44,7 +53,8 @@ const (
44	53		type Forwarder struct {
45	54		upstreams func() []string
46	55		port int
47		-	sem chan struct{}
	56	+	// Lifetime overrides TCPLifetime; set before Listen (tests).
	57	+	Lifetime time.Duration
48	58
49	59		mu sync.Mutex
50	60		listeners map[netip.Addr]*listener

@@ -55,6 +65,11 @@ type listener struct {
55	65		tcp net.Listener
56	66		cancel context.CancelFunc
57	67		done sync.WaitGroup
	68	+	// inflight and conns are the per-listener limits (counting semaphores).
	69	+	inflight chan struct{}
	70	+	conns chan struct{}
	71	+	// Dropped counts UDP queries dropped because the listener was saturated.
	72	+	dropped atomic.Int64
58	73		}
59	74
60	75		// New returns a Forwarder. upstreams returns the "host:port" resolvers to try

@@ -64,7 +79,7 @@ func New(upstreams func() []string, port int) *Forwarder {
64	79		return &Forwarder{
65	80		upstreams: upstreams,
66	81		port: port,
67		-	sem: make(chan struct{}, MaxInflight),
	82	+	Lifetime: TCPLifetime,
68	83		listeners: map[netip.Addr]*listener{},
69	84		}
70	85		}

@@ -93,12 +108,12 @@ func (f *Forwarder) Listen(ctx context.Context, addr netip.Addr) error {
93	108		return fmt.Errorf("dnsfwd: listen tcp %s: %w", hp, err)
94	109		}
95	110		lctx, cancel := context.WithCancel(context.WithoutCancel(ctx))
96		-	l := &listener{udp: pc, tcp: tl, cancel: cancel}
	111	+	l := &listener{udp: pc, tcp: tl, cancel: cancel, inflight: make(chan struct{}, MaxInflight), conns: make(chan struct{}, MaxTCPConns)}
97	112		f.listeners[addr] = l
98	113		log := zerowrap.FromCtx(ctx)
99	114		l.done.Add(2)
100		-	go func() { defer l.done.Done(); f.serveUDP(lctx, pc, log) }()
101		-	go func() { defer l.done.Done(); f.serveTCP(lctx, tl, log) }()
	115	+	go func() { defer l.done.Done(); f.serveUDP(lctx, l, log) }()
	116	+	go func() { defer l.done.Done(); f.serveTCP(lctx, l, log) }()
102	117		return nil
103	118		}
104	119

@@ -142,18 +157,19 @@ func (f *Forwarder) CloseAll() {
142	157		}
143	158		}
144	159
145		-	func (f *Forwarder) acquire(ctx context.Context) bool {
146		-	select {
147		-	case f.sem <- struct{}{}:
148		-	return true
149		-	case <-ctx.Done():
150		-	return false
	160	+	// Dropped returns how many UDP queries the listener on addr dropped because it
	161	+	// was saturated.
	162	+	func (f *Forwarder) Dropped(addr netip.Addr) int64 {
	163	+	f.mu.Lock()
	164	+	defer f.mu.Unlock()
	165	+	if l, ok := f.listeners[addr]; ok {
	166	+	return l.dropped.Load()
151	167		}
	168	+	return 0
152	169		}
153	170
154		-	func (f *Forwarder) release() { <-f.sem }
155		-
156		-	func (f *Forwarder) serveUDP(ctx context.Context, pc net.PacketConn, log zerowrap.Logger) {
	171	+	func (f Forwarder) serveUDP(ctx context.Context, l listener, log zerowrap.Logger) {
	172	+	pc := l.udp
157	173		buf := make([]byte, maxUDP)
158	174		for {
159	175		n, from, err := pc.ReadFrom(buf)

@@ -166,12 +182,17 @@ func (f *Forwarder) serveUDP(ctx context.Context, pc net.PacketConn, log zerowra
166	182		if n < hdrLen {
167	183		continue
168	184		}
169		-	q := append([]byte(nil), buf[:n]...)
170		-	if !f.acquire(ctx) {
171		-	return
	185	+	select {
	186	+	case l.inflight <- struct{}{}:
	187	+	default:
	188	+	// Saturated: drop. The client retries; blocking here would let one
	189	+	// guest's flood stall the read loop.
	190	+	l.dropped.Add(1)
	191	+	continue
172	192		}
	193	+	q := append([]byte(nil), buf[:n]...)
173	194		go func() {
174		-	defer f.release()
	195	+	defer func() { <-l.inflight }()
175	196		resp := f.exchange(ctx, q, false)
176	197		if len(resp) > 0 {
177	198		_, _ = pc.WriteTo(resp, from)

@@ -180,36 +201,52 @@ func (f *Forwarder) serveUDP(ctx context.Context, pc net.PacketConn, log zerowra
180	201		}
181	202		}
182	203
183		-	func (f *Forwarder) serveTCP(ctx context.Context, l net.Listener, log zerowrap.Logger) {
	204	+	func (f Forwarder) serveTCP(ctx context.Context, l listener, log zerowrap.Logger) {
184	205		var wg sync.WaitGroup
185	206		defer wg.Wait()
186	207		for {
187		-	c, err := l.Accept()
	208	+	c, err := l.tcp.Accept()
188	209		if err != nil {
189	210		if ctx.Err() == nil && !errors.Is(err, net.ErrClosed) {
190	211		log.Warn().Err(err).Msg("dns forwarder: tcp accept")
191	212		}
192	213		return
193	214		}
	215	+	select {
	216	+	case l.conns <- struct{}{}:
	217	+	default:
	218	+	_ = c.Close() // too many open connections on this listener
	219	+	continue
	220	+	}
194	221		wg.Add(1)
195	222		go func() {
196	223		defer wg.Done()
	224	+	defer func() { <-l.conns }()
197	225		defer func() { _ = c.Close() }()
198	226		stop := context.AfterFunc(ctx, func() { _ = c.Close() })
199	227		defer stop()
200		-	f.serveConn(ctx, c)
	228	+	// Absolute deadline from accept: neither idle nor slow-loris
	229	+	// connections keep their slot past TCPLifetime.
	230	+	deadline := time.Now().Add(f.Lifetime)
	231	+	_ = c.SetDeadline(deadline)
	232	+	f.serveConn(ctx, l, c, deadline)
201	233		}()
202	234		}
203	235		}
204	236
205		-	func (f *Forwarder) serveConn(ctx context.Context, c net.Conn) {
	237	+	func (f Forwarder) serveConn(ctx context.Context, l listener, c net.Conn, deadline time.Time) {
206	238		for range TCPMaxQueries {
207		-	_ = c.SetReadDeadline(time.Now().Add(TCPIdleTimeout))
208		-	var l [2]byte
209		-	if _, err := io.ReadFull(c, l[:]); err != nil {
	239	+	// The idle timeout can only shorten the absolute deadline.
	240	+	dl := time.Now().Add(TCPIdleTimeout)
	241	+	if deadline.Before(dl) {
	242	+	dl = deadline
	243	+	}
	244	+	_ = c.SetReadDeadline(dl)
	245	+	var lb [2]byte
	246	+	if _, err := io.ReadFull(c, lb[:]); err != nil {
210	247		return
211	248		}
212		-	n := int(binary.BigEndian.Uint16(l[:]))
	249	+	n := int(binary.BigEndian.Uint16(lb[:]))
213	250		if n < hdrLen {
214	251		return
215	252		}

@@ -217,18 +254,19 @@ func (f *Forwarder) serveConn(ctx context.Context, c net.Conn) {
217	254		if _, err := io.ReadFull(c, q); err != nil {
218	255		return
219	256		}
220		-	if !f.acquire(ctx) {
221		-	return
	257	+	select {
	258	+	case l.inflight <- struct{}{}:
	259	+	default:
	260	+	return // saturated: close, the client retries
222	261		}
223	262		resp := f.exchange(ctx, q, true)
224		-	f.release()
	263	+	<-l.inflight
225	264		if len(resp) == 0 \|\| len(resp) > 0xffff {
226	265		return
227	266		}
228	267		out := make([]byte, 2+len(resp))
229	268		binary.BigEndian.PutUint16(out, uint16(len(resp)))
230	269		copy(out[2:], resp)
231		-	_ = c.SetWriteDeadline(time.Now().Add(TCPIdleTimeout))
232	270		if _, err := c.Write(out); err != nil {
233	271		return
234	272		}


@@ -6,9 +6,12 @@
6		// policy lets guests reach nothing else on the host. It does not parse queries	6		// policy lets guests reach nothing else on the host. It does not parse queries
7		// beyond what framing needs, so it supports every record type and EDNS.	7		// beyond what framing needs, so it supports every record type and EDNS.
8		//	8		//
9	-	// Abuse limits: at most MaxInflight concurrent upstream exchanges across all	9	+	// Abuse limits, per listener (one listener per guest, so one guest cannot
10	-	// listeners, a deadline on every upstream exchange, idle and total timeouts on	10	+	// starve another): at most MaxInflight concurrent upstream exchanges (UDP
11	-	// TCP connections, and a cap on TCP queries per connection.	11	+	// queries beyond that are dropped, never queued), at most MaxTCPConns open TCP
			12	+	// connections, an absolute lifetime of TCPLifetime per TCP connection (a client
			13	+	// cannot hold a slot by trickling bytes), a deadline on every upstream
			14	+	// exchange, and a cap on queries per TCP connection.
12		package dnsfwd	15		package dnsfwd
13			16
14		import (	17		import (

@@ -21,18 +24,24 @@ import (
21		"net/netip"	24		"net/netip"
22		"strconv"	25		"strconv"
23		"sync"	26		"sync"
			27	+	"sync/atomic"
24		"time"	28		"time"
25			29
26		"github.com/bnema/zerowrap"	30		"github.com/bnema/zerowrap"
27		)	31		)
28			32
29		const (	33		const (
30	-	// MaxInflight bounds concurrent upstream exchanges.	34	+	// TCPLifetime is the absolute lifetime of a TCP connection, set at accept.
31	-	MaxInflight = 256	35	+	TCPLifetime = 30 * time.Second
32		// UpstreamTimeout is the deadline of one upstream attempt.	36		// UpstreamTimeout is the deadline of one upstream attempt.
33		UpstreamTimeout = 3 * time.Second	37		UpstreamTimeout = 3 * time.Second
34		// TCPIdleTimeout closes TCP client connections that stay silent.	38		// TCPIdleTimeout closes TCP client connections that stay silent.
35		TCPIdleTimeout = 10 * time.Second	39		TCPIdleTimeout = 10 * time.Second
			40	+
			41	+	// MaxInflight bounds concurrent upstream exchanges per listener.
			42	+	MaxInflight = 32
			43	+	// MaxTCPConns bounds open TCP connections per listener.
			44	+	MaxTCPConns = 16
36		// TCPMaxQueries caps queries per TCP connection.	45		// TCPMaxQueries caps queries per TCP connection.
37		TCPMaxQueries = 64	46		TCPMaxQueries = 64
38			47

@@ -44,7 +53,8 @@ const (
44		type Forwarder struct {	53		type Forwarder struct {
45		upstreams func() []string	54		upstreams func() []string
46		port int	55		port int
47	-	sem chan struct{}	56	+	// Lifetime overrides TCPLifetime; set before Listen (tests).
			57	+	Lifetime time.Duration
48			58
49		mu sync.Mutex	59		mu sync.Mutex
50		listeners map[netip.Addr]*listener	60		listeners map[netip.Addr]*listener

@@ -55,6 +65,11 @@ type listener struct {
55		tcp net.Listener	65		tcp net.Listener
56		cancel context.CancelFunc	66		cancel context.CancelFunc
57		done sync.WaitGroup	67		done sync.WaitGroup
			68	+	// inflight and conns are the per-listener limits (counting semaphores).
			69	+	inflight chan struct{}
			70	+	conns chan struct{}
			71	+	// Dropped counts UDP queries dropped because the listener was saturated.
			72	+	dropped atomic.Int64
58		}	73		}
59			74
60		// New returns a Forwarder. upstreams returns the "host:port" resolvers to try	75		// New returns a Forwarder. upstreams returns the "host:port" resolvers to try

@@ -64,7 +79,7 @@ func New(upstreams func() []string, port int) *Forwarder {
64		return &Forwarder{	79		return &Forwarder{
65		upstreams: upstreams,	80		upstreams: upstreams,
66		port: port,	81		port: port,
67	-	sem: make(chan struct{}, MaxInflight),	82	+	Lifetime: TCPLifetime,
68		listeners: map[netip.Addr]*listener{},	83		listeners: map[netip.Addr]*listener{},
69		}	84		}
70		}	85		}

@@ -93,12 +108,12 @@ func (f *Forwarder) Listen(ctx context.Context, addr netip.Addr) error {
93		return fmt.Errorf("dnsfwd: listen tcp %s: %w", hp, err)	108		return fmt.Errorf("dnsfwd: listen tcp %s: %w", hp, err)
94		}	109		}
95		lctx, cancel := context.WithCancel(context.WithoutCancel(ctx))	110		lctx, cancel := context.WithCancel(context.WithoutCancel(ctx))
96	-	l := &listener{udp: pc, tcp: tl, cancel: cancel}	111	+	l := &listener{udp: pc, tcp: tl, cancel: cancel, inflight: make(chan struct{}, MaxInflight), conns: make(chan struct{}, MaxTCPConns)}
97		f.listeners[addr] = l	112		f.listeners[addr] = l
98		log := zerowrap.FromCtx(ctx)	113		log := zerowrap.FromCtx(ctx)
99		l.done.Add(2)	114		l.done.Add(2)
100	-	go func() { defer l.done.Done(); f.serveUDP(lctx, pc, log) }()	115	+	go func() { defer l.done.Done(); f.serveUDP(lctx, l, log) }()
101	-	go func() { defer l.done.Done(); f.serveTCP(lctx, tl, log) }()	116	+	go func() { defer l.done.Done(); f.serveTCP(lctx, l, log) }()
102		return nil	117		return nil
103		}	118		}
104			119

@@ -142,18 +157,19 @@ func (f *Forwarder) CloseAll() {
142		}	157		}
143		}	158		}
144			159
145	-	func (f *Forwarder) acquire(ctx context.Context) bool {	160	+	// Dropped returns how many UDP queries the listener on addr dropped because it
146	-	select {	161	+	// was saturated.
147	-	case f.sem <- struct{}{}:	162	+	func (f *Forwarder) Dropped(addr netip.Addr) int64 {
148	-	return true	163	+	f.mu.Lock()
149	-	case <-ctx.Done():	164	+	defer f.mu.Unlock()
150	-	return false	165	+	if l, ok := f.listeners[addr]; ok {
			166	+	return l.dropped.Load()
151		}	167		}
			168	+	return 0
152		}	169		}
153			170
154	-	func (f *Forwarder) release() { <-f.sem }	171	+	func (f Forwarder) serveUDP(ctx context.Context, l listener, log zerowrap.Logger) {
155	-		172	+	pc := l.udp
156	-	func (f *Forwarder) serveUDP(ctx context.Context, pc net.PacketConn, log zerowrap.Logger) {
157		buf := make([]byte, maxUDP)	173		buf := make([]byte, maxUDP)
158		for {	174		for {
159		n, from, err := pc.ReadFrom(buf)	175		n, from, err := pc.ReadFrom(buf)

@@ -166,12 +182,17 @@ func (f *Forwarder) serveUDP(ctx context.Context, pc net.PacketConn, log zerowra
166		if n < hdrLen {	182		if n < hdrLen {
167		continue	183		continue
168		}	184		}
169	-	q := append([]byte(nil), buf[:n]...)	185	+	select {
170	-	if !f.acquire(ctx) {	186	+	case l.inflight <- struct{}{}:
171	-	return	187	+	default:
			188	+	// Saturated: drop. The client retries; blocking here would let one
			189	+	// guest's flood stall the read loop.
			190	+	l.dropped.Add(1)
			191	+	continue
172		}	192		}
			193	+	q := append([]byte(nil), buf[:n]...)
173		go func() {	194		go func() {
174	-	defer f.release()	195	+	defer func() { <-l.inflight }()
175		resp := f.exchange(ctx, q, false)	196		resp := f.exchange(ctx, q, false)
176		if len(resp) > 0 {	197		if len(resp) > 0 {
177		_, _ = pc.WriteTo(resp, from)	198		_, _ = pc.WriteTo(resp, from)

@@ -180,36 +201,52 @@ func (f *Forwarder) serveUDP(ctx context.Context, pc net.PacketConn, log zerowra
180		}	201		}
181		}	202		}
182			203
183	-	func (f *Forwarder) serveTCP(ctx context.Context, l net.Listener, log zerowrap.Logger) {	204	+	func (f Forwarder) serveTCP(ctx context.Context, l listener, log zerowrap.Logger) {
184		var wg sync.WaitGroup	205		var wg sync.WaitGroup
185		defer wg.Wait()	206		defer wg.Wait()
186		for {	207		for {
187	-	c, err := l.Accept()	208	+	c, err := l.tcp.Accept()
188		if err != nil {	209		if err != nil {
189		if ctx.Err() == nil && !errors.Is(err, net.ErrClosed) {	210		if ctx.Err() == nil && !errors.Is(err, net.ErrClosed) {
190		log.Warn().Err(err).Msg("dns forwarder: tcp accept")	211		log.Warn().Err(err).Msg("dns forwarder: tcp accept")
191		}	212		}
192		return	213		return
193		}	214		}
			215	+	select {
			216	+	case l.conns <- struct{}{}:
			217	+	default:
			218	+	_ = c.Close() // too many open connections on this listener
			219	+	continue
			220	+	}
194		wg.Add(1)	221		wg.Add(1)
195		go func() {	222		go func() {
196		defer wg.Done()	223		defer wg.Done()
			224	+	defer func() { <-l.conns }()
197		defer func() { _ = c.Close() }()	225		defer func() { _ = c.Close() }()
198		stop := context.AfterFunc(ctx, func() { _ = c.Close() })	226		stop := context.AfterFunc(ctx, func() { _ = c.Close() })
199		defer stop()	227		defer stop()
200	-	f.serveConn(ctx, c)	228	+	// Absolute deadline from accept: neither idle nor slow-loris
			229	+	// connections keep their slot past TCPLifetime.
			230	+	deadline := time.Now().Add(f.Lifetime)
			231	+	_ = c.SetDeadline(deadline)
			232	+	f.serveConn(ctx, l, c, deadline)
201		}()	233		}()
202		}	234		}
203		}	235		}
204			236
205	-	func (f *Forwarder) serveConn(ctx context.Context, c net.Conn) {	237	+	func (f Forwarder) serveConn(ctx context.Context, l listener, c net.Conn, deadline time.Time) {
206		for range TCPMaxQueries {	238		for range TCPMaxQueries {
207	-	_ = c.SetReadDeadline(time.Now().Add(TCPIdleTimeout))	239	+	// The idle timeout can only shorten the absolute deadline.
208	-	var l [2]byte	240	+	dl := time.Now().Add(TCPIdleTimeout)
209	-	if _, err := io.ReadFull(c, l[:]); err != nil {	241	+	if deadline.Before(dl) {
			242	+	dl = deadline
			243	+	}
			244	+	_ = c.SetReadDeadline(dl)
			245	+	var lb [2]byte
			246	+	if _, err := io.ReadFull(c, lb[:]); err != nil {
210		return	247		return
211		}	248		}
212	-	n := int(binary.BigEndian.Uint16(l[:]))	249	+	n := int(binary.BigEndian.Uint16(lb[:]))
213		if n < hdrLen {	250		if n < hdrLen {
214		return	251		return
215		}	252		}

@@ -217,18 +254,19 @@ func (f *Forwarder) serveConn(ctx context.Context, c net.Conn) {
217		if _, err := io.ReadFull(c, q); err != nil {	254		if _, err := io.ReadFull(c, q); err != nil {
218		return	255		return
219		}	256		}
220	-	if !f.acquire(ctx) {	257	+	select {
221	-	return	258	+	case l.inflight <- struct{}{}:
			259	+	default:
			260	+	return // saturated: close, the client retries
222		}	261		}
223		resp := f.exchange(ctx, q, true)	262		resp := f.exchange(ctx, q, true)
224	-	f.release()	263	+	<-l.inflight
225		if len(resp) == 0 \|\| len(resp) > 0xffff {	264		if len(resp) == 0 \|\| len(resp) > 0xffff {
226		return	265		return
227		}	266		}
228		out := make([]byte, 2+len(resp))	267		out := make([]byte, 2+len(resp))
229		binary.BigEndian.PutUint16(out, uint16(len(resp)))	268		binary.BigEndian.PutUint16(out, uint16(len(resp)))
230		copy(out[2:], resp)	269		copy(out[2:], resp)
231	-	_ = c.SetWriteDeadline(time.Now().Add(TCPIdleTimeout))
232		if _, err := c.Write(out); err != nil {	270		if _, err := c.Write(out); err != nil {
233		return	271		return
234		}	272		}

M internal/dnsfwd/dnsfwd_test.go +145 -0


@@ -5,6 +5,7 @@ package dnsfwd
5	5		import (
6	6		"context"
7	7		"encoding/binary"
	8	+	"errors"
8	9		"io"
9	10		"net"
10	11		"net/netip"

@@ -12,6 +13,8 @@ import (
12	13		"path/filepath"
13	14		"reflect"
14	15		"strings"
	16	+	"sync/atomic"
	17	+	"syscall"
15	18		"testing"
16	19		"time"
17	20		)

@@ -295,3 +298,145 @@ func TestResolversReload(t *testing.T) {
295	298		t.Fatal(got)
296	299		}
297	300		}
	301	+
	302	+	// S2: a listener holds at most MaxTCPConns connections; extra ones are closed
	303	+	// at once, and a free slot is usable again after a close.
	304	+	func TestTCPConnectionCapPerListener(t *testing.T) {
	305	+	up := fakeUpstream(t, answer)
	306	+	_, addr := start(t, up)
	307	+	var held []net.Conn
	308	+	for range MaxTCPConns {
	309	+	c, err := net.Dial("tcp", addr)
	310	+	if err != nil {
	311	+	t.Fatal(err)
	312	+	}
	313	+	held = append(held, c)
	314	+	defer func() { _ = c.Close() }()
	315	+	}
	316	+	// Let the accept loop register all of them.
	317	+	time.Sleep(200 * time.Millisecond)
	318	+	extra, err := net.Dial("tcp", addr)
	319	+	if err != nil {
	320	+	t.Fatal(err)
	321	+	}
	322	+	defer func() { _ = extra.Close() }()
	323	+	_ = extra.SetReadDeadline(time.Now().Add(3 * time.Second))
	324	+	if _, err := extra.Read(make([]byte, 1)); err != io.EOF && !errors.Is(err, syscall.ECONNRESET) {
	325	+	t.Fatalf("the connection over the cap must be closed by the server, got %v", err)
	326	+	}
	327	+	// Free one slot: a new connection is served.
	328	+	_ = held[0].Close()
	329	+	time.Sleep(200 * time.Millisecond)
	330	+	c, err := net.Dial("tcp", addr)
	331	+	if err != nil {
	332	+	t.Fatal(err)
	333	+	}
	334	+	defer func() { _ = c.Close() }()
	335	+	_ = c.SetDeadline(time.Now().Add(5 * time.Second))
	336	+	q := query(3, "slot.test")
	337	+	_, _ = c.Write(append([]byte{byte(len(q) >> 8), byte(len(q))}, q...))
	338	+	var l [2]byte
	339	+	if _, err := io.ReadFull(c, l[:]); err != nil {
	340	+	t.Fatalf("a freed slot must serve again: %v", err)
	341	+	}
	342	+	}
	343	+
	344	+	// S2: the lifetime is absolute from accept: a client that keeps the connection
	345	+	// alive by trickling bytes still loses it.
	346	+	func TestTCPAbsoluteDeadline(t *testing.T) {
	347	+	up := fakeUpstream(t, answer)
	348	+	f := New(func() []string { return []string{up} }, 0)
	349	+	f.Lifetime = 600 * time.Millisecond
	350	+	ip := netip.MustParseAddr("127.0.0.1")
	351	+	if err := f.Listen(context.Background(), ip); err != nil {
	352	+	t.Skip(err)
	353	+	}
	354	+	t.Cleanup(f.CloseAll)
	355	+	la, _ := f.Addr(ip)
	356	+	addr := la.String()
	357	+	c, err := net.Dial("tcp", addr)
	358	+	if err != nil {
	359	+	t.Fatal(err)
	360	+	}
	361	+	defer func() { _ = c.Close() }()
	362	+	start := time.Now()
	363	+	closed := make(chan struct{})
	364	+	go func() {
	365	+	_, _ = io.Copy(io.Discard, c) // returns when the server closes
	366	+	close(closed)
	367	+	}()
	368	+	// Trickle one byte every 100 ms, never finishing a frame: only an idle
	369	+	// timeout (10 s) would let this live; the absolute deadline must not.
	370	+	go func() {
	371	+	for i := 0; i < 40; i++ {
	372	+	if _, err := c.Write([]byte{0}); err != nil {
	373	+	return
	374	+	}
	375	+	time.Sleep(100 * time.Millisecond)
	376	+	}
	377	+	}()
	378	+	select {
	379	+	case <-closed:
	380	+	if d := time.Since(start); d > 3*time.Second {
	381	+	t.Fatalf("closed after %v, want about %v", d, f.Lifetime)
	382	+	}
	383	+	case <-time.After(5 * time.Second):
	384	+	t.Fatal("a trickling client kept its connection past the absolute deadline")
	385	+	}
	386	+	}
	387	+
	388	+	// S2/S9: a saturated listener drops UDP queries instead of blocking its read
	389	+	// loop, and another listener (another guest) keeps answering.
	390	+	func TestUDPSaturationDropsAndIsolatesListeners(t *testing.T) {
	391	+	release := make(chan struct{})
	392	+	slow := fakeUpstream(t, func(q []byte) []byte { <-release; return answer(q) })
	393	+	fast := fakeUpstream(t, answer)
	394	+	var useSlow atomic.Bool
	395	+	useSlow.Store(true)
	396	+	f := New(func() []string {
	397	+	if useSlow.Load() {
	398	+	return []string{slow}
	399	+	}
	400	+	return []string{fast}
	401	+	}, 0)
	402	+	a1, a2 := netip.MustParseAddr("127.0.0.1"), netip.MustParseAddr("127.0.0.2")
	403	+	if err := f.Listen(context.Background(), a1); err != nil {
	404	+	t.Skip(err)
	405	+	}
	406	+	if err := f.Listen(context.Background(), a2); err != nil {
	407	+	t.Skip(err)
	408	+	}
	409	+	t.Cleanup(f.CloseAll)
	410	+	t.Cleanup(func() { close(release) })
	411	+	l1, _ := f.Addr(a1)
	412	+	flood, err := net.Dial("udp", l1.String())
	413	+	if err != nil {
	414	+	t.Fatal(err)
	415	+	}
	416	+	defer func() { _ = flood.Close() }()
	417	+	for i := 0; i < MaxInflight*3; i++ {
	418	+	_, _ = flood.Write(query(uint16(i), "flood.test"))
	419	+	}
	420	+	deadline := time.Now().Add(3 * time.Second)
	421	+	for f.Dropped(a1) == 0 && time.Now().Before(deadline) {
	422	+	time.Sleep(20 * time.Millisecond)
	423	+	}
	424	+	if f.Dropped(a1) == 0 {
	425	+	t.Fatal("a saturated listener must drop queries")
	426	+	}
	427	+	// The other listener is unaffected: point the resolver at the fast upstream.
	428	+	useSlow.Store(false)
	429	+	l2, _ := f.Addr(a2)
	430	+	c, err := net.Dial("udp", l2.String())
	431	+	if err != nil {
	432	+	t.Fatal(err)
	433	+	}
	434	+	defer func() { _ = c.Close() }()
	435	+	_ = c.SetDeadline(time.Now().Add(4 * time.Second))
	436	+	_, _ = c.Write(query(77, "other-guest.test"))
	437	+	buf := make([]byte, 512)
	438	+	n, err := c.Read(buf)
	439	+	if err != nil \|\| string(buf[n-6:n]) != "ANSWER" {
	440	+	t.Fatalf("another listener must not be starved by the flood: n=%d err=%v", n, err)
	441	+	}
	442	+	}


@@ -5,6 +5,7 @@ package dnsfwd
5	import (	5		import (
6	"context"	6		"context"
7	"encoding/binary"	7		"encoding/binary"
		8	+	"errors"
8	"io"	9		"io"
9	"net"	10		"net"
10	"net/netip"	11		"net/netip"

@@ -12,6 +13,8 @@ import (
12	"path/filepath"	13		"path/filepath"
13	"reflect"	14		"reflect"
14	"strings"	15		"strings"
		16	+	"sync/atomic"
		17	+	"syscall"
15	"testing"	18		"testing"
16	"time"	19		"time"
17	)	20		)

@@ -295,3 +298,145 @@ func TestResolversReload(t *testing.T) {
295	t.Fatal(got)	298		t.Fatal(got)
296	}	299		}
297	}	300		}
		301	+
		302	+	// S2: a listener holds at most MaxTCPConns connections; extra ones are closed
		303	+	// at once, and a free slot is usable again after a close.
		304	+	func TestTCPConnectionCapPerListener(t *testing.T) {
		305	+	up := fakeUpstream(t, answer)
		306	+	_, addr := start(t, up)
		307	+	var held []net.Conn
		308	+	for range MaxTCPConns {
		309	+	c, err := net.Dial("tcp", addr)
		310	+	if err != nil {
		311	+	t.Fatal(err)
		312	+	}
		313	+	held = append(held, c)
		314	+	defer func() { _ = c.Close() }()
		315	+	}
		316	+	// Let the accept loop register all of them.
		317	+	time.Sleep(200 * time.Millisecond)
		318	+	extra, err := net.Dial("tcp", addr)
		319	+	if err != nil {
		320	+	t.Fatal(err)
		321	+	}
		322	+	defer func() { _ = extra.Close() }()
		323	+	_ = extra.SetReadDeadline(time.Now().Add(3 * time.Second))
		324	+	if _, err := extra.Read(make([]byte, 1)); err != io.EOF && !errors.Is(err, syscall.ECONNRESET) {
		325	+	t.Fatalf("the connection over the cap must be closed by the server, got %v", err)
		326	+	}
		327	+	// Free one slot: a new connection is served.
		328	+	_ = held[0].Close()
		329	+	time.Sleep(200 * time.Millisecond)
		330	+	c, err := net.Dial("tcp", addr)
		331	+	if err != nil {
		332	+	t.Fatal(err)
		333	+	}
		334	+	defer func() { _ = c.Close() }()
		335	+	_ = c.SetDeadline(time.Now().Add(5 * time.Second))
		336	+	q := query(3, "slot.test")
		337	+	_, _ = c.Write(append([]byte{byte(len(q) >> 8), byte(len(q))}, q...))
		338	+	var l [2]byte
		339	+	if _, err := io.ReadFull(c, l[:]); err != nil {
		340	+	t.Fatalf("a freed slot must serve again: %v", err)
		341	+	}
		342	+	}
		343	+
		344	+	// S2: the lifetime is absolute from accept: a client that keeps the connection
		345	+	// alive by trickling bytes still loses it.
		346	+	func TestTCPAbsoluteDeadline(t *testing.T) {
		347	+	up := fakeUpstream(t, answer)
		348	+	f := New(func() []string { return []string{up} }, 0)
		349	+	f.Lifetime = 600 * time.Millisecond
		350	+	ip := netip.MustParseAddr("127.0.0.1")
		351	+	if err := f.Listen(context.Background(), ip); err != nil {
		352	+	t.Skip(err)
		353	+	}
		354	+	t.Cleanup(f.CloseAll)
		355	+	la, _ := f.Addr(ip)
		356	+	addr := la.String()
		357	+	c, err := net.Dial("tcp", addr)
		358	+	if err != nil {
		359	+	t.Fatal(err)
		360	+	}
		361	+	defer func() { _ = c.Close() }()
		362	+	start := time.Now()
		363	+	closed := make(chan struct{})
		364	+	go func() {
		365	+	_, _ = io.Copy(io.Discard, c) // returns when the server closes
		366	+	close(closed)
		367	+	}()
		368	+	// Trickle one byte every 100 ms, never finishing a frame: only an idle
		369	+	// timeout (10 s) would let this live; the absolute deadline must not.
		370	+	go func() {
		371	+	for i := 0; i < 40; i++ {
		372	+	if _, err := c.Write([]byte{0}); err != nil {
		373	+	return
		374	+	}
		375	+	time.Sleep(100 * time.Millisecond)
		376	+	}
		377	+	}()
		378	+	select {
		379	+	case <-closed:
		380	+	if d := time.Since(start); d > 3*time.Second {
		381	+	t.Fatalf("closed after %v, want about %v", d, f.Lifetime)
		382	+	}
		383	+	case <-time.After(5 * time.Second):
		384	+	t.Fatal("a trickling client kept its connection past the absolute deadline")
		385	+	}
		386	+	}
		387	+
		388	+	// S2/S9: a saturated listener drops UDP queries instead of blocking its read
		389	+	// loop, and another listener (another guest) keeps answering.
		390	+	func TestUDPSaturationDropsAndIsolatesListeners(t *testing.T) {
		391	+	release := make(chan struct{})
		392	+	slow := fakeUpstream(t, func(q []byte) []byte { <-release; return answer(q) })
		393	+	fast := fakeUpstream(t, answer)
		394	+	var useSlow atomic.Bool
		395	+	useSlow.Store(true)
		396	+	f := New(func() []string {
		397	+	if useSlow.Load() {
		398	+	return []string{slow}
		399	+	}
		400	+	return []string{fast}
		401	+	}, 0)
		402	+	a1, a2 := netip.MustParseAddr("127.0.0.1"), netip.MustParseAddr("127.0.0.2")
		403	+	if err := f.Listen(context.Background(), a1); err != nil {
		404	+	t.Skip(err)
		405	+	}
		406	+	if err := f.Listen(context.Background(), a2); err != nil {
		407	+	t.Skip(err)
		408	+	}
		409	+	t.Cleanup(f.CloseAll)
		410	+	t.Cleanup(func() { close(release) })
		411	+	l1, _ := f.Addr(a1)
		412	+	flood, err := net.Dial("udp", l1.String())
		413	+	if err != nil {
		414	+	t.Fatal(err)
		415	+	}
		416	+	defer func() { _ = flood.Close() }()
		417	+	for i := 0; i < MaxInflight*3; i++ {
		418	+	_, _ = flood.Write(query(uint16(i), "flood.test"))
		419	+	}
		420	+	deadline := time.Now().Add(3 * time.Second)
		421	+	for f.Dropped(a1) == 0 && time.Now().Before(deadline) {
		422	+	time.Sleep(20 * time.Millisecond)
		423	+	}
		424	+	if f.Dropped(a1) == 0 {
		425	+	t.Fatal("a saturated listener must drop queries")
		426	+	}
		427	+	// The other listener is unaffected: point the resolver at the fast upstream.
		428	+	useSlow.Store(false)
		429	+	l2, _ := f.Addr(a2)
		430	+	c, err := net.Dial("udp", l2.String())
		431	+	if err != nil {
		432	+	t.Fatal(err)
		433	+	}
		434	+	defer func() { _ = c.Close() }()
		435	+	_ = c.SetDeadline(time.Now().Add(4 * time.Second))
		436	+	_, _ = c.Write(query(77, "other-guest.test"))
		437	+	buf := make([]byte, 512)
		438	+	n, err := c.Read(buf)
		439	+	if err != nil \|\| string(buf[n-6:n]) != "ANSWER" {
		440	+	t.Fatalf("another listener must not be starved by the flood: n=%d err=%v", n, err)
		441	+	}
		442	+	}