The path MTU is the smallest link MTU along a particular route. A sender cannot determine it solely from its own interface’s MTU because an intermediate link may carry smaller packets. Classic IPv4 Path MTU Discovery sends packets with Don’t Fragment set and uses ICMP fragmentation-needed feedback to reduce the estimated packet size. A working route and a working size estimate are separate requirements.

When it breaks

A tunnel or another constrained intermediate link can make large transfers fail while a TCP handshake and small request succeed. Their packets fit through the bottleneck, so those successes do not validate full-sized data. Repeatedly testing only a short ping or a tiny health response may therefore miss a path-MTU problem that affects uploads, larger responses, or sustained transfers.

Filtering the required ICMP feedback can create a black hole for oversized packets. The sender sees missing progress but lacks the classic signal identifying the smaller MTU. Other probing mechanisms may help, depending on transport and configuration; this is not proof that every TCP stack always hangs under such filtering. Identify which discovery or black-hole recovery mechanism is actually enabled before drawing conclusions.

Path information can also be cached. A test conducted after a successful discovery may behave differently from a fresh client, even after a firewall rule changes. Reusing that state can hide a broken feedback path. A route change or asymmetric return path complicates the picture further, so record direction, source, destination, and the conditions under which the estimate was learned.

Check it quickly

Save this as check-path-mtu.py and run python3 check-path-mtu.py with Docker. The disposable environment uses Python 3.11.16, iproute2 6.1.0, nftables 1.0.6, and tcpdump 4.99.3, tested on Linux 6.12.76. NET_ADMIN and SYS_ADMIN support four nested namespaces. Two routers connect the endpoints through newly created veth links; no host route or interface is changed:

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,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
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,text=True)
    stack.callback(stop,process);deadline=time.monotonic()+3
    while time.monotonic()<deadline:
        assert process.poll() is None,'capture exited'
        if select.select([process.stderr],[],[],.1)[0] and 'listening on' in process.stderr.readline():return process,path
    raise AssertionError('capture 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','-vv','-r',str(path)).stdout


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):
    code=r'''
import socket,json,hashlib,sys,struct
size=int(sys.argv[1]);stream=socket.socket();stream.settimeout(1.5);connected=False
try:
    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)).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"
 }
}
''')
results={}
for hidden in [False,True]:
    with lab() as (root,stack,setting):
        topology(setting);setting('client','net.ipv4.tcp_mtu_probing',0);server(stack)
        if hidden:hide_fragmentation_feedback()
        small=transfer(128);assert small['ok']
        before=ip('client','route','get','10.210.3.2').stdout
        tap,path=capture(stack,root,'client','c0','icmp or tcp port 18080','pmtu')
        large=transfer(65536);trace=captured(tap,path);after=ip('client','route','get','10.210.3.2').stdout
        if hidden:
            assert not large['ok'] and large['connected'] and large['type']=='TimeoutError',large
            count=counters('r1');assert count['fragmentation_feedback_dropped']>0
        else:
            assert large['ok'] and 'mtu 1280' in after,(large,after)
            assert 'unreachable - need to frag (mtu 1280)' in trace,trace;count={}
        results[str(hidden)]={'small_transfer':small,'large_transfer':large,'route_before':before,'route_after':after,
            'icmp_feedback_filtered':hidden,'counter':count,'capture':trace}
print(json.dumps(results))
"""
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)

Both endpoint links have MTU 1500, while the router-to-router link has MTU 1280. This placement matters: neither endpoint simply advertises a smaller local-interface MSS during setup. The client explicitly disables TCP’s separate MTU probing feature so the comparison isolates classic ICMP-based discovery. Each trial creates fresh namespaces, avoiding a learned destination estimate leaking from the healthy case into the filtered case.

With feedback permitted, a 128-byte transfer succeeds first. A new connection then sends 65536 bytes and verifies the server’s SHA-256 response. The capture contains an ICMP message reporting that fragmentation is needed with MTU 1280. The client’s route lookup subsequently shows a cached MTU of 1280, and the socket reports an effective sending MSS of 1228 in this timestamp-enabled exchange.

The second topology suppresses only the first router’s outgoing fragmentation-needed messages. Its small transfer still succeeds, and the larger transfer completes TCP setup before timing out. The rule counter records actual suppressed feedback, while the client’s route lookup has not learned the smaller MTU. These observations connect size-dependent failure to the deliberately missing discovery signal instead of blaming a stopped listener.

Failed test sockets are explicitly aborted so their queued retransmissions do not contaminate subsequent observations. The 1.5-second application deadline bounds this exercise; it is not the operating system’s complete retry lifetime. Successful cases validate exact payload content, while failed cases preserve both the local connection result and timeout type. That distinction keeps a successful handshake from being reported as successful data delivery.

Interpret captures at the correct processing point

Local captures may show large offload representations or apparent partial checksums. Such records are not automatically the final segments placed on a physical link, so a large captured length does not prove that an oversized wire packet crossed the bottleneck. Correlate route state, ICMP feedback, effective MSS, and application results rather than interpreting one capture field without its processing context.

Use several meaningful payload sizes when diagnosing a path, and account for IP, transport, option, and tunnel overhead. A ping payload size is not the full IP packet size, and an application write is not one TCP segment. Preserve the original failing traffic direction because success with a small response in the reverse direction may not exercise the same constraint.

Restore required feedback or correct the actual path configuration, then repeat a representative transfer from a fresh state. A transport-specific workaround may relieve one application without repairing discovery for other protocols. This laboratory demonstrates classic IPv4 discovery and its deliberate failure; it does not certify every alternate probing algorithm or establish a universal timeout for an arbitrary internet path.

Sources