plugin/forward: Fix incorrect retry of local DNS message serialization failures (#8313)

This PR fixes the forward plugin incorrectly retrying deterministic
local DNS message serialization failures as if they were upstream transport errors.

Signed-off-by: Yong Tang <yong.tang.github@outlook.com>

plugin/forward/forward.go

@@ -206,6 +206,10 @@ func (f *Forward) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg
 		upstreamErr = err
 
 		if err != nil {
+			if errors.Is(err, proxyPkg.ErrInvalidRequest) {
+				return dns.RcodeFormatError, err
+			}
+
 			// Kick off health check to see if *our* upstream is broken.
 			if f.maxfails != 0 {
 				proxy.Healthcheck()

plugin/forward/forward_test.go

@@ -304,3 +304,46 @@ func TestForwardFailoverStopsAfterAllUpstreams(t *testing.T) {
 		t.Errorf("Expected second upstream to be queried once, got %d", got)
 	}
 }
+
+func TestForwardDoesNotRetryLocalPackError(t *testing.T) {
+	req := new(dns.Msg)
+	req.SetQuestion("example.org.", dns.TypeA)
+	req.Answer = []dns.RR{&dns.TXT{
+		Hdr: dns.RR_Header{
+			Name:   "example.org.",
+			Rrtype: dns.TypeTXT,
+			Class:  dns.ClassINET,
+		},
+		// A TXT character-string is limited to 255 wire bytes. This is a
+		// deterministic local serialization error and is unrelated to HIP or
+		// compression-pointer handling in miekg/dns.
+		Txt: []string{strings.Repeat("x", 256)},
+	}}
+
+	f := New()
+	f.maxfails = 0
+	f.maxConnectAttempts = 2
+	f.opts.ForceTCP = true
+	f.proxies = []*proxy.Proxy{
+		proxy.NewProxy("forward", "127.0.0.1:1", transport.DNS),
+	}
+
+	tracer := mocktracer.New()
+	span := tracer.StartSpan("test")
+	ctx := opentracing.ContextWithSpan(context.Background(), span)
+
+	rcode, err := f.ServeDNS(ctx, &mockResponseWriter{}, req)
+	if rcode != dns.RcodeFormatError {
+		t.Fatalf("expected FORMERR, got %s", dns.RcodeToString[rcode])
+	}
+	if err == nil || !strings.Contains(err.Error(), "string exceeded 255 bytes in txt") {
+		t.Fatalf("expected local TXT packing error, got %v", err)
+	}
+
+	// ServeDNS starts one child span for each forwarding attempt. A local
+	// packing failure must stop after the first attempt even though the normal
+	// connect-attempt limit permits two attempts.
+	if got := len(tracer.FinishedSpans()); got != 1 {
+		t.Fatalf("expected one forwarding attempt, got %d", got)
+	}
+}

plugin/pkg/proxy/connect.go

@@ -26,6 +26,8 @@ const (
 	ErrTransportStopped = "proxy: transport stopped"
 )
 
+var ErrInvalidRequest = errors.New("proxy: invalid request")
+
 // limitTimeout is a utility function to auto-tune timeout values
 // average observed time is moved towards the last observed delay moderated by a weight
 // next timeout to use will be the double of the computed average, limited by min and max frame.
@@ -127,6 +129,21 @@ func (p *Proxy) lookupDNS(_ctx context.Context, state request.Request, opts Opti
 		proto = state.Proto()
 	}
 
+	originId := state.Req.Id
+	state.Req.Id = dns.Id()
+	defer func() {
+		state.Req.Id = originId
+	}()
+
+	var wire []byte
+	if state.Req.IsTsig() == nil {
+		var err error
+		wire, err = state.Req.Pack()
+		if err != nil {
+			return nil, nil, proto, fmt.Errorf("%w: %w", ErrInvalidRequest, err)
+		}
+	}
+
 	pc, cached, err := p.transport.Dial(proto)
 	if err != nil {
 		return nil, nil, proto, err
@@ -151,14 +168,12 @@ func (p *Proxy) lookupDNS(_ctx context.Context, state request.Request, opts Opti
 	pc.c.UDPSize = max(uint16(state.Size()), 512) // #nosec G115 -- UDP size fits in uint16
 
 	pc.c.SetWriteDeadline(time.Now().Add(maxTimeout))
-	// records the origin Id before upstream.
-	originId := state.Req.Id
-	state.Req.Id = dns.Id()
-	defer func() {
-		state.Req.Id = originId
-	}()
-
-	if err := pc.c.WriteMsg(state.Req); err != nil {
+	if wire != nil {
+		_, err = pc.c.Write(wire)
+	} else {
+		err = pc.c.WriteMsg(state.Req)
+	}
+	if err != nil {
 		pc.c.Close() // not giving it back
 		if err == io.EOF && cached {
 			return nil, localAddr, proto, ErrCachedClosed