Enforce upstream attempt and overall query deadlines

The initial upstream exchange ignored its supplied timeout and used hardcoded client limits. Bound each attempt with a context, copy the shared client for its timeout, and share the attempt budget with UDP-to-TCP fallback. Close private connections on cancellation to interrupt reads, and exclude caller cancellation from upstream failures so parallel losers remain healthy.

The normal engine pipeline never applied server.timeouts.query, allowing failover to outlive the configured query budget. Apply a hot-reloaded overall context deadline at Handle while retaining earlier caller deadlines. Add silent UDP/TCP/TLS, cancellation, fallback, failover, and reload regressions.

The server inflight test assumed receiving a TCP reply meant its sent counter had already been incremented. Wait for both completion counters before asserting them, preserving the existing assertions while removing the scheduling race.

STATUS.md

@@ -8,6 +8,25 @@ round before that brought the console onto the Go side of the API contract, so
 both halves of that are done. `AGENTS.md` holds the decisions, invariants and
 working rules; this file holds state and the plan.
 
+## Timeout enforcement update
+
+Foreground queries now enforce `server.timeouts.query` at `engine.Handle`,
+including all failover attempts. Reloads update the limit atomically; earlier
+caller deadlines and cancellation are preserved. `server.timeouts.upstream`
+bounds each complete upstream attempt, with UDP and its TCP truncation retry
+sharing one budget. Per-attempt client copies avoid mutating memoised clients,
+and cancellation closes the private connection to interrupt blocked reads.
+Caller cancellation (including parallel-race losers) does not degrade upstream
+health. An exhausted fallback budget returns an error; an immediate TCP failure
+still preserves the truncated UDP answer for the client to retry.
+
+Regression tests cover silent UDP/TCP/TLS upstreams, active-read cancellation,
+fallback budgeting, failover, configured overall deadlines and hot reload.
+The observed transient `.local` search-expansion incident remains unexplained:
+these timeout defects bound stalls but do not establish why upstream replies
+were absent at the time. The orchestrator's absolute RabbitMQ hostname remains
+the independent search-expansion mitigation.
+
 ## Conditional forwarding update
 
 `upstream.forward_zones` now routes an apex and its descendants to dedicated

internal/engine/engine.go

@@ -135,6 +135,8 @@ type Engine struct {
 
 	// upstreamTimeout bounds a single upstream exchange.
 	upstreamTimeout atomic.Int64
+	// queryTimeout bounds the entire foreground pipeline, including failover.
+	queryTimeout atomic.Int64
 
 	// groupStats tracks per-group counters for the UI.
 	groupStats *groupStatTable
@@ -212,6 +214,7 @@ func (e *Engine) applyConfig(cfg *config.Config) {
 	e.sem.Store(&sem)
 	e.semSize.Store(int64(semSize))
 	e.upstreamTimeout.Store(int64(cfg.Server.Timeouts.Upstre.D()))
+	e.queryTimeout.Store(int64(cfg.Server.Timeouts.Query.D()))
 
 	if e.aclH != nil {
 		e.aclH.Set(acl.Compile(cfg.ACL))
@@ -254,6 +257,11 @@ func (e *Engine) GroupStats() []acl.GroupSummary {
 // Handle runs the full pipeline for one request and returns the response to
 // send. It never returns nil: a synthetic SERVFAIL is the worst case.
 func (e *Engine) Handle(ctx context.Context, req Request) Response {
+	if timeout := time.Duration(e.queryTimeout.Load()); timeout > 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, timeout)
+		defer cancel()
+	}
 	start := req.Start
 	if start.IsZero() {
 		start = time.Now()

internal/engine/timeout_test.go

@@ -0,0 +1,52 @@
+package engine
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	"git.k8s-git-k6n.net/mats/fancy-dns/internal/config"
+	"github.com/miekg/dns"
+)
+
+func TestQueryDeadlineAcrossFailoverAndReload(t *testing.T) {
+	var servers []config.UpstreamServer
+	for i := 0; i < 3; i++ {
+		pc, err := net.ListenPacket("udp", "127.0.0.1:0")
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Cleanup(func() { _ = pc.Close() })
+		servers = append(servers, config.UpstreamServer{Addr: pc.LocalAddr().String(), Proto: "udp"})
+	}
+	h := newHarness(t, hopt{upstreams: servers, timeout: 100 * time.Millisecond, server: func(s *config.Server) { s.Timeouts.Query = config.Duration(150 * time.Millisecond) }})
+	cfg := h.pool.Config()
+	cfg.Policy = "failover"
+	cfg.MaxAttempts = 3
+	h.pool.Rebind(cfg)
+	for _, tc := range []struct {
+		name                 string
+		query, caller, limit time.Duration
+	}{
+		{"configured", 150 * time.Millisecond, time.Second, 250 * time.Millisecond},
+		{"reload", 40 * time.Millisecond, time.Second, 90 * time.Millisecond},
+		{"earlier-caller", time.Second, 30 * time.Millisecond, 90 * time.Millisecond},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			next := *h.cfg
+			next.Server.Timeouts.Query = config.Duration(tc.query)
+			if err := h.store.Swap(&next); err != nil {
+				t.Fatal(err)
+			}
+			ctx, cancel := context.WithTimeout(context.Background(), tc.caller)
+			defer cancel()
+			start := time.Now()
+			res := h.eng.Handle(ctx, h.request(tc.name+".test.", dns.TypeA))
+			requireRcode(t, res, dns.RcodeServerFailure)
+			if elapsed := time.Since(start); elapsed > tc.limit {
+				t.Fatalf("query took %v, limit %v", elapsed, tc.limit)
+			}
+		})
+	}
+}

internal/resolver/pool.go

@@ -461,7 +461,7 @@ func (p *Pool) clientFor(u *Upstream) *dns.Client {
 				MinVersion:         tls.VersionTLS12,
 				NextProtos:         []string{"dot"},
 			},
-			Dialer:  &net.Dialer{Timeout: 3 * time.Second},
+			Dialer:  &net.Dialer{},
 			Timeout: 5 * time.Second,
 		}
 	case "tcp":
@@ -475,7 +475,7 @@ func (p *Pool) clientFor(u *Upstream) *dns.Client {
 //
 // want restricts the query to the upstreams an ACL group named, by configured
 // or resolved address; an empty list means the whole pool. timeout bounds one
-// attempt.
+// attempt, including UDP-to-TCP fallback. The caller context bounds all attempts.
 func (p *Pool) Exchange(ctx context.Context, q *dns.Msg, want []string, timeout time.Duration) (*dns.Msg, string, error) {
 	st := p.state.Load()
 	candidates := st.defaults
@@ -518,6 +518,9 @@ func (p *Pool) Exchange(ctx context.Context, q *dns.Msg, want []string, timeout
 		// thing failover exists to avoid.
 		var lastErr error
 		for _, up := range chosen {
+			if err := ctx.Err(); err != nil {
+				return nil, "", err
+			}
 			msg, addr, err := p.exchangeOne(ctx, st.cfg, up, q, timeout)
 			if err == nil && msg != nil && msg.Rcode != dns.RcodeServerFailure {
 				return msg, addr, nil
@@ -627,21 +630,35 @@ func (p *Pool) exchangeOne(ctx context.Context, cfg config.Upstream, up *Upstrea
 	if timeout <= 0 {
 		timeout = 2 * time.Second
 	}
-	client := p.clientFor(up)
+	caller := ctx
-	resp, rtt, err := client.ExchangeContext(ctx, q.Copy(), up.addr)
+	ctx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+	// Copy the memoised client: per-attempt limits must never mutate shared state.
+	client := *p.clientFor(up)
+	client.Timeout = timeout
+	resp, rtt, err := exchangeContext(ctx, &client, q.Copy(), up.addr)
 	if err != nil {
-		up.Failure(err, thresholdFor(up, cfg.BreakerThreshold), cfg.BreakerCooldown.D())
+		// Cancellation by the caller (including losing a parallel race) says
+		// nothing about upstream health. Only an attempt failure counts.
+		if caller.Err() == nil {
+			up.Failure(err, thresholdFor(up, cfg.BreakerThreshold), cfg.BreakerCooldown.D())
+		}
 		return nil, up.addr, err
 	}
 	up.Success(rtt)
 	if !resp.Truncated {
 		return resp, up.addr, nil
 	}
-	// Retry over TCP for the full answer. An upstream that is UDP-only just
+	// Retry over TCP within the same attempt deadline. If TCP fails before
-	// fails here and we return the truncated answer; the client retries over
+	// the budget expires, retain the truncated answer for the client to retry.
-	// TCP itself.
 	tcpClient := &dns.Client{Net: "tcp", Timeout: timeout}
-	resp2, rtt2, err2 := tcpClient.ExchangeContext(ctx, q.Copy(), up.addr)
+	resp2, rtt2, err2 := exchangeContext(ctx, tcpClient, q.Copy(), up.addr)
+	if err := ctx.Err(); err != nil {
+		return nil, up.addr, err
+	}
+	if deadline, ok := ctx.Deadline(); ok && !time.Now().Before(deadline) {
+		return nil, up.addr, context.DeadlineExceeded
+	}
 	if err2 != nil || resp2 == nil || resp2.Truncated {
 		if err2 == nil {
 			err2 = errors.New("still truncated over TCP")
@@ -654,6 +671,26 @@ func (p *Pool) exchangeOne(ctx context.Context, cfg config.Upstream, up *Upstrea
 	return resp2, up.addr, nil
 }
 
+// exchangeContext also interrupts active socket I/O on caller cancellation;
+// miekg/dns applies context deadlines but only checks cancellation while dialing.
+func exchangeContext(ctx context.Context, client *dns.Client, q *dns.Msg, addr string) (*dns.Msg, time.Duration, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, 0, err
+	}
+	conn, err := client.DialContext(ctx, addr)
+	if err != nil {
+		return nil, 0, err
+	}
+	defer conn.Close()
+	stop := context.AfterFunc(ctx, func() { _ = conn.Close() })
+	defer stop()
+	resp, rtt, err := client.ExchangeWithConnContext(ctx, q, conn)
+	if ctx.Err() != nil {
+		return nil, rtt, ctx.Err()
+	}
+	return resp, rtt, err
+}
+
 // HealthCheck probes every upstream with a cheap query, refreshing health
 // state even when no client traffic is flowing. It returns the number of
 // upstreams that answered.

internal/resolver/timeout_test.go

@@ -0,0 +1,106 @@
+package resolver
+
+import (
+	"context"
+	"errors"
+	"net"
+	"testing"
+	"time"
+
+	"git.k8s-git-k6n.net/mats/fancy-dns/internal/config"
+)
+
+func TestAttemptTimeoutSilentUDP(t *testing.T) {
+	pc, err := net.ListenPacket("udp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer pc.Close()
+	p := NewPool(poolCfg(udpServer(pc.LocalAddr().String())))
+	start := time.Now()
+	_, _, err = p.Exchange(context.Background(), query("silent.test."), nil, 50*time.Millisecond)
+	if err == nil || time.Since(start) > 500*time.Millisecond {
+		t.Fatalf("elapsed %v, error %v", time.Since(start), err)
+	}
+}
+
+func TestCallerCancellationInterruptsRead(t *testing.T) {
+	pc, err := net.ListenPacket("udp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer pc.Close()
+	p := NewPool(poolCfg(udpServer(pc.LocalAddr().String())))
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	result := make(chan error, 1)
+	go func() { _, _, err := p.Exchange(ctx, query("cancel.test."), nil, 5*time.Second); result <- err }()
+	_ = pc.SetReadDeadline(time.Now().Add(time.Second))
+	if _, _, err := pc.ReadFrom(make([]byte, 4096)); err != nil {
+		t.Fatal(err)
+	}
+	cancel()
+	select {
+	case err := <-result:
+		if p.All()[0].Stats().Failures != 0 {
+			t.Fatal("caller cancellation counted as an upstream failure")
+		}
+		if !errors.Is(err, context.Canceled) {
+			t.Fatal(err)
+		}
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("cancellation did not interrupt read")
+	}
+}
+
+func TestFallbackSharesAttemptDeadline(t *testing.T) {
+	// Each leg takes 150ms: a fresh budget for TCP would incorrectly succeed.
+	up := newFakeUpstream(t, fakeOpt{delay: 150 * time.Millisecond, truncateUDP: true})
+	p := NewPool(poolCfg(udpServer(up.addr)))
+	start := time.Now()
+	_, _, err := p.Exchange(context.Background(), query("fallback.test."), nil, 250*time.Millisecond)
+	if err == nil || time.Since(start) > 450*time.Millisecond || up.tcpHits.Load() != 1 {
+		t.Fatalf("elapsed %v, TCP hits %d, error %v", time.Since(start), up.tcpHits.Load(), err)
+	}
+}
+
+func TestFailoverAfterAttemptTimeout(t *testing.T) {
+	pc, err := net.ListenPacket("udp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer pc.Close()
+	up := newFakeUpstream(t, fakeOpt{})
+	cfg := poolCfg(udpServer(pc.LocalAddr().String()), udpServer(up.addr))
+	cfg.Policy = "failover"
+	p := NewPool(cfg)
+	start := time.Now()
+	_, addr, err := p.Exchange(context.Background(), query("failover.test."), nil, 50*time.Millisecond)
+	if err != nil || addr != up.addr || time.Since(start) > 500*time.Millisecond {
+		t.Fatalf("elapsed %v, addr %s, error %v", time.Since(start), addr, err)
+	}
+}
+
+func TestAttemptTimeoutSilentStream(t *testing.T) {
+	for _, proto := range []string{"tcp", "tls"} {
+		t.Run(proto, func(t *testing.T) {
+			lc, err := net.Listen("tcp", "127.0.0.1:0")
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer lc.Close()
+			accepted := make(chan net.Conn, 1)
+			go func() { conn, _ := lc.Accept(); accepted <- conn }()
+			p := NewPool(poolCfg(config.UpstreamServer{Addr: lc.Addr().String(), Proto: proto, InsecureSkipVerify: true}))
+			start := time.Now()
+			_, _, err = p.Exchange(context.Background(), query("silent.test."), nil, 50*time.Millisecond)
+			if err == nil || time.Since(start) > 500*time.Millisecond {
+				t.Errorf("elapsed %v, error %v", time.Since(start), err)
+			}
+			_ = lc.Close()
+			if conn := <-accepted; conn != nil {
+				_ = conn.Close()
+			}
+		})
+	}
+}

internal/server/stats_test.go

@@ -488,7 +488,8 @@ func TestInflightTracksTheHandler(t *testing.T) {
 		t.Errorf("A = %q, want 203.0.113.9", a)
 	}
 	waitFor(t, 2*time.Second, "the in-flight count to fall back to zero", func() bool {
-		return h.srv.Stats().Inflight == 0
+		ts := h.srv.ByTransport()["tcp"]
+		return h.srv.Stats().Inflight == 0 && ts.Queries == 1 && ts.Sent == 1
 	})
 	ts := h.srv.ByTransport()["tcp"]
 	if ts.Queries != 1 || ts.Sent != 1 {