TCP 最大段大小选项 MSS，按照其定义的开销规则，通告接收端可接受的 TCP 数据大小。两个方向分别通告，数值不必相同。路由器转发 SYN 消息时可以减小这些值，使后续数据段适应受限路径。这改变的是相关连接的 TCP 行为，不会修改底层链路 MTU。把选项调整与网络本身修复区分开，才能准确说明它解决了什么。

IPv4 包预算从 1280 字节减去各 20 字节的 IP 与 TCP 基础头，得到通告 MSS 1240。实际选项还需额外留空间；原文报告有效值为 1228。独立的 SYN 与 SYN-ACK 箭头表明，各通告约束发往通告端的数据，且仅新连接获取调整。原文结果区分了恢复的 65536 字节 TCP 交付，与仍失败的 1400 字节 DF echo 载荷。本轮未重跑实验。
限制只改变新连接的 TCP 通告，不改变链路 MTU 或缺失的 ICMP 反馈。应保留对端较小上限，并分别验证两个方向。

坏掉的样子

中间瓶颈可能比两个端点接口都小。如果经典路径 MTU 反馈被过滤，连接可以成功建立，却在较大数据上停滞。降低 MSS 能帮助该 TCP 流量避开超大报文，但缺失的反馈仍然缺失。若把这个变通措施当作完整网络修复，其他协议与流量模式仍可能暴露原有问题，健康检查也可能继续遗漏受影响场景。

MSS 不是完整 IP 包大小。IPv4 路径 MTU 为 1280、普通 IPv4 与 TCP 基础头各占 20 字节时，基本计算得到通告 MSS 1240。形成真实数据段时，还要为实际 IP 或 TCP 选项另外留空间。把通告值直接等同于最终线路报文长度，可能使看似保守的设置仍然超限。应明确每个数字包含哪些头部，而不是只套一个固定减法。

方向和时机也很重要。只修改一个 SYN 方向，另一端通告仍保持原样；更新规则，也不会改写已有连接早先交换过的选项。真正的限制应降低过大通告，而不是抬高对端原本更小的上限。策略应针对真实瓶颈和相关流量，不能把一个硬编码值视为所有网络都合适，也不能假定旧连接会自动采用新配置。

快速验证

将脚本保存为 check-mss-clamp.py，准备 Docker 后运行 python3 check-mss-clamp.py。一次性环境使用 Python 3.11.16、iproute2 6.1.0、nftables 1.0.6 和 tcpdump 4.99.3，在 Linux 6.12.76 上验证。NET_ADMIN 与 SYS_ADMIN 支持四个嵌套命名空间。端点链路 MTU 为 1500，两台路由器之间的链路为 1280：

以下环境与数值结果来自原有文章报告，本次没有独立重现。容器镜像固定并不固定宿主内核，也不保证所列软件包版本仍然可取得。

启动读取器保留无缓冲字节中的诊断行与监听行，先消费已保留的完整行，再检查就绪状态。合成测试不验证真实管道调度、进程清理或抓包行为。

from pathlib import Path
from tempfile import TemporaryDirectory
import subprocess,json,uuid
recipe='FROM python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84\nRUN apt-get update && apt-get install -y --no-install-recommends iproute2=6.1.0-3 iputils-ping=3:20221126-1+deb12u1 nftables=1.0.6-2+deb12u2 tcpdump=4.99.3-1 traceroute=1:2.1.2-1 procps=2:4.0.2-3 && rm -rf /var/lib/apt/lists/*\nCMD ["python3"]\n'
program=r"""
from contextlib import ExitStack,contextmanager
from pathlib import Path
from tempfile import TemporaryDirectory
import json,os,re,select,signal,subprocess,time
def run(*args,check=True,input=None,timeout=8):
    result=subprocess.run(args,input=input,capture_output=True,text=True,timeout=timeout)
    if check:assert result.returncode==0,(args,result.returncode,result.stdout,result.stderr)
    return result
def ns(name,*args,**kwargs):return run('ip','netns','exec',name,*args,**kwargs)
def ip(name,*args,**kwargs):return ns(name,'ip',*args,**kwargs)
def stop(process):
    if process.poll() is None:
        process.terminate()
        try:process.wait(timeout=2)
        except subprocess.TimeoutExpired:process.kill();process.wait(timeout=2)
def spawn(stack,command):
    process=subprocess.Popen(command,stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True)
    stack.callback(stop,process);return process
def line(process,timeout=3):
    assert select.select([process.stdout],[],[],timeout)[0],'child response deadline exceeded'
    value=process.stdout.readline();assert value,process.stderr.read();return value
@contextmanager
def lab():
    with TemporaryDirectory(prefix='mtu-path-') as directory,ExitStack() as stack:
        root=Path(directory);proc=root/'proc';proc.mkdir();run('mount','-t','proc','proc',str(proc));stack.callback(run,'umount',str(proc))
        for name in ['client','r1','r2','target']:
            run('ip','netns','add',name);stack.callback(run,'ip','netns','delete',name,check=False);ip(name,'link','set','lo','up')
        def setting(name,key,value):
            assert name in ['client','r1','r2','target']
            path=proc/'sys'/key.replace('.','/')
            ns(name,'python3','-c',f'from pathlib import Path; p=Path({str(path)!r});p.write_text({str(value)!r});assert p.read_text().strip()=={str(value)!r}')
        yield root,stack,setting
def topology(setting,ipv6=False):
    links=[('client','c0','r1','a0',1,2,1,1500),('r1','a1','r2','b0',2,1,2,1280),('r2','b1','target','t0',3,1,2,1500)]
    for left,left_if,right,right_if,network,left_host,right_host,mtu in links:
        run('ip','link','add',left_if,'type','veth','peer','name',right_if)
        for name,interface,host in [(left,left_if,left_host),(right,right_if,right_host)]:
            run('ip','link','set',interface,'netns',name);ip(name,'link','set',interface,'mtu',str(mtu))
            address=f'2001:db8:{network}::{host}/64' if ipv6 else f'10.210.{network}.{host}/30'
            ip(name,'addr','add',address,'dev',interface,*(['nodad'] if ipv6 else []));ip(name,'link','set',interface,'up')
    family='-6' if ipv6 else '-4'
    for name in ['r1','r2']:setting(name,'net.ipv6.conf.all.forwarding' if ipv6 else 'net.ipv4.ip_forward',1)
    routes=[('client','default','2001:db8:1::1' if ipv6 else '10.210.1.1'),('target','default','2001:db8:3::1' if ipv6 else '10.210.3.1'),
        ('r1','2001:db8:3::/64' if ipv6 else '10.210.3.0/30','2001:db8:2::2' if ipv6 else '10.210.2.2'),
        ('r2','2001:db8:1::/64' if ipv6 else '10.210.1.0/30','2001:db8:2::1' if ipv6 else '10.210.2.1')]
    for name,destination,gateway in routes:ip(name,family,'route','add',destination,'via',gateway)
    if ipv6:
        deadline=time.monotonic()+4
        while any('tentative' in ip(name,'-6','address','show').stdout for name in ['client','r1','r2','target']):
            assert time.monotonic()<deadline,'IPv6 link-local DAD did not finish';time.sleep(.02)
def nft(name,script):return ns(name,'nft','-f','-',input=script)
def counters(name):
    result={}
    for item in json.loads(ns(name,'nft','-j','list','ruleset').stdout)['nftables']:
        rule=item.get('rule',{})
        if 'comment' in rule:
            for expression in rule['expr']:
                if 'counter' in expression:result[rule['comment']]=expression['counter']['packets']
    return result
class DeadlineLines:
    def __init__(self,read_chunk,wait_readable,now,limit=65536):
        self.read_chunk=read_chunk;self.wait_readable=wait_readable;self.now=now
        self.pending=bytearray();self.eof=False;self.limit=limit
    def read(self,timeout):
        deadline=self.now()+timeout
        while True:
            end=self.pending.find(b'\n')
            if end>=0:
                if end+1>self.limit:raise ValueError('child status line too long')
                value=bytes(self.pending[:end+1]);del self.pending[:end+1]
                return value.decode('utf-8')
            if len(self.pending)>self.limit:raise ValueError('child status line too long')
            if self.eof:raise EOFError(('child status EOF before newline',bytes(self.pending[:160])))
            remaining=deadline-self.now()
            if remaining<=0 or not self.wait_readable(remaining):
                raise TimeoutError('child status deadline exceeded; this is not a TCP diagnosis')
            chunk=self.read_chunk(4096)
            if not chunk:self.eof=True
            else:self.pending.extend(chunk)
def reader(stream):
    # Only use this with an unbuffered binary pipe, with no other reader.
    return DeadlineLines(stream.read,lambda timeout:bool(select.select([stream],[],[],timeout)[0]),time.monotonic)
def capture(stack,root,name,interface,expression,label):
    path=root/(label+'.pcap')
    process=subprocess.Popen(['ip','netns','exec',name,'tcpdump','--immediate-mode','-Z','root','-U','-n','-i',interface,'-w',str(path),expression],
        stdout=subprocess.DEVNULL,stderr=subprocess.PIPE,bufsize=0)
    stack.callback(process.stderr.close);stack.callback(stop,process);messages=reader(process.stderr)
    deadline=time.monotonic()+3
    while time.monotonic()<deadline:
        assert process.poll() is None,'tcpdump exited before listening'
        if 'listening on' in messages.read(max(0,deadline-time.monotonic())):return process,path
    raise AssertionError('tcpdump did not become ready')
def captured(process,path):
    process.send_signal(signal.SIGINT);_,diagnostic=process.communicate(timeout=3);assert process.returncode==0,diagnostic
    return run('tcpdump','-nn','-tt','-S','-r',str(path)).stdout

def advertisement(trace,source,destination,acknowledges):
    # This deliberately accepts only the pinned tcpdump IPv4 SYN display.
    prefix=re.compile(r'^\S+ IP '+re.escape(source)+r' > '+re.escape(destination)+r': Flags \[([^]]+)\], (.*)$')
    values=set()
    for row in trace.splitlines():
        match=prefix.fullmatch(row)
        if not match:continue
        flags,fields=match.groups()
        if 'S' not in flags:continue
        assert set(flags)<=set('S.EW') and ('.' in flags)==acknowledges,('unexpected SYN flags',row)
        assert '[|' not in row and '[bad' not in row,('incomplete SYN',row)
        seq=re.findall(r'(?:^|, )seq (\d+)(?=,|$)',fields)
        ack=re.findall(r'(?:^|, )ack (\d+)(?=,|$)',fields)
        options=re.findall(r'(?:^|, )options \[([^]]*)\](?=,|$)',fields)
        assert len(seq)==1 and len(ack)==int(acknowledges) and len(options)==1,('missing/ambiguous SYN fields',row)
        mss=re.findall(r'(?:^|,)mss (\d+)(?=,|$)',options[0])
        assert len(mss)==1 and re.search(r'(?:^|, )length 0$',fields),('invalid MSS or SYN payload',row)
        value=(int(seq[0]),int(ack[0]) if ack else None,int(mss[0]))
        assert 0<=value[0]<2**32 and (value[1] is None or 0<=value[1]<2**32),row
        assert 0<value[2]<=1240,('unclamped advertisement',row)
        values.add(value)
    assert len(values)==1,('missing or inconsistent handshake',source,destination,values)
    seq,ack,mss=values.pop()
    return {'source':source,'destination':destination,'seq':seq,'ack':ack,'mss':mss}
def verify_advertisements(target_trace,client_trace,port):
    client=f'10.210.1.2.{port}';target='10.210.3.2.18080'
    syn=advertisement(target_trace,client,target,False)
    synack=advertisement(client_trace,target,client,True)
    assert synack['ack']==(syn['seq']+1)%2**32,('unrelated SYN-ACK',syn,synack)
    return {'syn_at_target':syn,'synack_at_client':synack}



def server(stack):
    code=r'''
import socketserver,hashlib
class Handler(socketserver.StreamRequestHandler):
    def handle(self):
        self.request.settimeout(3)
        header=self.rfile.readline()
        if not header:return
        length=int(header);data=self.rfile.read(length)
        if len(data)!=length:return
        self.wfile.write(hashlib.sha256(data).hexdigest().encode()+b'\n');self.wfile.flush()
class Server(socketserver.ThreadingTCPServer):allow_reuse_address=True;daemon_threads=True
service=Server(('10.210.3.2',18080),Handler);print('ready',flush=True);service.serve_forever()
'''
    process=spawn(stack,['ip','netns','exec','target','python3','-u','-c',code]);assert line(process).strip()=='ready'
def transfer(size,port):
    code=r'''
import socket,json,hashlib,sys,struct
size=int(sys.argv[1]);stream=socket.socket();stream.settimeout(1.5);connected=False
try:
    stream.bind(('10.210.1.2',int(sys.argv[2])));stream.connect(('10.210.3.2',18080));connected=True
    stream.sendall(str(size).encode()+b'\n'+b'x'*size)
    with stream.makefile('rb') as incoming:reply=incoming.readline().decode().strip()
    assert reply==hashlib.sha256(b'x'*size).hexdigest()
    print(json.dumps({'ok':True,'connected':True,'bytes':size,'sha256':reply,'snd_mss':stream.getsockopt(socket.IPPROTO_TCP,socket.TCP_MAXSEG)}))
except OSError as error:
    stream.setsockopt(socket.SOL_SOCKET,socket.SO_LINGER,struct.pack('ii',1,0))
    print(json.dumps({'ok':False,'connected':connected,'type':type(error).__name__,'errno':error.errno}))
finally:stream.close()
'''
    return json.loads(ns('client','python3','-c',code,str(size),str(port)).stdout)
def hide_fragmentation_feedback():
    nft('r1','''table ip hidden_mtu {
 chain output { type filter hook output priority 0; policy accept;
  icmp type destination-unreachable icmp code frag-needed counter drop comment "fragmentation_feedback_dropped"
 }
}
''')
with lab() as (root,stack,setting):
    topology(setting);setting('client','net.ipv4.tcp_mtu_probing',0);server(stack);hide_fragmentation_feedback()
    failed=transfer(65536,40100);assert failed['connected'] and not failed['ok'] and failed['type']=='TimeoutError'
    initial_counter=counters('r1')['fragmentation_feedback_dropped'];assert initial_counter>0
    nft('r1','''table ip clamp {
 chain forward { type filter hook forward priority 0; policy accept;
  tcp flags & syn == syn tcp option maxseg size > 1240 tcp option maxseg size set 1240 counter comment "mss_rewritten"
 }
}
''')
    # Inspect each advertisement after forwarding, at its receiving endpoint.
    client_tap,client_path=capture(stack,root,'client','c0','tcp port 18080','mss-client')
    target_tap,target_path=capture(stack,root,'target','t0','tcp port 18080','mss-target')
    recovered=transfer(65536,40101);assert recovered['ok'] and recovered['snd_mss']<=1240,recovered
    client_trace=captured(client_tap,client_path);target_trace=captured(target_tap,target_path)
    advertisements=verify_advertisements(target_trace,client_trace,40101)
    after_transfer=counters('r1');assert after_transfer['mss_rewritten']>=2 and after_transfer['fragmentation_feedback_dropped']==initial_counter,after_transfer
    large_ping=ns('client','ping','-n','-M','do','-s','1400','-c','1','-W','1','10.210.3.2',check=False)
    assert large_ping.returncode!=0 and counters('r1')['fragmentation_feedback_dropped']>initial_counter
    print(json.dumps({'before_clamp':failed,'after_clamp':recovered,'counters_after_tcp':after_transfer,
        'icmp_echo_still_fails':{'returncode':large_ping.returncode,'stdout':large_ping.stdout,'stderr':large_ping.stderr},
        'advertisements':advertisements,'capture':{'client_c0':client_trace,'target_t0':target_trace}}))
"""
name='xbcatm-mtupath-'+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)
  build=docker('build','-t',name,str(root));assert build.returncode==0,build.stderr
  result=docker('run','--rm','-i','--name',name,'--network','none','--cap-add','NET_ADMIN','--cap-add','SYS_ADMIN','--memory','256m','--pids-limit','128',name,'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)

第一台路由器抑制向外发送的 ICMP 需要分片消息，客户端关闭独立 TCP MTU 探测，从而构成受控的经典发现失败。65536 字节传输完成 TCP 建立后达到应用截止时间，路由器计数确认反馈被抑制。失败套接字随后中止，防止其残留重传被误认成后面比较中的报文。这里首先建立明确失败基线，而不是只展示添加规则后的成功。

脚本添加转发规则，匹配带 SYN 且通告 MSS 大于 1240 的 TCP 报文，将该选项设为 1240 并计数。SYN-ACK 同样带有 SYN，所以相应返回通告也被覆盖。比较使用全新连接，因为已建立字节流仍保留自身握手期间收到的选项。条件限制保证不会主动抬高原本较小的值，但不能因此省略生产环境的方向与范围审阅。

修订后的断言使用新客户端端口 40101，与失败连接的 40100 区分。在目标端捕获转发后的 SYN，在客户端捕获返回的 SYN-ACK；绝对序号把确认号与该 SYN 对应起来，同方向重传不能代替缺失的另一个方向。两个通告均须不超过 1240，聚合计数仍不能独立证明双向成立。合成包迹只验证这些检查逻辑；真实捕获、刻意单向限制的失败对照，以及完整传输仍需另行授权验证。原有报告记载 SHA-256 响应校验成功、选项开销后的实际发送 MSS 为 1228，并且需要分片反馈的丢弃计数没有增加。这些继承结果不等于修订程序已被重跑，也不表示反馈过滤已移除。

之后，载荷 1400 字节并设置禁止分片的 IPv4 echo 探测仍然失败，被抑制反馈的计数再次增加。这个较大的 ICMP 包不属于 TCP MSS 机制：修改 TCP 握手选项，既没有把它缩小，也没有修复路由器反馈策略。两种流量的不同结果，通过实际通信明确了变通措施范围，不能把 TCP 成功推广成所有协议都已恢复。

让变通措施对应实际包大小约束

本地捕获包含可能大于最终 TCP 分段的卸载表示，也可能显示看似不完整的校验和。不能仅凭这些记录，就断言 MSS 被忽略，或巨大报文已经穿过 1280 字节链路。选项、套接字诊断、计数器和经过验证的应用响应，分别提供不同处理阶段的证据。保持观察点一致，才能避免把内核表示与最终线路帧混淆。

合适上限应依据真实封装和 IP 版本推导。IPv6 外层头或额外隧道开销会改变包预算，路径变化也可能使曾经足够的固定值失效。本隔离拓扑使用已知瓶颈，规则不是自动发现算法，不能保证学习之后的每次路由变化。部署到其他网络前，应重新核对相应路径，而不是因为示例数值通过就直接复用。

生产策略生效后，应测试新连接和相关非 TCP 流量，保留对端本来更小的通告，并在上传、下载行为不同时检查两个方向。条件允许时，优先修复必要反馈与路径配置。MSS 调整的价值来自职责明确、范围匹配，并且目标 TCP 交互经过真实验证。只看到规则存在或握手完成，仍不足以宣布大传输问题已经解决。

来源