fancy-dns/internal/server/server.go

// Package server runs the DNS listeners: plain DNS on UDP/TCP, DNS-over-TLS
// (RFC 7858) and DNS-over-HTTPS (RFC 8484).
//
// All four transports funnel into one handler, so a query behaves identically
// whichever way it arrived. The transports are deliberately thin: parse, hand
// to the engine, write. Anything heavier belongs behind the engine.
package server

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"encoding/base64"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net"
	"net/http"
	"net/netip"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/miekg/dns"
	"golang.org/x/net/http2"
	"golang.org/x/net/http2/h2c"

	"git.k8s-git-k6n.net/mats/fancy-dns/internal/config"
	"git.k8s-git-k6n.net/mats/fancy-dns/internal/engine"
)

// Handler is the interface the transports use to answer a query.
type Handler interface {
	Handle(ctx context.Context, req engine.Request) engine.Response
	Inflight() int64
	Flush()
}

// identityFunc resolves a client IP to its reverse-DNS names. The app installs
// it so the server does not depend on the engine's cache directly.
type identityFunc func(ip string) []string

// tagFunc derives ACL tags from a client IP.
type tagFunc func(ip string) []string

// Server owns every DNS listener.
type Server struct {
	eng     Handler
	cfg     *config.Store
	log     *slog.Logger
	version string

	names atomic.Pointer[identityFunc]
	tags  atomic.Pointer[tagFunc]

	mu        sync.Mutex
	started   bool
	shutdown  context.CancelFunc
	wait      sync.WaitGroup
	listeners []listener
	// counters is keyed by "<transport>.<field>".
	counters sync.Map
	// active is the listener snapshot the API serves.
	active atomic.Pointer[[]ListenerInfo]
}

// listener couples a close function with the info the API reports for it.
type listener struct {
	close func() error
	info  ListenerInfo
}

// ListenerInfo describes one bound listener.
type ListenerInfo struct {
	Transport string `json:"transport"`
	Addr      string `json:"addr"`
	Enabled   bool   `json:"enabled"`
	Running   bool   `json:"running"`
	Err       string `json:"err,omitempty"`
}

// Options configures a Server.
type Options struct {
	// Engine handles queries.
	Engine Handler
	// Config is the hot-reloadable config store; the server subscribes to it.
	Config *config.Store
	// Logger is the structured logger.
	Logger *slog.Logger
	// Version is reported in the DoH server header and the admin API.
	Version string
}

// New builds a Server.
func New(opt Options) *Server {
	log := opt.Logger
	if log == nil {
		log = slog.Default()
	}
	return &Server{
		eng:     opt.Engine,
		cfg:     opt.Config,
		log:     log.With("component", "dns"),
		version: opt.Version,
	}
}

// SetIdentity installs the client reverse-name and tag resolvers.
func (s *Server) SetIdentity(names identityFunc, tags tagFunc) {
	s.names.Store(&names)
	s.tags.Store(&tags)
}

// Start binds every enabled listener and begins serving. It returns once all
// listeners have been bound, so a bind failure surfaces synchronously.
func (s *Server) Start(ctx context.Context) error {
	cfg := s.cfg.Get()
	ctx, cancel := context.WithCancel(ctx)

	s.mu.Lock()
	if s.started {
		s.mu.Unlock()
		cancel()
		return errors.New("server: already started")
	}
	s.started = true
	s.shutdown = cancel
	s.mu.Unlock()

	var failures []error
	for _, addr := range cfg.DNS.UDP {
		if err := s.serveUDP(ctx, addr); err != nil {
			failures = append(failures, err)
		}
	}
	for _, addr := range cfg.DNS.TCP {
		if err := s.serveTCP(ctx, addr); err != nil {
			failures = append(failures, err)
		}
	}
	if cfg.DNS.DoT.Enabled {
		if err := s.serveDoT(ctx, cfg); err != nil {
			failures = append(failures, err)
		}
	}
	if cfg.DNS.DoH.Enabled {
		if err := s.serveDoH(ctx, cfg); err != nil {
			failures = append(failures, err)
		}
	}
	s.publish()
	if err := errors.Join(failures...); err != nil {
		s.log.Warn("some listeners failed to start", "error", err)
	}
	if s.count() == 0 && len(failures) > 0 {
		cancel()
		return fmt.Errorf("server: no listener started: %w", errors.Join(failures...))
	}
	return nil
}

func (s *Server) count() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return len(s.listeners)
}

func (s *Server) add(l listener) {
	s.mu.Lock()
	s.listeners = append(s.listeners, l)
	s.mu.Unlock()
}

func (s *Server) publish() {
	s.mu.Lock()
	infos := make([]ListenerInfo, 0, len(s.listeners))
	for _, l := range s.listeners {
		infos = append(infos, l.info)
	}
	s.mu.Unlock()
	s.active.Store(&infos)
}

// Listeners returns the current listener snapshot.
func (s *Server) Listeners() []ListenerInfo {
	if p := s.active.Load(); p != nil {
		return *p
	}
	return nil
}

// Stop shuts every listener down and waits for their goroutines.
func (s *Server) Stop(ctx context.Context) error {
	s.mu.Lock()
	if !s.started {
		s.mu.Unlock()
		return nil
	}
	s.started = false
	shutdown := s.shutdown
	ls := s.listeners
	s.listeners = nil
	s.mu.Unlock()

	if shutdown != nil {
		shutdown()
	}
	var errs []error
	for _, l := range ls {
		if err := l.close(); err != nil {
			errs = append(errs, err)
		}
	}
	done := make(chan struct{})
	go func() {
		s.wait.Wait()
		close(done)
	}()
	select {
	case <-done:
	case <-ctx.Done():
		errs = append(errs, ctx.Err())
	}
	s.publish()
	return errors.Join(errs...)
}

// serveUDP binds a UDP listener. One short-lived goroutine per packet keeps
// lock hold times at zero; a large socket buffer absorbs bursts instead of
// dropping them.
func (s *Server) serveUDP(ctx context.Context, addr string) error {
	pc, err := net.ListenPacket("udp", addr)
	if err != nil {
		return fmt.Errorf("bind udp %s: %w", addr, err)
	}
	if uc, ok := pc.(*net.UDPConn); ok {
		_ = uc.SetReadBuffer(4 << 20)
		_ = uc.SetWriteBuffer(4 << 20)
	}
	// The read deadline is what lets the loop notice cancellation promptly;
	// without it a blocked ReadFrom would outlive shutdown. It has to be reset
	// on every iteration: a deadline set once expires for good and turns the
	// loop into a spin that never reads another packet.
	const readTick = 500 * time.Millisecond
	s.add(listener{close: pc.Close, info: ListenerInfo{
		Transport: "udp", Addr: pc.LocalAddr().String(), Enabled: true, Running: true,
	}})
	s.log.Info("dns listener up", "transport", "udp", "addr", pc.LocalAddr().String())

	s.wait.Add(1)
	go func() {
		defer s.wait.Done()
		buf := make([]byte, dns.MaxMsgSize)
		for {
			if ctx.Err() != nil {
				return
			}
			_ = pc.SetReadDeadline(time.Now().Add(readTick))
			n, from, err := pc.ReadFrom(buf)
			if err != nil {
				if ctx.Err() != nil {
					return
				}
				if isTimeout(err) {
					continue
				}
				// A closed listener during shutdown lands here.
				if errors.Is(err, net.ErrClosed) {
					return
				}
				s.log.Debug("udp read", "error", err)
				continue
			}
			pkt := make([]byte, n)
			copy(pkt, buf[:n])
			// No ResolveUDPAddr here: the peer is already a *net.UDPAddr, and
			// re-parsing from.String() costs a string build plus a SplitHostPort
			// on the hottest path in the process. servePacket only needs the
			// address to write back to, so pass what we already have.
			addr, _ := from.(*net.UDPAddr)
			go s.servePacket(ctx, pc, pkt, from, addr)
		}
	}()
	return nil
}

func isTimeout(err error) bool {
	var ne net.Error
	return errors.As(err, &ne) && ne.Timeout()
}

// servePacket answers one UDP query. The listener is passed in so the reply
// goes out on the very same socket the query arrived on: a fresh socket per
// answer would burn an ephemeral port and a connect handshake per query.
func (s *Server) servePacket(ctx context.Context, pc net.PacketConn, pkt []byte, from net.Addr, to *net.UDPAddr) {
	m := new(dns.Msg)
	if err := m.Unpack(pkt); err != nil {
		// An unparsable message cannot be replied to: there is no ID and no
		// question to echo. Dropping it is the only correct response.
		s.bump("udp", "errors")
		return
	}
	// A message with no question is still answerable (FORMERR), so it goes
	// through the engine's validation rather than being dropped here.
	// Cap the client's advertised buffer: nobody needs 64 KiB of DNS, and an
	// unbounded value is a cheap amplification vector.
	if o := m.IsEdns0(); o != nil && o.UDPSize() > 4096 {
		o.SetUDPSize(4096)
	}
	req := s.request(ctx, m, from, "udp")
	res := s.eng.Handle(ctx, req)
	s.bump("udp", "queries")
	s.classify("udp", res)
	if res.Drop || res.Msg == nil {
		s.bump("udp", "dropped")
		return
	}
	if to == nil {
		return
	}
	out := s.fitUDP(res.Msg, m, 512)
	if out == nil {
		return
	}
	if c, ok := pc.(interface{ SetWriteDeadline(time.Time) error }); ok {
		_ = c.SetWriteDeadline(time.Now().Add(2 * time.Second))
	}
	s.bump("udp", "sent")
	_, _ = pc.WriteTo(out, to)
}

// fitUDP packs a reply into a UDP datagram, setting TC and returning a
// question-only response when it does not fit (RFC 1035 §4.2.1, RFC 2181 §9).
func (s *Server) fitUDP(res *dns.Msg, q *dns.Msg, floor int) []byte {
	limit := floor
	if o := q.IsEdns0(); o != nil {
		if n := int(o.UDPSize()); n >= 512 {
			limit = n
		}
	}
	if limit > 4096 {
		limit = 4096
	}
	out, err := res.Pack()
	if err != nil {
		return nil
	}
	if len(out) <= limit {
		return out
	}
	tc := new(dns.Msg)
	tc.SetReply(q)
	tc.Rcode = res.Rcode
	tc.Truncated = true
	tc.AuthenticatedData = res.AuthenticatedData
	if o := q.IsEdns0(); o != nil {
		tc.SetEdns0(uint16(limit), o.Do())
	}
	out, err = tc.Pack()
	if err != nil {
		return nil
	}
	s.bump("udp", "truncated")
	return out
}

// request builds an engine.Request, filling in the client's identity.
func (s *Server) request(ctx context.Context, m *dns.Msg, from net.Addr, transport string) engine.Request {
	ip := addrToAddr(from)
	key := ip.String()
	req := engine.Request{
		Msg:       m,
		Remote:    ip,
		Transport: transport,
		Start:     time.Now(),
	}
	if f := s.names.Load(); f != nil {
		req.ClientNames = (*f)(key)
	}
	if f := s.tags.Load(); f != nil {
		req.ClientTags = (*f)(key)
	}
	return req
}

// classify records per-transport outcome counters.
func (s *Server) classify(transport string, res engine.Response) {
	switch res.Action {
	case "block":
		s.bump(transport, "blocked")
	case "cached":
		s.bump(transport, "cached")
	case "error", "drop":
		s.bump(transport, "errors")
	}
}

// serveTCP binds a TCP listener for plain DNS.
func (s *Server) serveTCP(ctx context.Context, addr string) error {
	ln, err := net.Listen("tcp", addr)
	if err != nil {
		return fmt.Errorf("bind tcp %s: %w", addr, err)
	}
	return s.serveStream(ctx, ln, "tcp", nil, 0)
}

// serveStream runs the accept loop shared by TCP and DoT.
func (s *Server) serveStream(ctx context.Context, ln net.Listener, transport string, tlsCfg *tls.Config, idleTimeout time.Duration) error {
	if tlsCfg != nil {
		ln = tls.NewListener(ln, tlsCfg)
	}
	s.add(listener{close: ln.Close, info: ListenerInfo{
		Transport: transport, Addr: ln.Addr().String(), Enabled: true, Running: true,
	}})
	s.log.Info("dns listener up", "transport", transport, "addr", ln.Addr().String())
	s.wait.Add(1)
	go func() {
		defer s.wait.Done()
		var wg sync.WaitGroup
		defer wg.Wait()
		for {
			conn, err := ln.Accept()
			if err != nil {
				if ctx.Err() != nil {
					return
				}
				if isTimeout(err) {
					continue
				}
				return
			}
			wg.Add(1)
			go func() {
				defer wg.Done()
				s.serveConn(ctx, conn, transport, idleTimeout)
			}()
		}
	}()
	return nil
}

// serveConn handles one stream connection with RFC 1035 §4.2.2 length
// prefixes, so a client may pipeline several queries.
func (s *Server) serveConn(ctx context.Context, conn net.Conn, transport string, idleTimeout time.Duration) {
	defer func() { _ = conn.Close() }()
	if idleTimeout > 0 {
		_ = conn.SetDeadline(time.Now().Add(idleTimeout))
	}
	co := &dns.Conn{Conn: conn}
	for {
		if ctx.Err() != nil {
			return
		}
		if idleTimeout > 0 {
			_ = conn.SetReadDeadline(time.Now().Add(idleTimeout))
		}
		m, err := co.ReadMsg()
		if err != nil {
			return
		}
		if len(m.Question) == 0 {
			s.bump(transport, "errors")
			return
		}
		req := s.request(ctx, m, conn.RemoteAddr(), transport)
		req.TCP = true
		res := s.eng.Handle(ctx, req)
		s.bump(transport, "queries")
		s.classify(transport, res)
		if res.Drop || res.Msg == nil {
			s.bump(transport, "dropped")
			return
		}
		if idleTimeout > 0 {
			_ = conn.SetWriteDeadline(time.Now().Add(idleTimeout))
		}
		if err := co.WriteMsg(res.Msg); err != nil {
			return
		}
		s.bump(transport, "sent")
	}
}

// serveDoT binds the DNS-over-TLS listener.
func (s *Server) serveDoT(ctx context.Context, cfg *config.Config) error {
	tlsCfg, err := tlsConfig(cfg.DNS.DoT.CertFile, cfg.DNS.DoT.KeyFile, cfg.DNS.DoT.ClientCA, cfg.DNS.DoT.MinVersion)
	if err != nil {
		return fmt.Errorf("dot: %w", err)
	}
	if tlsCfg == nil {
		return errors.New("dot: enabled but no certificate configured")
	}
	tlsCfg.NextProtos = []string{"dot"}
	timeout := cfg.Server.Timeouts.TLS.D()
	var errs []error
	for _, addr := range cfg.DNS.DoT.Listen {
		ln, err := net.Listen("tcp", addr)
		if err != nil {
			errs = append(errs, fmt.Errorf("bind dot %s: %w", addr, err))
			continue
		}
		if err := s.serveStream(ctx, ln, "dot", tlsCfg, timeout); err != nil {
			errs = append(errs, err)
		}
	}
	return errors.Join(errs...)
}

// serveDoH binds the DNS-over-HTTPS listener. GET (with ?dns=) and POST
// (application/dns-message) are both supported per RFC 8484, and the handler
// speaks h2 with an h2c fallback so it also works on a plain TCP listener.
func (s *Server) serveDoH(ctx context.Context, cfg *config.Config) error {
	base := normaliseBase(cfg.DNS.DoH.BasePath)
	mux := http.NewServeMux()
	h := s.doHHandler(base, cfg.DNS.DoH.AllowPost)
	mux.Handle(base, h)
	// RFC 8484 §4.1 reserves /.well-known/dns-query as an alias.
	if base != "/dns-query" {
		mux.Handle("/.well-known/dns-query", h)
	}
	mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) {
		w.Header().Set("Content-Type", "text/plain")
		_, _ = w.Write([]byte("ok\n"))
	})

	h2s := &http2.Server{IdleTimeout: 90 * time.Second}
	handler := h2c.NewHandler(mux, h2s)

	tlsCfg, tlsErr := tlsConfig(cfg.DNS.DoH.CertFile, cfg.DNS.DoH.KeyFile, "", cfg.DNS.DoH.MinVersion)
	if tlsCfg != nil {
		tlsCfg.NextProtos = []string{"h2", "http/1.1"}
	}

	var errs []error
	for _, addr := range cfg.DNS.DoH.Listen {
		ln, err := net.Listen("tcp", addr)
		if err != nil {
			errs = append(errs, fmt.Errorf("bind doh %s: %w", addr, err))
			continue
		}
		actual := ln.Addr().String()
		srv := &http.Server{
			Handler:           handler,
			ReadHeaderTimeout: 10 * time.Second,
			IdleTimeout:       90 * time.Second,
			// No WriteTimeout: a long-lived h2 stream should not be cut off,
			// and ReadHeaderTimeout plus IdleTimeout already bound the damage.
		}
		closeFn := ln.Close
		if tlsCfg != nil {
			srv.TLSConfig = tlsCfg
			ln = tls.NewListener(ln, tlsCfg)
			closeFn = ln.Close
		}
		s.add(listener{close: closeFn, info: ListenerInfo{
			Transport: "doh", Addr: actual, Enabled: true, Running: true,
		}})
		if tlsErr != nil {
			s.log.Warn("doh: running without TLS", "error", tlsErr)
		}
		s.log.Info("dns listener up", "transport", "doh", "addr", actual, "path", base)
		s.wait.Add(1)
		go func() {
			defer s.wait.Done()
			if err := srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) {
				s.log.Debug("doh server stopped", "error", err)
			}
		}()
		s.wait.Add(1)
		go func() {
			defer s.wait.Done()
			<-ctx.Done()
			shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			defer cancel()
			_ = srv.Shutdown(shutCtx)
		}()
	}
	return errors.Join(errs...)
}

const contentTypeDNS = "application/dns-message"

// maxGETResponse bounds a DoH GET response so no intermediary truncates it.
const maxGETResponse = 65535

// doHHandler builds the RFC 8484 handler.
func (s *Server) doHHandler(base string, allowPost bool) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.Method != http.MethodGet && r.Method != http.MethodPost {
			w.Header().Set("Allow", "GET, POST")
			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
			return
		}
		var (
			payload []byte
			err     error
		)
		switch r.Method {
		case http.MethodGet:
			// RFC 8484 §4.1 defines exactly one parameter, "dns", holding
			// unpadded base64url. An unknown parameter is a 400 rather than a
			// 404 so a client can tell its own mistake from a wrong path.
			q := r.URL.Query()
			if !q.Has("dns") {
				writeDoHError(w, http.StatusBadRequest, "missing dns parameter")
				return
			}
			if len(q) > 1 {
				for k := range q {
					if k != "dns" {
						writeDoHError(w, http.StatusBadRequest, "unsupported parameter "+k)
						return
					}
				}
			}
			payload, err = decodeBase64URL(q.Get("dns"))
			if err != nil {
				writeDoHError(w, http.StatusBadRequest, "malformed dns parameter")
				return
			}
		case http.MethodPost:
			if !allowPost {
				w.Header().Set("Allow", "GET")
				http.Error(w, "POST not allowed", http.StatusMethodNotAllowed)
				return
			}
			if ct := r.Header.Get("Content-Type"); ct != contentTypeDNS {
				writeDoHError(w, http.StatusUnsupportedMediaType, "content-type must be "+contentTypeDNS)
				return
			}
			payload, err = readCapped(r, dns.MaxMsgSize)
			if err != nil {
				writeDoHError(w, http.StatusBadRequest, "cannot read body")
				return
			}
		}

		m := new(dns.Msg)
		if err := m.Unpack(payload); err != nil {
			writeDoHError(w, http.StatusBadRequest, "malformed DNS message")
			return
		}
		if len(m.Question) == 0 {
			writeDoHError(w, http.StatusBadRequest, "no question")
			return
		}
		req := s.request(r.Context(), m, httpRemoteAddr(r), "doh")
		req.TCP = true
		req.DoHPath = r.URL.Path
		res := s.eng.Handle(r.Context(), req)
		s.bump("doh", "queries")
		s.classify("doh", res)
		if res.Drop || res.Msg == nil {
			// A dropped query still needs a body: unlike UDP, this is a
			// well-formed HTTP exchange, so hanging up would read as a
			// transport error rather than a policy decision.
			res.Msg = new(dns.Msg)
			res.Msg.SetRcode(m, dns.RcodeSuccess)
		}
		wire, err := res.Msg.Pack()
		if err != nil {
			http.Error(w, "pack failed", http.StatusInternalServerError)
			return
		}
		w.Header().Set("Content-Type", contentTypeDNS)
		w.Header().Set("Cache-Control", "max-age=0")
		if !cfgHideIdentity(s) {
			w.Header().Set("Server", "fancy-dns/"+s.version)
		}
		w.WriteHeader(http.StatusOK)
		_, _ = w.Write(wire)
	})
}

func cfgHideIdentity(s *Server) bool {
	if s.cfg == nil {
		return false
	}
	return s.cfg.Get().DNS.HideIdentity
}

func writeDoHError(w http.ResponseWriter, code int, msg string) {
	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
	w.Header().Set("X-Content-Type-Options", "nosniff")
	http.Error(w, msg, code)
}

// tlsConfig loads a certificate for the encrypted listeners. It returns
// (nil, nil) when no certificate is configured, which means "run plaintext".
func tlsConfig(certFile, keyFile, clientCA, minVersion string) (*tls.Config, error) {
	if certFile == "" || keyFile == "" {
		return nil, nil
	}
	cert, err := tls.LoadX509KeyPair(certFile, keyFile)
	if err != nil {
		return nil, fmt.Errorf("load certificate: %w", err)
	}
	cfg := &tls.Config{
		Certificates: []tls.Certificate{cert},
		MinVersion:   parseTLSVersion(minVersion),
	}
	if clientCA != "" {
		pem, err := os.ReadFile(clientCA)
		if err != nil {
			return nil, fmt.Errorf("read client CA: %w", err)
		}
		pool := x509.NewCertPool()
		if !pool.AppendCertsFromPEM(pem) {
			return nil, errors.New("client CA: no certificates found")
		}
		cfg.ClientCAs = pool
		cfg.ClientAuth = tls.RequireAndVerifyClientCert
	}
	return cfg, nil
}

func parseTLSVersion(s string) uint16 {
	switch s {
	case "1.3":
		return tls.VersionTLS13
	case "1.0":
		return tls.VersionTLS10
	default:
		return tls.VersionTLS12
	}
}

func normaliseBase(p string) string {
	if p == "" {
		return "/dns-query"
	}
	if !strings.HasPrefix(p, "/") {
		p = "/" + p
	}
	for len(p) > 1 && p[len(p)-1] == '/' {
		p = p[:len(p)-1]
	}
	return p
}

func readCapped(r *http.Request, max int64) ([]byte, error) {
	if r.ContentLength > max {
		return nil, fmt.Errorf("body too large: %d", r.ContentLength)
	}
	// LimitReader plus one extra byte is how you detect "too large" without
	// trusting Content-Length, which a client may lie about.
	body := io.LimitReader(r.Body, max+1)
	buf, err := io.ReadAll(body)
	if err != nil {
		return nil, err
	}
	if int64(len(buf)) > max {
		return nil, errors.New("body too large")
	}
	return buf, nil
}

func decodeBase64URL(s string) ([]byte, error) {
	// RFC 8484 requires unpadded base64url, but every major client sends
	// padding, so accept both rather than fail a legitimate request.
	if b, err := base64.RawURLEncoding.DecodeString(s); err == nil {
		return b, nil
	}
	return base64.URLEncoding.DecodeString(s)
}

// httpRemoteAddr turns an http.Request's RemoteAddr into a net.Addr so the
// shared request builder can be used. Over plain HTTP this is the socket peer;
// behind a reverse proxy it is the proxy, which is why the caller must trust
// X-Forwarded-For before that matters (handled in the API layer).
func httpRemoteAddr(r *http.Request) net.Addr {
	if r.RemoteAddr == "" {
		return &net.TCPAddr{}
	}
	ap, err := netip.ParseAddrPort(r.RemoteAddr)
	if err != nil {
		return &net.TCPAddr{}
	}
	return net.UDPAddrFromAddrPort(ap)
}

// addrToAddr extracts a netip.Addr from a net.Addr.
func addrToAddr(a net.Addr) netip.Addr {
	switch v := a.(type) {
	case *net.UDPAddr:
		return v.AddrPort().Addr().Unmap()
	case *net.TCPAddr:
		return v.AddrPort().Addr().Unmap()
	default:
		ap, err := netip.ParseAddrPort(a.String())
		if err == nil {
			return ap.Addr().Unmap()
		}
		if ip, err := netip.ParseAddr(a.String()); err == nil {
			return ip.Unmap()
		}
		return netip.Addr{}
	}
}

// Stats aggregates the per-transport counters plus the engine's own view.
func (s *Server) Stats() HandlerStats {
	var out HandlerStats
	s.counters.Range(func(k, v any) bool {
		key, _ := k.(string)
		n := v.(*atomic.Uint64).Load()
		dot := strings.IndexByte(key, '.')
		if dot < 0 {
			return true
		}
		switch key[dot+1:] {
		case "queries":
			out.Queries += n
		case "blocked":
			out.Blocked += n
		case "cached":
			out.Cached += n
		case "errors":
			out.Errors += n
		case "dropped":
			out.Dropped += n
		}
		return true
	})
	if s.eng != nil {
		out.Inflight = s.eng.Inflight()
	}
	return out
}

func (s *Server) bump(transport, field string) {
	if transport == "" {
		transport = "udp"
	}
	key := transport + "." + field
	v, _ := s.counters.LoadOrStore(key, new(atomic.Uint64))
	v.(*atomic.Uint64).Add(1)
}

// flusher is implemented by the engine: it pushes buffered cache and log state
// to disk on demand (used by the admin API's "flush" action).
type flusher interface{ Flush() }

// Flush pushes buffered state to disk.
func (s *Server) Flush() {
	if f, ok := s.eng.(flusher); ok {
		f.Flush()
	}
}