nebula/udp/udp_sendmsgx_darwin_test.go

//go:build !e2e_testing

package udp

import (
	"fmt"
	"log/slog"
	"net"
	"net/netip"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

// recvAll reads n datagrams from pc, or fails the test after a second without one.
func recvAll(t *testing.T, pc net.PacketConn, n int) []string {
	t.Helper()
	var got []string
	buf := make([]byte, 64)
	for len(got) < n {
		require.NoError(t, pc.SetReadDeadline(time.Now().Add(time.Second)))
		l, _, err := pc.ReadFrom(buf)
		require.NoError(t, err, "after %d of %d datagrams", len(got), n)
		got = append(got, string(buf[:l]))
	}
	return got
}

// A batch longer than sendBatch reaches each peer whole and in order through sendmsg_x, without falling
// back to sendto: a run to one peer that spans two calls, then datagrams alternating between peers.
func TestWriteBatchSendsEachToItsPeer(t *testing.T) {
	for _, tc := range []struct{ name, listen string }{
		{"v4", "127.0.0.1"},
		{"v6", "::1"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			c, err := NewListener(slog.New(slog.DiscardHandler), Settings{Listen: netip.AddrPortFrom(netip.MustParseAddr(tc.listen), 0)})
			require.NoError(t, err)
			defer c.Close()

			var peers [2]net.PacketConn
			var dsts [2]netip.AddrPort
			for i := range peers {
				peers[i], err = net.ListenPacket("udp", net.JoinHostPort(tc.listen, "0"))
				require.NoError(t, err)
				defer peers[i].Close()
				dsts[i] = netip.MustParseAddrPort(peers[i].LocalAddr().String())
			}

			const total = sendBatch*2 + 5
			bufs := make([][]byte, total)
			addrs := make([]netip.AddrPort, total)
			var want [2][]string
			for i := range bufs {
				bufs[i] = fmt.Appendf(nil, "pkt-%03d", i)
				p := i % 2
				if i < sendBatch+6 {
					p = 0
				}
				addrs[i] = dsts[p]
				want[p] = append(want[p], string(bufs[i]))
			}

			n, err := c.WriteBatch(bufs, addrs)
			require.NoError(t, err)
			assert.Equal(t, total, n)
			assert.False(t, c.(*StdConn).noSendmsgX.Load(), "fell back to sendto")
			for i := range peers {
				assert.Equal(t, want[i], recvAll(t, peers[i], len(want[i])))
			}
		})
	}
}

// Destinations that can't be sent to cost only their own packets: an IPv6 peer on an IPv4 socket is
// skipped before the call, and a broadcast address without SO_BROADCAST is refused by the kernel
// mid-batch. Neither stops the packets around it, delivers any twice, or turns sendmsg_x off.
func TestWriteBatchSkipsUnsendableDestinations(t *testing.T) {
	c, err := NewListener(slog.New(slog.DiscardHandler), Settings{Listen: netip.MustParseAddrPort("127.0.0.1:0")})
	require.NoError(t, err)
	defer c.Close()

	peer, err := net.ListenPacket("udp", "127.0.0.1:0")
	require.NoError(t, err)
	defer peer.Close()
	good := netip.MustParseAddrPort(peer.LocalAddr().String())
	bcast := netip.AddrPortFrom(netip.MustParseAddr("255.255.255.255"), good.Port())
	v6 := netip.MustParseAddrPort("[::1]:4242")

	addrs := []netip.AddrPort{good, bcast, bcast, good, v6, good, bcast}
	bufs := make([][]byte, len(addrs))
	for i := range bufs {
		bufs[i] = fmt.Appendf(nil, "pkt-%d", i)
	}

	n, err := c.WriteBatch(bufs, addrs)
	require.NoError(t, err)
	assert.Equal(t, 3, n)
	assert.False(t, c.(*StdConn).noSendmsgX.Load(), "fell back to sendto")
	assert.Equal(t, []string{"pkt-0", "pkt-3", "pkt-5"}, recvAll(t, peer, 3))
	require.NoError(t, peer.SetReadDeadline(time.Now().Add(100*time.Millisecond)))
	_, _, err = peer.ReadFrom(make([]byte, 64))
	assert.Error(t, err, "a datagram arrived twice")
}

// A datagram the kernel refuses as its own fault (EMSGSIZE here, as ENOBUFS would be under a full
// interface queue) costs only itself, whether it fails after others in the same call (a short count) or
// first (the errno alone): the rest of its run still arrives, in order, and sendmsg_x stays on.
func TestWriteBatchLosesOnlyTheFailedDatagram(t *testing.T) {
	for _, tc := range []struct {
		name string
		big  int
	}{
		{"mid-call", 2},
		{"first", 0},
	} {
		t.Run(tc.name, func(t *testing.T) {
			c, err := NewListener(slog.New(slog.DiscardHandler), Settings{Listen: netip.MustParseAddrPort("127.0.0.1:0")})
			require.NoError(t, err)
			defer c.Close()

			peer, err := net.ListenPacket("udp", "127.0.0.1:0")
			require.NoError(t, err)
			defer peer.Close()
			dst := netip.MustParseAddrPort(peer.LocalAddr().String())

			bufs := make([][]byte, 5)
			addrs := make([]netip.AddrPort, len(bufs))
			var want []string
			for i := range bufs {
				addrs[i] = dst
				if i == tc.big {
					// Past the 65535-byte IP datagram limit, so the kernel can only refuse it.
					bufs[i] = make([]byte, 70000)
					continue
				}
				bufs[i] = fmt.Appendf(nil, "pkt-%d", i)
				want = append(want, string(bufs[i]))
			}

			n, err := c.WriteBatch(bufs, addrs)
			require.NoError(t, err)
			assert.Equal(t, len(want), n)
			assert.False(t, c.(*StdConn).noSendmsgX.Load(), "fell back to sendto")
			assert.Equal(t, want, recvAll(t, peer, len(want)))
		})
	}
}