checkpt

overlay/tio/single.go

@@ -0,0 +1,50 @@
+package tio
+
+import "io"
+
+// singleQueue adapts a legacy one-datagram-per-Read source into a Queue.
+// Read fills a private scratch buffer and returns exactly one Packet whose
+// Bytes borrow from that buffer, valid only until the next Read, per the
+// Queue contract. Single-reader like every Queue; Write is exactly as safe
+// for concurrent use as the underlying source's Write.
+type singleQueue struct {
+	rw     io.ReadWriter
+	closer io.Closer // nil: Close is a no-op (the source is shared and owned elsewhere)
+	buf    []byte
+	ret    [1]Packet
+}
+
+// NewSingleQueue wraps a one-datagram-per-Read ReadWriteCloser (a legacy tun
+// device) into a Queue. bufSize is the per-queue read scratch size and must
+// be at least the largest datagram the source can return. Close closes rwc.
+func NewSingleQueue(rwc io.ReadWriteCloser, bufSize int) Queue {
+	return &singleQueue{rw: rwc, closer: rwc, buf: make([]byte, bufSize)}
+}
+
+// NewSingleQueueNoClose is NewSingleQueue for a source owned by someone else,
+// e.g. several queues sharing one device. Close on the returned Queue is a
+// no-op so one queue can't tear the shared source out from under its
+// siblings; the owner remains responsible for closing the source itself.
+func NewSingleQueueNoClose(rw io.ReadWriter, bufSize int) Queue {
+	return &singleQueue{rw: rw, buf: make([]byte, bufSize)}
+}
+
+func (q *singleQueue) Read() ([]Packet, error) {
+	n, err := q.rw.Read(q.buf)
+	if err != nil {
+		return nil, err
+	}
+	q.ret[0] = Packet{Bytes: q.buf[:n]}
+	return q.ret[:], nil
+}
+
+func (q *singleQueue) Write(p []byte) (int, error) {
+	return q.rw.Write(p)
+}
+
+func (q *singleQueue) Close() error {
+	if q.closer == nil {
+		return nil
+	}
+	return q.closer.Close()
+}

overlay/tun_android.go

@@ -19,32 +19,12 @@ import (
 )
 
 type tun struct {
-	rwc         io.ReadWriteCloser
+	io.ReadWriteCloser
 	fd          int
 	vpnNetworks []netip.Prefix
 	Routes      atomic.Pointer[[]Route]
 	routeTree   atomic.Pointer[bart.Table[routing.Gateways]]
 	l           *slog.Logger
-
-	readBuf  []byte
-	batchRet [1]tio.Packet
-}
-
-func (t *tun) Read() ([]tio.Packet, error) {
-	n, err := t.rwc.Read(t.readBuf)
-	if err != nil {
-		return nil, err
-	}
-	t.batchRet[0] = tio.Packet{Bytes: t.readBuf[:n]}
-	return t.batchRet[:], nil
-}
-
-func (t *tun) Write(p []byte) (int, error) {
-	return t.rwc.Write(p)
-}
-
-func (t *tun) Close() error {
-	return t.rwc.Close()
 }
 
 func newTunFromFd(c *config.C, l *slog.Logger, deviceFd int, vpnNetworks []netip.Prefix) (*tun, error) {
@@ -53,11 +33,10 @@ func newTunFromFd(c *config.C, l *slog.Logger, deviceFd int, vpnNetworks []netip
 	file := os.NewFile(uintptr(deviceFd), "/dev/net/tun")
 
 	t := &tun{
-		rwc:         file,
-		fd:          deviceFd,
-		vpnNetworks: vpnNetworks,
-		l:           l,
-		readBuf:     make([]byte, defaultBatchBufSize),
+		ReadWriteCloser: file,
+		fd:              deviceFd,
+		vpnNetworks:     vpnNetworks,
+		l:               l,
 	}
 
 	err := t.reload(c, true)
@@ -127,5 +106,5 @@ func (t *tun) NewMultiQueueReader() error {
 }
 
 func (t *tun) Readers() []tio.Queue {
-	return []tio.Queue{t}
+	return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}
 }

overlay/tun_darwin.go

@@ -31,9 +31,6 @@ type tun struct {
 	routeTree   atomic.Pointer[bart.Table[routing.Gateways]]
 	linkAddr    *netroute.LinkAddr
 	l           *slog.Logger
-
-	readBuf  []byte
-	batchRet [1]tio.Packet
 }
 
 type ifReq struct {
@@ -129,7 +126,6 @@ func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, _ bool) (*t
 		vpnNetworks: vpnNetworks,
 		DefaultMTU:  c.GetInt("tun.mtu", DefaultMTU),
 		l:           l,
-		readBuf:     make([]byte, defaultBatchBufSize),
 	}
 
 	err = t.reload(c, true)
@@ -519,9 +515,9 @@ func tunWritev(fd int, iovecs []unix.Iovec) (n int, err error)
 //go:noescape
 func tunReadv(fd int, iovecs []unix.Iovec) (n int, err error)
 
-// readOne pulls one IP packet off the utun device, scattering the 4 byte protocol header away from
+// Read pulls one IP packet off the utun device, scattering the 4 byte protocol header away from
 // the packet so the payload lands directly in to.
-func (t *tun) readOne(to []byte) (int, error) {
+func (t *tun) Read(to []byte) (int, error) {
 	var head [4]byte
 
 	rc, err := t.f.SyscallConn()
@@ -554,15 +550,6 @@ func (t *tun) readOne(to []byte) (int, error) {
 	return n - 4, nil
 }
 
-func (t *tun) Read() ([]tio.Packet, error) {
-	n, err := t.readOne(t.readBuf)
-	if err != nil {
-		return nil, err
-	}
-	t.batchRet[0] = tio.Packet{Bytes: t.readBuf[:n]}
-	return t.batchRet[:], nil
-}
-
 // Write pushes one IP packet onto the utun device. Only valid for single threaded use.
 func (t *tun) Write(from []byte) (int, error) {
 	if len(from) == 0 {
@@ -628,5 +615,5 @@ func (t *tun) NewMultiQueueReader() error {
 }
 
 func (t *tun) Readers() []tio.Queue {
-	return []tio.Queue{t}
+	return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}
 }

overlay/tun_disabled.go

@@ -25,39 +25,21 @@ type disabledTun struct {
 	numReaders int
 }
 
-// disabledQueue is one tio.Queue view onto a shared disabledTun. Each queue
-// owns a private batchRet so concurrent Read calls from different reader
-// goroutines do not race on the returned slice.
-type disabledQueue struct {
-	parent   *disabledTun
-	batchRet [1]tio.Packet
-}
-
-func (q *disabledQueue) Read() ([]tio.Packet, error) {
-	r, ok := <-q.parent.read
+// Read hands the next queued packet to a reader, copying it into b. Reads
+// from concurrent queues are safe: the channel receive serializes them and
+// each queue copies into its own private scratch buffer.
+func (t *disabledTun) Read(b []byte) (int, error) {
+	r, ok := <-t.read
 	if !ok {
-		return nil, io.EOF
+		return 0, io.EOF
 	}
 
-	q.parent.tx.Inc(1)
-	if q.parent.l.Enabled(context.Background(), slog.LevelDebug) {
-		q.parent.l.Debug("Write payload", "raw", prettyPacket(r))
+	t.tx.Inc(1)
+	if t.l.Enabled(context.Background(), slog.LevelDebug) {
+		t.l.Debug("Write payload", "raw", prettyPacket(r))
 	}
 
-	q.batchRet[0] = tio.Packet{Bytes: r}
-	return q.batchRet[:], nil
-}
-
-// Write on a queue forwards to the underlying disabledTun. All queues share
-// one ICMP-handling/log path so this is a thin pass-through.
-func (q *disabledQueue) Write(b []byte) (int, error) {
-	return q.parent.Write(b)
-}
-
-// Close on a queue is a no-op. The shared channel and metrics are owned by
-// the disabledTun; Close on the device tears them down once for everybody.
-func (q *disabledQueue) Close() error {
-	return nil
+	return copy(b, r), nil
 }
 
 func newDisabledTun(vpnNetworks []netip.Prefix, queueLen int, metricsEnabled bool, l *slog.Logger) *disabledTun {
@@ -138,7 +120,9 @@ func (t *disabledTun) NewMultiQueueReader() error {
 func (t *disabledTun) Readers() []tio.Queue {
 	out := make([]tio.Queue, t.numReaders)
 	for i := range t.numReaders {
-		out[i] = &disabledQueue{parent: t}
+		// NoClose: the shared channel and metrics are owned by the
+		// disabledTun; Close on the device tears them down once for everybody.
+		out[i] = tio.NewSingleQueueNoClose(t, defaultBatchBufSize)
 	}
 	return out
 }

overlay/tun_freebsd.go

@@ -102,9 +102,6 @@ type tun struct {
 	readPoll  [2]unix.PollFd
 	writePoll [2]unix.PollFd
 	closed    atomic.Bool
-
-	readBuf  []byte
-	batchRet [1]tio.Packet
 }
 
 // blockOnRead waits until the tun fd is readable or shutdown has been signaled.
@@ -159,16 +156,7 @@ func (t *tun) blockOnWrite() error {
 	return nil
 }
 
-func (t *tun) Read() ([]tio.Packet, error) {
-	n, err := t.readOne(t.readBuf)
-	if err != nil {
-		return nil, err
-	}
-	t.batchRet[0] = tio.Packet{Bytes: t.readBuf[:n]}
-	return t.batchRet[:], nil
-}
-
-func (t *tun) readOne(to []byte) (int, error) {
+func (t *tun) Read(to []byte) (int, error) {
 	// first 4 bytes is protocol family, in network byte order
 	var head [4]byte
 	iovecs := [2]syscall.Iovec{
@@ -386,7 +374,6 @@ func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, _ bool) (*t
 		MTU:         c.GetInt("tun.mtu", DefaultMTU),
 		l:           l,
 		fd:          fd,
-		readBuf:     make([]byte, defaultBatchBufSize),
 		shutdownR:   shutdownR,
 		shutdownW:   shutdownW,
 		readPoll: [2]unix.PollFd{
@@ -606,7 +593,7 @@ func (t *tun) addRoutes(logErrors bool) error {
 }
 
 func (t *tun) Readers() []tio.Queue {
-	return []tio.Queue{t}
+	return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}
 }
 
 func (t *tun) removeRoutes(routes []Route) error {

overlay/tun_netbsd.go

@@ -66,22 +66,10 @@ type tun struct {
 	l           *slog.Logger
 	f           *os.File
 	fd          int
-
-	readBuf  []byte
-	batchRet [1]tio.Packet
-}
-
-func (t *tun) Read() ([]tio.Packet, error) {
-	n, err := t.readOne(t.readBuf)
-	if err != nil {
-		return nil, err
-	}
-	t.batchRet[0] = tio.Packet{Bytes: t.readBuf[:n]}
-	return t.batchRet[:], nil
 }
 
 func (t *tun) Readers() []tio.Queue {
-	return []tio.Queue{t}
+	return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}
 }
 
 var deviceNameRE = regexp.MustCompile(`^tun[0-9]+$`)
@@ -118,7 +106,6 @@ func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, _ bool) (*t
 		vpnNetworks: vpnNetworks,
 		MTU:         c.GetInt("tun.mtu", DefaultMTU),
 		l:           l,
-		readBuf:     make([]byte, defaultBatchBufSize),
 	}
 
 	err = t.reload(c, true)
@@ -158,7 +145,7 @@ func (t *tun) Close() error {
 	return nil
 }
 
-func (t *tun) readOne(to []byte) (int, error) {
+func (t *tun) Read(to []byte) (int, error) {
 	rc, err := t.f.SyscallConn()
 	if err != nil {
 		return 0, fmt.Errorf("failed to get syscall conn for tun: %w", err)

overlay/tun_openbsd.go

@@ -57,18 +57,6 @@ type tun struct {
 	l           *slog.Logger
 	f           *os.File
 	fd          int
-
-	readBuf  []byte
-	batchRet [1]tio.Packet
-}
-
-func (t *tun) Read() ([]tio.Packet, error) {
-	n, err := t.readOne(t.readBuf)
-	if err != nil {
-		return nil, err
-	}
-	t.batchRet[0] = tio.Packet{Bytes: t.readBuf[:n]}
-	return t.batchRet[:], nil
 }
 
 var deviceNameRE = regexp.MustCompile(`^tun[0-9]+$`)
@@ -105,7 +93,6 @@ func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, _ bool) (*t
 		vpnNetworks: vpnNetworks,
 		MTU:         c.GetInt("tun.mtu", DefaultMTU),
 		l:           l,
-		readBuf:     make([]byte, defaultBatchBufSize),
 	}
 
 	err = t.reload(c, true)
@@ -151,9 +138,9 @@ func tunWritev(fd int, iovecs []unix.Iovec) (n int, err error)
 //go:noescape
 func tunReadv(fd int, iovecs []unix.Iovec) (n int, err error)
 
-// readOne pulls one IP packet off the tun device, scattering the 4 byte protocol header away from
+// Read pulls one IP packet off the tun device, scattering the 4 byte protocol header away from
 // the packet so the payload lands directly in to.
-func (t *tun) readOne(to []byte) (int, error) {
+func (t *tun) Read(to []byte) (int, error) {
 	var head [4]byte
 
 	rc, err := t.f.SyscallConn()
@@ -439,7 +426,7 @@ func (t *tun) deviceBytes() (o [16]byte) {
 }
 
 func (t *tun) Readers() []tio.Queue {
-	return []tio.Queue{t}
+	return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}
 }
 
 func addRoute(prefix netip.Prefix, gateways []netip.Prefix) error {

overlay/tun_tester.go

@@ -29,8 +29,6 @@ type TestTun struct {
 	closed    atomic.Bool
 	rxPackets chan []byte // Packets to receive into nebula
 	TxPackets chan []byte // Packets transmitted outside by nebula
-
-	batchRet [1]tio.Packet
 }
 
 func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, _ bool) (*TestTun, error) {
@@ -51,9 +49,6 @@ func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, _ bool) (*T
 		l:           l,
 		rxPackets:   make(chan []byte, 10),
 		TxPackets:   make(chan []byte, 10),
-		batchRet: [1]tio.Packet{
-			tio.Packet{Bytes: make([]byte, udp.MTU)},
-		},
 	}, nil
 }
 
@@ -168,17 +163,7 @@ func (t *TestTun) Close() error {
 	return nil
 }
 
-func (t *TestTun) Read() ([]tio.Packet, error) {
-	t.batchRet[0].Bytes = t.batchRet[0].Bytes[:udp.MTU]
-	n, err := t.read(t.batchRet[0].Bytes)
-	if err != nil {
-		return nil, err
-	}
-	t.batchRet[0].Bytes = t.batchRet[0].Bytes[:n]
-	return t.batchRet[:], nil
-}
-
-func (t *TestTun) read(b []byte) (int, error) {
+func (t *TestTun) Read(b []byte) (int, error) {
 	p, ok := <-t.rxPackets
 	if !ok {
 		return 0, os.ErrClosed
@@ -194,7 +179,7 @@ func (t *TestTun) read(b []byte) (int, error) {
 }
 
 func (t *TestTun) Readers() []tio.Queue {
-	return []tio.Queue{t}
+	return []tio.Queue{tio.NewSingleQueue(t, udp.MTU)}
 }
 
 func (t *TestTun) SupportsMultiqueue() bool {

overlay/tun_windows.go

@@ -45,18 +45,10 @@ type winTun struct {
 	l               *slog.Logger
 
 	tun *wintun.NativeTun
-
-	readBuf  []byte
-	batchRet [1]tio.Packet
 }
 
-func (t *winTun) Read() ([]tio.Packet, error) {
-	n, err := t.tun.Read(t.readBuf, 0)
-	if err != nil {
-		return nil, err
-	}
-	t.batchRet[0] = tio.Packet{Bytes: t.readBuf[:n]}
-	return t.batchRet[:], nil
+func (t *winTun) Read(b []byte) (int, error) {
+	return t.tun.Read(b, 0)
 }
 
 func newTunFromFd(_ *config.C, _ *slog.Logger, _ int, _ []netip.Prefix) (Device, error) {
@@ -81,7 +73,6 @@ func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, _ bool) (*w
 	}
 
 	t := &winTun{
-		readBuf:         make([]byte, defaultBatchBufSize),
 		Device:          deviceName,
 		vpnNetworks:     vpnNetworks,
 		MTU:             c.GetInt("tun.mtu", DefaultMTU),
@@ -281,7 +272,7 @@ func (t *winTun) NewMultiQueueReader() error {
 }
 
 func (t *winTun) Readers() []tio.Queue {
-	return []tio.Queue{t}
+	return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}
 }
 
 func (t *winTun) Close() error {

overlay/user.go

@@ -39,40 +39,6 @@ type UserDevice struct {
 	inboundWriter *io.PipeWriter
 }
 
-// userDeviceQueue is a single tio.Queue over a UserDevice's shared pipes.
-// One is handed to each tun read goroutine by Readers(). All queues delegate
-// reads to the same outboundReader and writes to the same inboundWriter (the
-// io.Pipe serializes concurrent callers), but every queue owns a private
-// readBuf/batchRet so the borrowed Packet.Bytes slice one goroutine returns is
-// never clobbered by another goroutine's concurrent Read.
-type userDeviceQueue struct {
-	outboundReader *io.PipeReader
-	inboundWriter  *io.PipeWriter
-
-	readBuf  []byte
-	batchRet [1]tio.Packet
-}
-
-func (q *userDeviceQueue) Read() ([]tio.Packet, error) {
-	n, err := q.outboundReader.Read(q.readBuf)
-	if err != nil {
-		return nil, err
-	}
-	q.batchRet[0] = tio.Packet{Bytes: q.readBuf[:n]}
-	return q.batchRet[:], nil
-}
-
-func (q *userDeviceQueue) Write(p []byte) (int, error) {
-	return q.inboundWriter.Write(p)
-}
-
-// Close is a no-op: the shared pipes are owned by the UserDevice and torn
-// down by UserDevice.Close, so an individual queue must not close them out
-// from under its siblings.
-func (q *userDeviceQueue) Close() error {
-	return nil
-}
-
 func (d *UserDevice) Activate() error {
 	return nil
 }
@@ -95,13 +61,11 @@ func (d *UserDevice) NewMultiQueueReader() error {
 func (d *UserDevice) Readers() []tio.Queue {
 	out := make([]tio.Queue, d.numReaders)
 	for i := range d.numReaders {
-		// Each queue shares the underlying pipes but owns its own scratch
-		// buffer so concurrent Reads across queues never alias.
-		out[i] = &userDeviceQueue{
-			outboundReader: d.outboundReader,
-			inboundWriter:  d.inboundWriter,
-			readBuf:        make([]byte, defaultBatchBufSize),
-		}
+		// All queues share the underlying pipes (the io.Pipe serializes
+		// concurrent callers) but each owns a private scratch buffer so
+		// concurrent Reads across queues never alias. NoClose: the pipes are
+		// owned by the UserDevice and torn down once by UserDevice.Close.
+		out[i] = tio.NewSingleQueueNoClose(d, defaultBatchBufSize)
 	}
 	return out
 }
@@ -110,6 +74,10 @@ func (d *UserDevice) Pipe() (*io.PipeReader, *io.PipeWriter) {
 	return d.inboundReader, d.outboundWriter
 }
 
+func (d *UserDevice) Read(p []byte) (n int, err error) {
+	return d.outboundReader.Read(p)
+}
+
 func (d *UserDevice) Write(p []byte) (n int, err error) {
 	return d.inboundWriter.Write(p)
 }

overlay/user_test.go

@@ -26,10 +26,11 @@ func newTestUserDevice(t *testing.T) *UserDevice {
 
 // TestUserDeviceReadersDistinctBuffers is the regression test for the
 // multiqueue packet-corruption bug: Readers() used to hand the same
-// *UserDevice (and therefore the same readBuf/batchRet) to every queue, so one
-// reader's borrowed Packet.Bytes was overwritten by another reader's
-// concurrent Read. Readers() must now return numReaders DISTINCT queue objects,
-// each with its own backing buffer.
+// *UserDevice (and therefore the same read scratch buffer) to every queue, so
+// one reader's borrowed Packet.Bytes was overwritten by another reader's
+// concurrent Read. Readers() must now return numReaders DISTINCT queue
+// objects, each with its own backing buffer — verified behaviorally below by
+// holding one queue's borrowed slice across the other queue's Read.
 func TestUserDeviceReadersDistinctBuffers(t *testing.T) {
 	d := newTestUserDevice(t)
 
@@ -43,21 +44,10 @@ func TestUserDeviceReadersDistinctBuffers(t *testing.T) {
 		t.Fatalf("Readers() returned %d queues, want 2", len(readers))
 	}
 
-	q0 := readers[0].(*userDeviceQueue)
-	q1 := readers[1].(*userDeviceQueue)
-
 	// Distinct queue objects.
-	if q0 == q1 {
+	if readers[0] == readers[1] {
 		t.Fatal("Readers() returned the same queue object twice")
 	}
-	// Distinct backing buffers (the actual regression: shared readBuf).
-	if &q0.readBuf[0] == &q1.readBuf[0] {
-		t.Fatal("queues share the same readBuf backing array")
-	}
-	// Shared underlying pipes.
-	if q0.outboundReader != q1.outboundReader || q0.inboundWriter != q1.inboundWriter {
-		t.Fatal("queues do not share the underlying pipes")
-	}
 
 	// Drive one packet through each queue and confirm the borrowed bytes from
 	// the first read are NOT clobbered by the second read. With a shared