diff --git a/server/coverage_boost_test.go b/server/coverage_boost_test.go index f291a5c..51f8fdc 100644 --- a/server/coverage_boost_test.go +++ b/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() diff --git a/xfr/client.go b/xfr/client.go index 9affdee..ca3c63c 100644 --- a/xfr/client.go +++ b/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. diff --git a/xfr/transfer_cancel_test.go b/xfr/transfer_cancel_test.go new file mode 100644 index 0000000..6374ab2 --- /dev/null +++ b/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") + } + }) +}