overlay/darwin: batch tun writes with sendmsg_x, drain reads, 64 pending packets

tun.WriteBatch writes up to 64 packets per sendmsg_x on the utun control
socket, and Passthrough.Flush hands its lane to it whole through the new
tio.BatchWriter (Caleb Jasik's #1896). Only the darwin tun implements
BatchWriter, so every other platform keeps one Write per packet.

tunQueue.Read drains every packet already queued on the utun, up to 64,
so a burst reaches the send batch together.

xnu lets only one packet wait on a utun control socket by default, so
each read found one packet. The tun now raises
UTUN_OPT_MAX_PENDING_PACKETS to 64, growing SO_RCVBUF to hold that many
packets first; on failure it logs and keeps xnu's default.

Tests: #1896's WriteBatch tests over sendmsg_x and the writev fallback,
the drain tests, the pending-packet setsockopts, and #1906's framing and
zero-alloc Read/Write tests.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

overlay/batch/multi_coalesce_test.go

@@ -435,3 +435,65 @@ func testPP(pkt []byte) *firewall.ParsedPacket {
 	}
 	return pp
 }
+
+// fakeTunBatchWriter records each WriteBatch call and each plain Write separately.
+type fakeTunBatchWriter struct {
+	writes  [][]byte
+	batches [][][]byte
+}
+
+func (w *fakeTunBatchWriter) Write(p []byte) (int, error) {
+	w.writes = append(w.writes, bytes.Clone(p))
+	return len(p), nil
+}
+
+func (w *fakeTunBatchWriter) WriteBatch(pkts [][]byte) error {
+	var b [][]byte
+	for _, p := range pkts {
+		b = append(b, bytes.Clone(p))
+	}
+	w.batches = append(w.batches, b)
+	return nil
+}
+
+// TestPassthroughUsesWriteBatch pins that a writer implementing tio.BatchWriter gets the whole
+// passthrough lane in one WriteBatch per Flush, in order, and no plain Writes; an empty Flush
+// makes no call.
+func TestPassthroughUsesWriteBatch(t *testing.T) {
+	w := &fakeTunBatchWriter{}
+	m := newTestMultiCoalescer(t, w)
+	k := &keySeq{epoch: 1}
+
+	var want [][]byte
+	for i := range 5 {
+		icmp := make([]byte, 28)
+		icmp[0] = 0x45
+		icmp[3] = 28
+		icmp[9] = 1
+		icmp[20] = byte(i)
+		want = append(want, icmp)
+		if err := m.Commit(icmp, k.next(), testPP(icmp)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := m.Flush(); err != nil {
+		t.Fatal(err)
+	}
+	if err := m.Flush(); err != nil {
+		t.Fatal(err)
+	}
+	if len(w.writes) != 0 {
+		t.Fatalf("got %d plain Writes, want 0", len(w.writes))
+	}
+	if len(w.batches) != 1 {
+		t.Fatalf("got %d WriteBatch calls, want 1", len(w.batches))
+	}
+	for i, p := range w.batches[0] {
+		if !bytes.Equal(p, want[i]) {
+			t.Fatalf("packet %d: got % x, want % x", i, p, want[i])
+		}
+	}
+	if len(w.batches[0]) != len(want) {
+		t.Fatalf("got %d packets, want %d", len(w.batches[0]), len(want))
+	}
+}

overlay/batch/passthrough.go

@@ -2,18 +2,23 @@ package batch
 
 import (
 	"io"
+
+	"github.com/slackhq/nebula/overlay/tio"
 )
 
-// Passthrough is MultiCoalescer's verbatim lane: no batching, packets are written at Flush in the
-// order enqueued.
+// Passthrough is MultiCoalescer's verbatim lane: no coalescing, packets are written at Flush in the
+// order enqueued, in one WriteBatch when out implements tio.BatchWriter.
 type Passthrough struct {
 	out   io.Writer
+	bw    tio.BatchWriter // nil: Flush writes one packet per Write
 	slots [][]byte
 }
 
 func NewPassthrough(w io.Writer) *Passthrough {
+	bw, _ := w.(tio.BatchWriter)
 	return &Passthrough{
 		out:   w,
+		bw:    bw,
 		slots: make([][]byte, 0, 128),
 	}
 }
@@ -26,10 +31,16 @@ func (p *Passthrough) enqueue(pkt []byte) error {
 
 func (p *Passthrough) Flush() error {
 	var firstErr error
-	for _, s := range p.slots {
-		_, err := p.out.Write(s)
-		if err != nil && firstErr == nil {
-			firstErr = err
+	if p.bw != nil {
+		if len(p.slots) > 0 {
+			firstErr = p.bw.WriteBatch(p.slots)
+		}
+	} else {
+		for _, s := range p.slots {
+			_, err := p.out.Write(s)
+			if err != nil && firstErr == nil {
+				firstErr = err
+			}
 		}
 	}
 	clear(p.slots)

overlay/tio/tio.go

@@ -145,3 +145,14 @@ func SupportsGSO(w io.Writer, want GSOProto) (GSOWriter, bool) {
 		return gw, false
 	}
 }
+
+// BatchWriter is implemented by Queues that can emit several already-segmented packets in one call.
+// Backends without a batched write do not implement it and callers loop over Write.
+//
+// pkts are borrowed: WriteBatch must not retain them past its return.
+// A returned error doesn't say which packets were delivered, so callers must not retry the batch.
+// Safe for concurrent use, like Write.
+type BatchWriter interface {
+	io.Writer
+	WriteBatch(pkts [][]byte) error
+}

overlay/tun_darwin.go

@@ -8,12 +8,14 @@ import (
 	"log/slog"
 	"net/netip"
 	"os"
+	"sync"
 	"sync/atomic"
 	"syscall"
 	"unsafe"
 
 	"github.com/gaissmai/bart"
 	"github.com/slackhq/nebula/config"
+	"github.com/slackhq/nebula/internal/msgx"
 	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 	"github.com/slackhq/nebula/util"
@@ -33,8 +35,23 @@ type tun struct {
 	// and routes for it. NEPacketTunnelProvider on darwin does this.
 	hostOwned bool
 	l         *slog.Logger
+
+	// batchMu guards batch, WriteBatch's sendmsg_x scratch.
+	batchMu sync.Mutex
+	batch   tunBatch
 }
 
+// tunBatch is WriteBatch's scratch. The kernel reads every entry in place.
+type tunBatch struct {
+	heads [msgx.Batch][4]byte
+	iovs  [msgx.Batch][2]unix.Iovec
+	hdrs  [msgx.Batch]msgx.Hdr
+	pkts  [msgx.Batch][]byte
+}
+
+// noTunSendX is set once sendmsg_x is refused on the tun; WriteBatch then writes one packet at a time.
+var noTunSendX atomic.Bool
+
 type ifReq struct {
 	Name  [unix.IFNAMSIZ]byte
 	Flags uint16
@@ -615,16 +632,12 @@ func (t *tun) Write(from []byte) (int, error) {
 		return 0, syscall.EIO
 	}
 
-	ipVer := from[0] >> 4
 	var head [4]byte
-	switch ipVer {
-	case 4:
-		head[3] = syscall.AF_INET
-	case 6:
-		head[3] = syscall.AF_INET6
-	default:
-		return 0, fmt.Errorf("unable to determine IP version from packet")
+	af, err := tunAF(from)
+	if err != nil {
+		return 0, err
 	}
+	head[3] = af
 
 	// Grab rc as a local so the compiler can devirtualize the call and keep the closure on the stack.
 	rc, err := t.f.SyscallConn()
@@ -657,6 +670,129 @@ func (t *tun) Write(from []byte) (int, error) {
 	return n - 4, nil
 }
 
+// tunAF returns the utun address-family prefix byte for an IP packet.
+func tunAF(pkt []byte) (byte, error) {
+	switch pkt[0] >> 4 {
+	case 4:
+		return syscall.AF_INET, nil
+	case 6:
+		return syscall.AF_INET6, nil
+	default:
+		return 0, fmt.Errorf("unable to determine IP version from packet")
+	}
+}
+
+// WriteBatch writes pkts to the utun device with sendmsg_x, up to msgx.Batch packets per syscall. utun is a
+// connected datagram socket, so xnu takes the batched send path, though it still hands utun each packet on its own:
+// this saves syscalls, not per-packet kernel work. Safe for concurrent use.
+//
+// An empty packet or one with no IP version is skipped and reported, as Write would. A short count goes on with the
+// rest; EMSGSIZE, which sendmsg_x returns only when the first packet is past the send buffer, sends that one through
+// Write, which reports it. Any other error means the kernel took an unknown prefix and dropped the rest, so it is
+// reported rather than retried, which could deliver packets twice. Under mbuf exhaustion sendmsg_x reports packets it
+// freed unsent as a short count, so they are lost where writev would have delivered them.
+func (t *tun) WriteBatch(pkts [][]byte) error {
+	if noTunSendX.Load() {
+		return t.writeEach(pkts)
+	}
+
+	t.batchMu.Lock()
+	defer t.batchMu.Unlock()
+	b := &t.batch
+	var firstErr error
+	keep := func(err error) {
+		if err != nil && firstErr == nil {
+			firstErr = err
+		}
+	}
+	for len(pkts) > 0 {
+		n := 0
+		for len(pkts) > 0 && n < msgx.Batch {
+			p := pkts[0]
+			pkts = pkts[1:]
+			if len(p) == 0 {
+				keep(syscall.EIO)
+				continue
+			}
+			af, err := tunAF(p)
+			if err != nil {
+				keep(err)
+				continue
+			}
+			b.heads[n] = [4]byte{3: af}
+			b.iovs[n] = [2]unix.Iovec{{Base: &b.heads[n][0], Len: 4}, {Base: &p[0], Len: uint64(len(p))}}
+			b.hdrs[n] = msgx.Hdr{Iov: &b.iovs[n][0], Iovlen: 2}
+			b.pkts[n] = p
+			n++
+		}
+
+		for off := 0; off < n; {
+			sent, err := t.sendmsgX(b.hdrs[off:n])
+			switch {
+			case err == nil && sent > 0:
+				off += sent
+			case err == nil:
+				keep(t.writeEach(b.pkts[off:n]))
+				off = n
+			case err == unix.EMSGSIZE:
+				keep(t.writeEach(b.pkts[off : off+1]))
+				off++
+			case isUnsupported(err):
+				if noTunSendX.CompareAndSwap(false, true) {
+					t.l.Warn("sendmsg_x unavailable on the tun device, writing one packet per syscall", "error", err)
+				}
+				keep(t.writeEach(b.pkts[off:n]))
+				keep(t.writeEach(pkts))
+				pkts, off = nil, n
+			default:
+				keep(err)
+				off = n
+			}
+		}
+		clear(b.iovs[:n])
+		clear(b.pkts[:n])
+	}
+	return firstErr
+}
+
+// sendmsgX hands hdrs to sendmsg_x, waiting while the socket is full, and returns how many the kernel took.
+func (t *tun) sendmsgX(hdrs []msgx.Hdr) (int, error) {
+	rc, err := t.f.SyscallConn()
+	if err != nil {
+		return 0, err
+	}
+	var sent int
+	var errno syscall.Errno
+	err = rc.Write(func(fd uintptr) bool {
+		sent, errno = msgx.Send(fd, hdrs)
+		// sendmsg_x reports EAGAIN only when it took nothing; a partial batch returns its count.
+		return errno != syscall.EAGAIN
+	})
+	if err != nil {
+		return 0, err
+	}
+	if errno != 0 {
+		return 0, errno
+	}
+	return sent, nil
+}
+
+func isUnsupported(err error) bool {
+	errno, ok := err.(syscall.Errno)
+	return ok && msgx.Unsupported(errno)
+}
+
+// writeEach writes pkts one at a time with Write.
+func (t *tun) writeEach(pkts [][]byte) error {
+	var firstErr error
+	for _, p := range pkts {
+		if _, err := t.Write(p); err != nil && firstErr == nil {
+			firstErr = err
+		}
+	}
+	return firstErr
+}
+
 func (t *tun) Networks() []netip.Prefix {
 	return t.vpnNetworks
 }
@@ -666,5 +802,137 @@ func (t *tun) Name() string {
 }
 
 func (t *tun) Queues(int) ([]tio.Queue, error) {
-	return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil
+	t.raisePendingPackets()
+	return []tio.Queue{&tunQueue{t: t, buf: make([]byte, tunReadArena)}}, nil
+}
+
+// tunReadBatch is the most packets one tunQueue.Read returns.
+const tunReadBatch = 64
+
+// tunReadArena is tunQueue's receive buffer. Draining stops once less than defaultBatchBufSize of it is left, so
+// every readv has room for the largest packet utun can return at any device MTU.
+const tunReadArena = 4 * defaultBatchBufSize
+
+// tunQueue is the darwin tun's Queue. Read drains every packet already queued on the utun, up to tunReadBatch, so
+// the caller encrypts and sends them as one batch rather than one per wakeup.
+type tunQueue struct {
+	t    *tun
+	buf  []byte
+	pkts [tunReadBatch]tio.Packet
+}
+
+// Read waits for the utun to become readable, then reads packets until it would block. An error after at least one
+// packet is dropped in favor of returning those packets; a persistent one recurs on the next Read.
+func (q *tunQueue) Read() ([]tio.Packet, error) {
+	rc, err := q.t.f.SyscallConn()
+	if err != nil {
+		return nil, err
+	}
+
+	var head [4]byte
+	n, off := 0, 0
+	var callErr error
+	err = rc.Read(func(fd uintptr) bool {
+		for n < tunReadBatch && len(q.buf)-off >= defaultBatchBufSize {
+			iovecs := [2]unix.Iovec{
+				{Base: &head[0], Len: 4},
+				{Base: &q.buf[off], Len: uint64(len(q.buf) - off)},
+			}
+			l, e := tunReadv(int(fd), iovecs[:])
+			if e != nil {
+				if errno, ok := e.(syscall.Errno); ok && errno.Temporary() {
+					// Park on the poller only while there is nothing to hand back.
+					return n > 0
+				}
+				callErr = e
+				return true
+			}
+			if l < 4 {
+				// A datagram too short to carry the AF prefix, or end of file; stop rather than spin on it.
+				return true
+			}
+			end := off + l - 4
+			q.pkts[n] = tio.Packet{Bytes: q.buf[off:end:end]}
+			n++
+			off = end
+		}
+		return true
+	})
+	if n > 0 {
+		return q.pkts[:n], nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	return nil, callErr
+}
+
+func (q *tunQueue) Write(p []byte) (int, error) { return q.t.Write(p) }
+
+func (q *tunQueue) WriteBatch(pkts [][]byte) error { return q.t.WriteBatch(pkts) }
+
+func (q *tunQueue) Close() error { return q.t.Close() }
+
+// xnu lets one packet wait on a utun control socket by default (if_utun.c utun_ctl_bind): utun_start stops moving
+// packets from the interface's send queue to the socket once one is waiting, and resumes only when a read wakes the
+// interface's starter thread. So every read finds one packet. tunPendingPackets lets a whole read batch wait
+// instead (_UTUN_OPT_MAX_PENDING_PACKETS, bsd/net/if_utun.h, set at SYSPROTO_CONTROL level).
+const (
+	tunPendingPackets             = tunReadBatch
+	_UTUN_OPT_MAX_PENDING_PACKETS = 16
+	_SYSPROTO_CONTROL             = 2
+)
+
+// tunSetsockoptInt and tunGetsockoptInt are setsockopt and getsockopt; tests swap them.
+var (
+	tunSetsockoptInt = unix.SetsockoptInt
+	tunGetsockoptInt = unix.GetsockoptInt
+)
+
+// raisePendingPackets lets up to tunPendingPackets packets wait on the utun control socket, first growing its
+// receive buffer to hold that many packets and their 4 byte AF prefix: utun drops a packet the buffer has no room
+// for, while one over the limit waits in the interface's queue. It sizes for the larger of tun.mtu and the device's
+// current MTU. Failures are logged and leave xnu's default; the tun works either way.
+func (t *tun) raisePendingPackets() {
+	if t.f == nil {
+		return
+	}
+	rc, err := t.f.SyscallConn()
+	if err != nil {
+		return
+	}
+	mtu := max(t.DefaultMTU, t.currentMTU())
+	_ = rc.Control(func(fd uintptr) {
+		pending := tunPendingPackets
+		need := pending * (mtu + 4)
+		if have, err := tunGetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_RCVBUF); err == nil && have < need {
+			if err := tunSetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_RCVBUF, need); err != nil {
+				pending = max(have/(mtu+4), 1)
+				t.l.Warn("could not grow the tun control socket's receive buffer, capping pending packets",
+					"error", err, "rcvbuf", have, "wanted", need, "pending", pending)
+			}
+		}
+		if err := tunSetsockoptInt(int(fd), _SYSPROTO_CONTROL, _UTUN_OPT_MAX_PENDING_PACKETS, pending); err != nil {
+			t.l.Warn("could not raise the tun's pending packet limit, reads will return one packet each",
+				"error", err, "pending", pending)
+		}
+	})
+}
+
+// currentMTU is the utun's MTU as the kernel has it now, or 0 if that can't be read. A host-owned device's MTU
+// comes from the host, not tun.mtu.
+func (t *tun) currentMTU() int {
+	if t.Device == "" {
+		return 0
+	}
+	s, err := unix.Socket(unix.AF_INET, unix.SOCK_DGRAM, unix.IPPROTO_IP)
+	if err != nil {
+		return 0
+	}
+	defer unix.Close(s)
+	ifr, err := unix.IoctlGetIfreqMTU(s, t.Device)
+	if err != nil {
+		return 0
+	}
+	return int(ifr.MTU)
 }

overlay/tun_darwin_test.go

@@ -0,0 +1,359 @@
+//go:build !ios && !e2e_testing
+
+package overlay
+
+import (
+	"bytes"
+	"os"
+	"syscall"
+	"testing"
+
+	"github.com/slackhq/nebula/internal/msgx"
+	"github.com/slackhq/nebula/test"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sys/unix"
+)
+
+// newSocketpairTun returns a tun writing to one end of a connected AF_UNIX datagram socketpair, which
+// takes sendmsg_x the way a utun control socket does, and the other end's fd for reading back.
+// sndbuf is the writer's send buffer, which caps the largest datagram it accepts.
+func newSocketpairTun(t *testing.T, sndbuf int) (*tun, int) {
+	t.Helper()
+	fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_DGRAM, 0)
+	require.NoError(t, err)
+	require.NoError(t, unix.SetsockoptInt(fds[0], unix.SOL_SOCKET, unix.SO_SNDBUF, sndbuf))
+	require.NoError(t, unix.SetsockoptInt(fds[1], unix.SOL_SOCKET, unix.SO_RCVBUF, 1<<20))
+	// A dropped datagram fails the read-back instead of hanging it.
+	require.NoError(t, unix.SetsockoptTimeval(fds[1], unix.SOL_SOCKET, unix.SO_RCVTIMEO, &unix.Timeval{Sec: 2}))
+	require.NoError(t, unix.SetNonblock(fds[0], true))
+	f := os.NewFile(uintptr(fds[0]), "socketpair")
+	t.Cleanup(func() {
+		_ = f.Close()
+		_ = unix.Close(fds[1])
+	})
+	return &tun{f: f, l: test.NewLogger()}, fds[1]
+}
+
+// testTunPkt returns packet i of size bytes, alternating IPv4 and IPv6, and the datagram the utun
+// device should receive for it.
+func testTunPkt(i, size int) (pkt, wire []byte) {
+	pkt = make([]byte, size)
+	pkt[0] = 0x45
+	af := byte(syscall.AF_INET)
+	if i%2 == 1 {
+		pkt[0] = 0x60
+		af = syscall.AF_INET6
+	}
+	pkt[1] = byte(i)
+	pkt[2] = byte(i >> 8)
+	return pkt, append([]byte{0, 0, 0, af}, pkt...)
+}
+
+// readTunBack asserts r receives exactly want, in order.
+func readTunBack(t *testing.T, r int, want [][]byte) {
+	t.Helper()
+	buf := make([]byte, 4096)
+	for i, w := range want {
+		n, _, err := unix.Recvfrom(r, buf, 0)
+		require.NoError(t, err, "datagram %d", i)
+		if !bytes.Equal(w, buf[:n]) {
+			t.Fatalf("datagram %d: got % x, want % x", i, buf[:n], w)
+		}
+	}
+	_, _, err := unix.Recvfrom(r, buf, unix.MSG_DONTWAIT)
+	assert.ErrorIs(t, err, unix.EAGAIN, "extra datagrams delivered")
+}
+
+// forEachTunWritePath runs fn once over sendmsg_x and once over the per-packet writev fallback.
+func forEachTunWritePath(t *testing.T, fn func(t *testing.T, fallback bool)) {
+	for _, fallback := range []bool{false, true} {
+		name := "sendmsg_x"
+		if fallback {
+			name = "writev"
+		}
+		t.Run(name, func(t *testing.T) {
+			noTunSendX.Store(fallback)
+			t.Cleanup(func() { noTunSendX.Store(false) })
+			fn(t, fallback)
+		})
+	}
+}
+
+// TestTunWriteBatch pins that WriteBatch delivers every packet whole, in order and behind the right
+// utun AF prefix, across several sendmsg_x calls, and returns nil.
+func TestTunWriteBatch(t *testing.T) {
+	forEachTunWritePath(t, func(t *testing.T, fallback bool) {
+		tn, r := newSocketpairTun(t, 1<<20)
+
+		var pkts, want [][]byte
+		for i := range 3*msgx.Batch + 5 {
+			p, w := testTunPkt(i, 60+i)
+			pkts = append(pkts, p)
+			want = append(want, w)
+		}
+		require.NoError(t, tn.WriteBatch(pkts))
+		assert.Equal(t, fallback, noTunSendX.Load())
+		readTunBack(t, r, want)
+	})
+}
+
+// TestTunWriteBatchSkipsInvalid pins that an empty packet or one with no IP version is skipped
+// without shifting the packets after it, and that the first such error is the one reported.
+func TestTunWriteBatchSkipsInvalid(t *testing.T) {
+	forEachTunWritePath(t, func(t *testing.T, fallback bool) {
+		tn, r := newSocketpairTun(t, 1<<20)
+
+		var pkts, want [][]byte
+		for i := range 20 {
+			p, w := testTunPkt(i, 60)
+			pkts = append(pkts, p)
+			want = append(want, w)
+		}
+		pkts = append(pkts[:10], append([][]byte{{0x10, 1, 2}, {}}, pkts[10:]...)...)
+
+		err := tn.WriteBatch(pkts)
+		assert.ErrorContains(t, err, "IP version")
+		readTunBack(t, r, want)
+	})
+}
+
+// TestTunWriteBatchOversize pins that a packet larger than the socket's send buffer drops only
+// itself and is reported, whether sendmsg_x reaches it mid-call (a short count) or first (EMSGSIZE
+// for the whole call). The send buffer is 1024 bytes and the oversize packet 2000.
+func TestTunWriteBatchOversize(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		at   int
+	}{
+		{"mid-call", msgx.Batch + 6},
+		{"first-in-call", msgx.Batch},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			forEachTunWritePath(t, func(t *testing.T, fallback bool) {
+				tn, r := newSocketpairTun(t, 1024)
+
+				var pkts, want [][]byte
+				for i := range 2*msgx.Batch + 10 {
+					size := 100
+					if i == tc.at {
+						size = 2000
+					}
+					p, w := testTunPkt(i, size)
+					pkts = append(pkts, p)
+					if i != tc.at {
+						want = append(want, w)
+					}
+				}
+				err := tn.WriteBatch(pkts)
+				assert.ErrorIs(t, err, unix.EMSGSIZE)
+				assert.Equal(t, fallback, noTunSendX.Load())
+				readTunBack(t, r, want)
+			})
+		})
+	}
+}
+
+// TestTunQueueReadDrains pins that Read hands back every queued packet, AF prefix stripped and in
+// order, capped at tunReadBatch per call, and doesn't wait for more once the queue is empty.
+func TestTunQueueReadDrains(t *testing.T) {
+	tn, w := newSocketpairTun(t, 1<<20)
+	// An AF_UNIX datagram send is bounded by the receiver's buffer, here the tun's end.
+	rc, err := tn.f.SyscallConn()
+	require.NoError(t, err)
+	require.NoError(t, rc.Control(func(fd uintptr) {
+		require.NoError(t, unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_RCVBUF, 1<<20))
+	}))
+	q := &tunQueue{t: tn, buf: make([]byte, tunReadArena)}
+	total := tunReadBatch + 3
+	var want [][]byte
+	for i := range total {
+		pkt, wire := testTunPkt(i, 60+i)
+		_, err := unix.Write(w, wire)
+		require.NoError(t, err)
+		want = append(want, pkt)
+	}
+
+	var got [][]byte
+	for _, size := range []int{tunReadBatch, 3} {
+		pkts, err := q.Read()
+		require.NoError(t, err)
+		require.Len(t, pkts, size)
+		for _, p := range pkts {
+			got = append(got, p.Clone().Bytes)
+		}
+	}
+	assert.Equal(t, want, got)
+}
+
+// TestTunQueueReadClipsPackets pins that each returned packet's capacity ends at its own bytes, so
+// appending to one can't overwrite the next.
+func TestTunQueueReadClipsPackets(t *testing.T) {
+	tn, w := newSocketpairTun(t, 1<<20)
+	q := &tunQueue{t: tn, buf: make([]byte, tunReadArena)}
+	for i := range 2 {
+		_, wire := testTunPkt(i, 100)
+		_, err := unix.Write(w, wire)
+		require.NoError(t, err)
+	}
+	pkts, err := q.Read()
+	require.NoError(t, err)
+	require.Len(t, pkts, 2)
+	assert.Equal(t, len(pkts[0].Bytes), cap(pkts[0].Bytes))
+}
+
+// TestTunRaisePendingPackets pins that the pending limit is set at SYSPROTO_CONTROL level after the receive buffer is
+// grown to hold that many packets of the tun's MTU, and capped to what the buffer holds when it can't grow.
+func TestTunRaisePendingPackets(t *testing.T) {
+	type call struct{ level, opt, value int }
+	stub := func(t *testing.T, rcvbuf int, failLevel, failOpt int) *[]call {
+		set, get := tunSetsockoptInt, tunGetsockoptInt
+		var calls []call
+		tunSetsockoptInt = func(fd, level, opt, value int) error {
+			calls = append(calls, call{level, opt, value})
+			if level == failLevel && opt == failOpt {
+				return unix.ENOBUFS
+			}
+			return nil
+		}
+		tunGetsockoptInt = func(fd, level, opt int) (int, error) { return rcvbuf, nil }
+		t.Cleanup(func() { tunSetsockoptInt, tunGetsockoptInt = set, get })
+		return &calls
+	}
+	for _, tc := range []struct {
+		name      string
+		rcvbuf    int
+		mtu       int
+		failLevel int
+		failOpt   int
+		want      []call
+	}{
+		{"grows rcvbuf first", 1 << 16, 9000, -1, -1, []call{
+			{unix.SOL_SOCKET, unix.SO_RCVBUF, 64 * 9004},
+			{2, 16, 64},
+		}},
+		{"rcvbuf already big enough", 1 << 20, 1300, -1, -1, []call{{2, 16, 64}}},
+		{"rcvbuf refused caps pending", 10 * 1304, 1300, unix.SOL_SOCKET, unix.SO_RCVBUF, []call{
+			{unix.SOL_SOCKET, unix.SO_RCVBUF, 64 * 1304},
+			{2, 16, 10},
+		}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			calls := stub(t, tc.rcvbuf, tc.failLevel, tc.failOpt)
+			tn, _ := newSocketpairTun(t, 1<<20)
+			tn.DefaultMTU = tc.mtu
+			_, err := tn.Queues(1)
+			require.NoError(t, err)
+			assert.Equal(t, tc.want, *calls)
+		})
+	}
+}
+
+func TestTun_WriteFraming(t *testing.T) {
+	r, w, err := os.Pipe()
+	require.NoError(t, err)
+	defer r.Close()
+	defer w.Close()
+	tn := &tun{f: w}
+
+	for _, tc := range []struct {
+		name string
+		pkt  []byte
+		af   byte
+	}{
+		// Literal families rather than syscall.AF_*, so a wrong constant in Write can't also fix the expectation.
+		{"ipv4", []byte{0x45, 1, 2, 3, 4, 5}, 2},
+		{"ipv6", []byte{0x60, 1, 2, 3, 4, 5}, 30},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			n, err := tn.Write(tc.pkt)
+			require.NoError(t, err)
+			require.Equal(t, len(tc.pkt), n)
+
+			got := make([]byte, 64)
+			n, err = r.Read(got)
+			require.NoError(t, err)
+			require.Equal(t, append([]byte{0, 0, 0, tc.af}, tc.pkt...), got[:n])
+		})
+	}
+
+	_, err = tn.Write([]byte{0x10, 1, 2, 3})
+	require.Error(t, err)
+	_, err = tn.Write([]byte{})
+	require.ErrorIs(t, err, syscall.EIO)
+
+	// The rejected writes must not have put anything on the fd, so the next read sees only this packet.
+	pkt := []byte{0x45, 9, 9}
+	_, err = tn.Write(pkt)
+	require.NoError(t, err)
+	got := make([]byte, 64)
+	n, err := r.Read(got)
+	require.NoError(t, err)
+	require.Equal(t, append([]byte{0, 0, 0, 2}, pkt...), got[:n])
+}
+
+func TestTun_ReadStripsHeader(t *testing.T) {
+	r, w, err := os.Pipe()
+	require.NoError(t, err)
+	defer r.Close()
+	defer w.Close()
+	tn := &tun{f: r}
+
+	pkt := []byte{0x45, 1, 2, 3, 4, 5}
+	_, err = w.Write(append([]byte{0, 0, 0, syscall.AF_INET}, pkt...))
+	require.NoError(t, err)
+
+	got := make([]byte, 64)
+	n, err := tn.Read(got)
+	require.NoError(t, err)
+	require.Equal(t, pkt, got[:n])
+}
+
+func TestTun_ReadShort(t *testing.T) {
+	r, w, err := os.Pipe()
+	require.NoError(t, err)
+	defer r.Close()
+	defer w.Close()
+	tn := &tun{f: r}
+
+	// Fewer bytes than the protocol header is not a packet.
+	_, err = w.Write([]byte{0, 0})
+	require.NoError(t, err)
+	n, err := tn.Read(make([]byte, 64))
+	require.NoError(t, err)
+	require.Zero(t, n)
+
+	// The next packet reads cleanly only if the short read consumed exactly what was written.
+	pkt := []byte{0x45, 1, 2, 3}
+	_, err = w.Write(append([]byte{0, 0, 0, syscall.AF_INET}, pkt...))
+	require.NoError(t, err)
+	got := make([]byte, 64)
+	n, err = tn.Read(got)
+	require.NoError(t, err)
+	require.Equal(t, pkt, got[:n])
+}
+
+// A heap allocation per packet rarely shows in a short throughput run, only across GC pauses, so
+// this pins Read and Write to zero allocations directly.
+func TestTun_ReadWriteDoNotAllocate(t *testing.T) {
+	r, w, err := os.Pipe()
+	require.NoError(t, err)
+	defer r.Close()
+	defer w.Close()
+	tw := &tun{f: w}
+	tr := &tun{f: r}
+
+	pkts := [][]byte{bytes.Repeat([]byte{0x45}, 1300), bytes.Repeat([]byte{0x60}, 1300)}
+	buf := make([]byte, 1500)
+	i := 0
+	allocs := testing.AllocsPerRun(1000, func() {
+		i++
+		if _, err := tw.Write(pkts[i%len(pkts)]); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := tr.Read(buf); err != nil {
+			t.Fatal(err)
+		}
+	})
+	require.Zero(t, allocs)
+}