diff --git a/STATUS.md b/STATUS.md index 373d850..b22a4a8 100644 --- a/STATUS.md +++ b/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 diff --git a/internal/engine/engine.go b/internal/engine/engine.go index 7a0f2cf..37a5bc5 100644 --- a/internal/engine/engine.go +++ b/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() diff --git a/internal/engine/timeout_test.go b/internal/engine/timeout_test.go new file mode 100644 index 0000000..d8fd76c --- /dev/null +++ b/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) + } + }) + } +} diff --git a/internal/resolver/pool.go b/internal/resolver/pool.go index 53bcfbb..74aaee2 100644 --- a/internal/resolver/pool.go +++ b/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) - resp, rtt, err := client.ExchangeContext(ctx, q.Copy(), up.addr) + caller := ctx + 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 - // fails here and we return the truncated answer; the client retries over - // TCP itself. + // Retry over TCP within the same attempt deadline. If TCP fails before + // the budget expires, retain the truncated answer for the client to retry. 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. diff --git a/internal/resolver/timeout_test.go b/internal/resolver/timeout_test.go new file mode 100644 index 0000000..0de4952 --- /dev/null +++ b/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() + } + }) + } +} diff --git a/internal/server/stats_test.go b/internal/server/stats_test.go index 4d820e2..4c95882 100644 --- a/internal/server/stats_test.go +++ b/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 {