DNS 同时支持 UDP 和 TCP。服务器因可用消息空间而截断响应时，会设置 TC 标志。客户端收到该标志，需要处理结果不完整的情况，通常通过 TCP 重新查询。这涉及当前使用的 DNS 响应限制，并不直接证明物理网络无法承载任何更大的 UDP 数据报。排障时应先确认协议层发生了什么，再讨论传输路径的实际容量。

坏掉的样子

允许 UDP 五十三端口却阻止 TCP 五十三端口的网络，可能小查询正常，大回答失败。DNSSEC 数据和较大记录集容易暴露这种差异，但 TCP 并不只用于这些特殊情况。通用 DNS 实现需要支持 TCP，客户端也可以直接选择它。因此，一次小型 UDP 查询成功，不能证明整个 DNS 服务的传输路径完整，尤其不能代替对 TCP 监听和访问规则的检查。

被截断的回答仍可能包含部分记录。如果忽略 TC，把这些记录当成完整集合，就可能漏掉有效目的地或必需数据。NOERROR 不会抵消截断标志。当应用看到的记录比另一种传输方式下的权威查询少时，应保留完整问题、响应标志和记录数量，不要只提取一个地址或文本值。仅凭有内容返回，并不能判断这次 DNS 交换已经完整结束。

TCP 也没有取消所有 DNS 消息大小限制。TCP 流中，每个 DNS 消息之前都有两字节长度字段，因此单个消息最多六万五千五百三十五字节；区域传送可以由多个消息组成。流读取可能拆开长度字段或正文，程序应循环读取所需字节数，不能假定一次 recv 就返回整个回答。传输可靠与应用消息边界明确，是两项不同责任。

快速验证

将脚本保存为 check-dns-tcp.py，准备 Docker 后执行 python3 check-dns-tcp.py。固定 BIND 9.18.49 环境在隔离容器中提供包含四条记录的 TXT 集合。第一次查询明确关闭 EDNS，以观察传统五百一十二字节 UDP 限制。之后程序使用真实 TCP 连接，并建立一个有意只提供 UDP 的中继端点；不修改现有防火墙或公共解析器：

from pathlib import Path
from tempfile import TemporaryDirectory
import json,subprocess,uuid
recipe='FROM python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84\nRUN apt-get update && apt-get install -y --no-install-recommends bind9=1:9.18.49-1~deb12u2 bind9-dnsutils=1:9.18.49-1~deb12u2 dnsmasq-base=2.90-4~deb12u2 python3-dnspython=2.3.0-1 && rm -rf /var/lib/apt/lists/*\nCMD ["/usr/bin/python3"]\n'
program=r"""
from contextlib import ExitStack,contextmanager
from pathlib import Path
from tempfile import TemporaryDirectory
import json,re,signal,subprocess,time
import dns.exception,dns.message,dns.query,dns.rcode,dns.rdatatype

def run(*args,check=True):
    result=subprocess.run(args,capture_output=True,text=True,timeout=15)
    if check:assert result.returncode==0,(args,result.returncode,result.stdout,result.stderr)
    return result
def query(name,kind='A',port=1053):
    return dns.query.udp(dns.message.make_query(name,kind),'127.0.0.1',port=port,timeout=1)
def values(response,kind='A'):
    wanted=dns.rdatatype.from_text(kind)
    return [item.to_text() for rrset in response.answer if rrset.rdtype==wanted for item in rrset]
def show(response):
    def section(records):return [{'name':str(rr.name),'type':dns.rdatatype.to_text(rr.rdtype),'ttl':rr.ttl,'data':[item.to_text() for item in rr]} for rr in records]
    return {'rcode':dns.rcode.to_text(response.rcode()),'answer':section(response.answer),'authority':section(response.authority)}
def wait(predicate,timeout=15):
    deadline=time.monotonic()+timeout
    while time.monotonic()<deadline:
        try:
            if predicate():return
        except (OSError,dns.exception.Timeout):pass
        time.sleep(.05)
    raise AssertionError('DNS fixture condition did not become true')
def stop(process):
    if process.poll() is None:
        process.terminate()
        try:process.wait(timeout=5)
        except subprocess.TimeoutExpired:process.kill();process.wait(timeout=5)

@contextmanager
def lab(records,soa_ttl=8,minimum=4,extra=''):
    with TemporaryDirectory(prefix='xbcatm-dns-') as directory,ExitStack() as stack:
        root=Path(directory);zone=root/'zone';conf=root/'named.conf';log=root/'named.log';serial=1
        def save(body):
            zone.write_text(f'$ORIGIN lab.test.\n$TTL 30\n@ {soa_ttl} IN SOA ns.lab.test. hostmaster.lab.test. ({serial} 60 60 3600 {minimum})\n@ IN NS ns.lab.test.\nns IN A 127.0.0.1\n'+body+'\n')
            run('named-checkzone','lab.test',str(zone))
        save(records)
        conf.write_text(f'''options {{ directory "{root}"; listen-on port 1053 {{127.0.0.1;}}; listen-on-v6 {{none;}};
recursion no; dnssec-validation no; notify no; querylog yes; max-cache-size 8m; pid-file "{root}/named.pid"; session-keyfile none;
{extra} }};
controls {{}};
zone "lab.test" {{ type primary; file "{zone}"; }};
''')
        run('named-checkconf',str(conf))
        output=stack.enter_context(log.open('w'))
        authority=subprocess.Popen(['named','-g','-c',str(conf),'-n','1','-U','1'],stdout=output,stderr=subprocess.STDOUT)
        stack.callback(stop,authority)
        def ready():
            assert authority.poll() is None,log.read_text()
            response=query('lab.test.','SOA');return response.rcode()==0 and response.answer[0][0].serial==serial
        wait(ready)
        def update(body):
            nonlocal serial
            serial+=1;save(body);authority.send_signal(signal.SIGHUP);wait(ready)
        def cache(port=1054,upstream=1053,size=150,options=()):
            logfile=root/('cache-'+str(port)+'.log');stream=stack.enter_context(logfile.open('w'))
            command=['dnsmasq','--no-daemon','--conf-file=/dev/null','--no-resolv','--no-hosts','--bind-interfaces',
                '--listen-address=127.0.0.1','--port='+str(port),'--server=127.0.0.1#'+str(upstream),
                '--cache-size='+str(size),'--log-queries','--log-facility=-',*options]
            process=subprocess.Popen(command,stdout=stream,stderr=subprocess.STDOUT);stack.callback(stop,process)
            def cache_ready():
                assert process.poll() is None,logfile.read_text()
                return query('lab.test.','SOA',port).rcode()==0
            wait(cache_ready);return process,logfile
        def queries(name,kind='A'):
            return sum(bool(re.search(r'query: '+re.escape(name.rstrip('.'))+r' IN '+kind+r'\b',line)) for line in log.read_text().splitlines())
        yield root,update,cache,queries
import dns.flags,socket,struct,threading
texts=[f'{i}-'+('x'*200) for i in range(4)]
records='small 30 IN A 127.0.0.2\n'+'\n'.join('large 30 IN TXT "'+s+'"' for s in texts)
def exact(stream,size):
    data=b''
    while len(data)<size:
        part=stream.recv(size-len(data));assert part,'unexpected TCP EOF';data+=part
    return data
def description(response):return {'wire_bytes':len(response.to_wire()),'TC':bool(response.flags&dns.flags.TC),'answer_records':sum(len(rr) for rr in response.answer)}
with lab(records) as (_,update,cache,queries):
    q=dns.message.make_query('large.lab.test.','TXT',use_edns=False)
    udp=dns.query.udp(q,'127.0.0.1',port=1053,timeout=2)
    assert udp.flags&dns.flags.TC and len(udp.to_wire())<=512
    tcp=dns.query.tcp(q,'127.0.0.1',port=1053,timeout=2)
    assert not tcp.flags&dns.flags.TC and set(values(tcp,'TXT'))=={'"'+s+'"' for s in texts}
    assert len(tcp.to_wire())>512
    with socket.create_connection(('127.0.0.1',1053),timeout=2) as stream:
        lengths=[]
        for name,kind in [('large.lab.test.','TXT'),('small.lab.test.','A')]:
            request=dns.message.make_query(name,kind);wire=request.to_wire();stream.sendall(struct.pack('!H',len(wire))+wire)
            size=struct.unpack('!H',exact(stream,2))[0];reply=dns.message.from_wire(exact(stream,size));assert reply.id==request.id and reply.rcode()==0
            assert values(reply,kind);lengths.append(size)
    with socket.socket() as reservation,socket.socket(socket.AF_INET,socket.SOCK_DGRAM) as relay:
        reservation.bind(('127.0.0.1',0));port=reservation.getsockname()[1];relay.bind(('127.0.0.1',port));relay.settimeout(3)
        errors=[]
        def one_datagram():
            try:
                data,peer=relay.recvfrom(4096);answer=dns.query.udp(dns.message.from_wire(data),'127.0.0.1',port=1053,timeout=2);relay.sendto(answer.to_wire(),peer)
            except Exception as error:errors.append(type(error).__name__)
        thread=threading.Thread(target=one_datagram);thread.start()
        limited=dns.query.udp(q,'127.0.0.1',port=port,timeout=2);assert limited.flags&dns.flags.TC
        thread.join(timeout=4);assert not thread.is_alive() and not errors
        try:dns.query.tcp(q,'127.0.0.1',port=port,timeout=1);raise AssertionError('UDP-only endpoint unexpectedly accepted TCP')
        except OSError as error:failure=type(error).__name__
    print(json.dumps({'UDP':description(udp),'TCP':description(tcp),'two_messages_one_connection_lengths':lengths,
        'UDP_only_path':{'TC_received':True,'TCP_error':failure}},sort_keys=True))
"""
name='xbcatm-dns-'+uuid.uuid4().hex[:10]
def docker(*args,**kwargs):return subprocess.run(['docker',*args],capture_output=True,text=True,timeout=300,**kwargs)
try:
    with TemporaryDirectory(prefix=name+'-') as directory:
        root=Path(directory);(root/'Dockerfile').write_text(recipe)
        built=docker('build','-t',name,str(root));assert built.returncode==0,built.stderr
        result=docker('run','--rm','-i','--name',name,'--network','none','--memory','384m','--pids-limit','128',name,
            '/usr/bin/python3','-',input=program)
        assert result.returncode==0,result.stderr
        print(json.dumps(json.loads(result.stdout),sort_keys=True))
finally:
    docker('rm','-f',name);docker('image','rm',name)

实测 UDP 响应四百六十二字节，设置 TC，只带回四条 TXT 中的两条。TCP 响应八百九十二字节，TC 清除，并包含全部四个预期值。脚本比较完整记录内容，不是看到字节数变大就认定缺失数据已经恢复。这一结果还直接表明：截断响应可能保留部分有用数据，但调用方不能把部分集合当成权威完整结果使用。

接着程序只打开一个 TCP 连接，先查询较大的 TXT，再查询一个小 A 回答。每次都读取两字节长度和对应正文，记录中的消息长度分别为八百九十二与四十八，并检查响应标识。两次交换使用同一条流，证明每个查询不必重新握手。长度字段不计入所描述的 DNS 消息正文大小，也不会因为 TCP 分段方式改变而失去作用。

另一个端点只转发一次 UDP 问题到真实权威，并返回截断回答。在相同端点编号上，保留一个已绑定但未监听的 TCP 套接字，使那里确实没有 TCP 服务。UDP 查询带 TC 成功返回，TCP 尝试则得到 ConnectionRefusedError。这个对照复现了传输方式不匹配，不依赖无关公网服务是否偶然失败，也不需要改变共享网络策略。

分开诊断截断与连接失败

检查响应标志、存在时的 EDNS 宣告容量、TCP 建连结果及最终 DNS 内容。先收到 TC 再连接被拒绝，与 UDP 完全无响应或权威返回 NXDOMAIN，原因不同。中继有意只支持一次交换，是受控失败实验，不是可部署的通用 DNS 代理。读者应提取它展示的传输边界，而不是把实验代码的范围误认为生产服务的完整能力。

让客户端、解析器、服务器监听及中间访问规则共同允许所需 TCP 路径，并根据负载选择连接截止时间和资源限制。复用可以减少重复建立连接的工作，但空闲连接、连接上限和服务器行为仍然重要。不能把一次新连接多出的握手，转换成每一次 TCP 查询都必须支付的固定延迟。实际成本应结合复用策略和真实请求序列测量。

EDNS 能允许更大的 UDP 回答，在不少场景中避免截断，却不会消除 TCP 需求。更大数据报可能先遭遇路径 MTU 或分片问题，客户端甚至收不到有效 TC 提示。应同时验证配置的 UDP 大小和完整 TCP 交换，不要把特别大的宣告容量当成第二条传输路径已经可用。协议宣告与沿途能否交付数据，是两个检查维度。

最后，TCP 提供可靠字节流，不直接提供 DNS 保密性或记录数据认证。普通 TCP DNS 仍是明文，加密 DNS 传输与 DNSSEC 分别处理另外的安全属性。本实验的证据限定在截断处理、完整记录恢复、流消息边界、连接复用，以及只有 UDP 时的具体失败。描述结果时保留这些边界，才能避免把一种传输方式的成功扩大成全部 DNS 能力已验证。

延伸阅读

来源