fix(xfr): honour context cancellation during an in-flight transfer (F53)

server/coverage_boost_test.go

@@ -1333,7 +1333,7 @@ func TestDoTHandleWriteResponseError(t *testing.T) {
 	}
 
 	// Now send a query but close the connection before reading the response
-	query2 := buildMinimalQuery(0xF002)
+	query2 := buildMinimalQuery(0xF53)
 	binary.Write(conn, binary.BigEndian, uint16(len(query2)))
 	conn.Write(query2)
 	conn.Close()

xfr/client.go

@@ -189,7 +189,18 @@ func transfer(ctx context.Context, cfg ClientConfig, qtype uint16, serial uint32
 	}
 	defer conn.Close()
 
-	return runTransfer(conn, cfg, qtype, serial)
+	// F53: the per-message deadlines do not observe ctx, so a primary that
+	// stalls after the query would pin the caller (and a Manager shutdown or a
+	// withdrawn catalog member) for the full timeout per message. Closing the
+	// connection on cancellation unblocks the pending read immediately.
+	stop := context.AfterFunc(ctx, func() { conn.Close() })
+	defer stop()
+
+	res, err := runTransfer(conn, cfg, qtype, serial)
+	if err != nil && ctx.Err() != nil {
+		return nil, fmt.Errorf("xfr: transfer from %s cancelled: %w", addr, ctx.Err())
+	}
+	return res, err
 }
 
 // dial connects to the primary, over TLS when configured.

xfr/transfer_cancel_test.go

@@ -0,0 +1,74 @@
+package xfr
+
+import (
+	"context"
+	"errors"
+	"net"
+	"testing"
+	"time"
+)
+
+// stallingPrimary accepts one connection, reads the query and never answers.
+// `queried` closes once the query has been consumed, i.e. the client is in its
+// read loop. release unblocks and tears the server down.
+func stallingPrimary(t *testing.T) (addr string, queried <-chan struct{}, release func()) {
+	t.Helper()
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	gate := make(chan struct{})
+	q := make(chan struct{})
+	go func() {
+		c, err := ln.Accept()
+		if err != nil {
+			return
+		}
+		buf := make([]byte, 512)
+		_, _ = c.Read(buf)
+		close(q)
+		<-gate
+		c.Close()
+	}()
+	return ln.Addr().String(), q, func() { close(gate); ln.Close() }
+}
+
+// TestTransfer_ContextCancelUnblocksStalledPrimary pins that cancelling ctx
+// ends an in-flight transfer even when the per-message timeout is very long.
+func TestTransfer_ContextCancelUnblocksStalledPrimary(t *testing.T) {
+	run := map[string]func(context.Context, ClientConfig) error{
+		"AXFR": func(ctx context.Context, cfg ClientConfig) error { _, err := AXFR(ctx, cfg); return err },
+		"IXFR": func(ctx context.Context, cfg ClientConfig) error { _, err := IXFR(ctx, cfg, 1); return err },
+	}
+	for name, fn := range run {
+		t.Run(name, func(t *testing.T) {
+			addr, queried, release := stallingPrimary(t)
+			defer release()
+
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			done := make(chan error, 1)
+			go func() {
+				done <- fn(ctx, ClientConfig{PrimaryAddr: addr, Zone: "example.test", Timeout: time.Hour})
+			}()
+			<-queried
+			cancel()
+			select {
+			case err := <-done:
+				if !errors.Is(err, context.Canceled) {
+					t.Fatalf("want context.Canceled, got %v", err)
+				}
+			case <-time.After(5 * time.Second):
+				t.Fatal("transfer still blocked after ctx cancel")
+			}
+		})
+	}
+
+	t.Run("already cancelled ctx fails fast", func(t *testing.T) {
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+		if _, err := AXFR(ctx, ClientConfig{PrimaryAddr: "127.0.0.1:1", Zone: "example.test", Timeout: time.Hour}); err == nil {
+			t.Fatal("expected error")
+		}
+	})
+}