tstest/natlab/vnet: deflake TestPacketSideEffects and TestProtocolQEMU

Both tests started flaking after my 910735448 ("tstest/natlab/vnet:
send unsolicited IPv6 Router Advertisements") added background RA
traffic on v6-enabled networks.

TestPacketSideEffects races the periodic unsolicited-RA goroutine
against its synchronous packet-count assertions: when the multicast
RA fires after the test has registered its sinks, both sinks receive
it and "got 1 packet, want N" becomes "got N+2".

TestProtocolQEMU's reader was doing raw Read on the SOCK_STREAM unix
socket and comparing the whole result to the expected length-prefixed
packet. The kernel is free to coalesce the on-register RA frame and
the test packet into one Read, in which case bytes.Equal fails and
the entire chunk (including the test packet's bytes) gets discarded
as "unexpected", leading to a 5s i/o timeout. Parse the QEMU uint32
length-prefix framing with io.ReadFull instead so we read exactly one
frame per iteration regardless of how the kernel buffers them. The
SOCK_DGRAM path (TestProtocolUnixDgram) keeps the original raw Read
since datagram boundaries are preserved.

These where the top two flakes in oss on the flakes dashboards.

Updates #13038

Signed-off-by: Brad Fitzpatrick <bradfitz@tailscale.com>
Change-Id: I32983656b692921a0f43a4a5e9a8a6ab2555ee49

tstest/natlab/vnet/vnet.go

@@ -637,6 +637,12 @@ type network struct {
 
 	blackholeMu  sync.Mutex
 	blackholeMap map[netip.Addr]netip.Addr // blackholeMap contains address pairs for dropping traffic (in either direction)
+
+	// raStopMu guards raStopped and serializes with the unsolicited RA
+	// goroutine's send so that StopUnsolicitedRAsForTest can deterministically
+	// silence the background traffic.
+	raStopMu  sync.Mutex
+	raStopped bool
 }
 
 // registerWriter registers a client address with a MAC address.
@@ -1142,6 +1148,18 @@ func (s *Server) MACs() iter.Seq[MAC] {
 	return maps.Keys(s.nodeByMAC)
 }
 
+// StopUnsolicitedRAsForTest stops all networks from sending periodic
+// unsolicited IPv6 Router Advertisements. It blocks until any in-progress
+// send has finished, so callers may safely register sinks afterwards
+// without races against background RA traffic.
+func (s *Server) StopUnsolicitedRAsForTest() {
+	for n := range s.networks {
+		n.raStopMu.Lock()
+		n.raStopped = true
+		n.raStopMu.Unlock()
+	}
+}
+
 func (s *Server) RegisterSinkForTest(mac MAC, fn func(eth []byte)) {
 	n, ok := s.nodeByMAC[mac]
 	if !ok {
@@ -2075,6 +2093,15 @@ func (n *network) handleIPv6RouterSolicitation(ep EthernetPacket, _ *layers.ICMP
 func (n *network) startUnsolicitedRAs() {
 	n.s.wg.Go(func() {
 		send := func() {
+			// Hold raStopMu across the writeEth so that
+			// StopUnsolicitedRAsForTest can synchronize with any
+			// in-progress send: once StopUnsolicitedRAsForTest returns,
+			// no further unsolicited RAs will be delivered to writers.
+			n.raStopMu.Lock()
+			defer n.raStopMu.Unlock()
+			if n.raStopped {
+				return
+			}
 			pkt, err := n.buildIPv6RouterAdvertisement(macAllNodes, ipv6AllNodes)
 			if err != nil {
 				n.logf("building unsolicited RA: %v", err)

tstest/natlab/vnet/vnet_test.go

@@ -8,6 +8,7 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"io"
 	"net"
 	"net/netip"
 	"path/filepath"
@@ -194,6 +195,10 @@ func TestPacketSideEffects(t *testing.T) {
 			if err != nil {
 				t.Fatal(err)
 			}
+			// Silence the periodic unsolicited Router Advertisements
+			// before registering sinks; otherwise a background RA can
+			// race with the synchronous packet-count assertions below.
+			s.StopUnsolicitedRAsForTest()
 			defer s.Close()
 
 			for _, tt := range tt.tests {
@@ -558,7 +563,7 @@ func TestProtocolQEMU(t *testing.T) {
 		go s.ServeUnixConn(conn.(*net.UnixConn), ProtocolQEMU)
 	}
 
-	sendBetweenClients(t, clientc, s, mkLenPrefixed)
+	sendBetweenClients(t, clientc, s, ProtocolQEMU)
 }
 
 // TestProtocolUnixDgram tests the protocol that macOS Virtualization.framework
@@ -601,11 +606,11 @@ func TestProtocolUnixDgram(t *testing.T) {
 		clientc[i] = c
 	}
 
-	sendBetweenClients(t, clientc, s, nil)
+	sendBetweenClients(t, clientc, s, ProtocolUnixDGRAM)
 }
 
 // sendBetweenClients is a test helper that tries to send an ethernet frame from
-// one client to another.
+// one client to another using the given vnet wire protocol.
 //
 // It first makes the two clients send a packet to a fictitious node 3, which
 // forces their src MACs to be registered with a networkWriter internally so
@@ -615,12 +620,11 @@ func TestProtocolUnixDgram(t *testing.T) {
 // effect here, so this does it manually.
 //
 // It also then waits for them to be registered.
-//
-// wrap is an optional function that wraps the packet before sending it.
-func sendBetweenClients(t testing.TB, clientc [2]*net.UnixConn, s *Server, wrap func([]byte) []byte) {
+func sendBetweenClients(t testing.TB, clientc [2]*net.UnixConn, s *Server, proto Protocol) {
 	t.Helper()
-	if wrap == nil {
-		wrap = func(b []byte) []byte { return b }
+	wrap := func(b []byte) []byte { return b }
+	if proto == ProtocolQEMU {
+		wrap = mkLenPrefixed
 	}
 	for i, c := range clientc {
 		must.Get(c.Write(wrap(mkEth(nodeMac(3), nodeMac(i+1), testingEthertype, []byte("hello")))))
@@ -637,18 +641,41 @@ func sendBetweenClients(t testing.TB, clientc [2]*net.UnixConn, s *Server, wrap
 	t.Logf("writing % 02x", pkt)
 	must.Get(clientc[0].Write(pkt))
 
-	// vnet sends an unsolicited Router Advertisement at writer-register time
-	// on v6-enabled networks; loop until we see the test packet, skipping any
-	// noise that arrived first.
-	buf := make([]byte, 2048)
+	// vnet sends a Router Advertisement at writer-register time on v6-enabled
+	// networks (and may also send periodic unsolicited RAs). Loop until we
+	// see the test packet, skipping any noise that arrived first.
+	//
+	// For the QEMU stream protocol the kernel is free to coalesce multiple
+	// length-prefixed frames into one Read, so the reader must parse the
+	// framing or it may swallow the test packet alongside an RA.
 	deadline := time.Now().Add(5 * time.Second)
+	clientc[1].SetReadDeadline(deadline)
+	readFrame := func() ([]byte, error) {
+		if proto == ProtocolUnixDGRAM {
+			buf := make([]byte, 2048)
+			n, err := clientc[1].Read(buf)
+			if err != nil {
+				return nil, err
+			}
+			return buf[:n], nil
+		}
+		var hdr [4]byte
+		if _, err := io.ReadFull(clientc[1], hdr[:]); err != nil {
+			return nil, err
+		}
+		n := binary.BigEndian.Uint32(hdr[:])
+		frame := make([]byte, 4+n)
+		copy(frame, hdr[:])
+		if _, err := io.ReadFull(clientc[1], frame[4:]); err != nil {
+			return nil, err
+		}
+		return frame, nil
+	}
 	for {
-		clientc[1].SetReadDeadline(deadline)
-		n, err := clientc[1].Read(buf)
+		got, err := readFrame()
 		if err != nil {
 			t.Fatalf("did not receive test packet: %v", err)
 		}
-		got := buf[:n]
 		if bytes.Equal(got, pkt) {
 			return
 		}