net/tstun,wgengine/netstack: avoid deadlocks on full outbound queues

Make link queue writes nonblocking: gVisor can generate replies on the
reader needed to drain the queue. A full queue drops only the rejected
packet and reports ErrNoBufferSpace, which gVisor counts and TCP handles
by arming its retransmit timer. UDP sockets do not see the error unless
IP_RECVERR is set, matching Linux. Count WireGuard-bound drops as
reason="queue_full" in tailscaled_outbound_dropped_packets_total, and
per-queue drops in debug and client metrics.

Add Wrapper.TryInjectOutbound for best-effort replies generated while
processing packets (TSMP rejects and pongs, conn25 rejects, fake-mode
echo replies), so those paths drop instead of stalling behind the
injection queue's reader. Its drops are also counted as queue_full.

Make outbound injection waits cancellable and close link queues before
abort-generated replies. Stop the WireGuard injection pump on ErrClosed.
Release injected packets still queued at Close, outside closeOnce to
allow reentrant release callbacks; a sender whose send races with Close
releases the packet itself. Release temporary capture buffers.

Copy injected payload views directly into the WireGuard slab, avoiding
an intermediate allocation and copy.

Test queue saturation, multiqueue progress, cancellation, concurrent
shutdown, reference ownership, fragmented payloads, and GSO.

Performance versus the parent commit, on a 4-core Xeon 6975P-C:

  inject+read 64B / 1280B       245ns / 479ns -> 187ns / 195ns, 0 allocs
  4 injectors, 1 reader         843ns -> 451ns
  64KiB GSO inject+split        10.9us -> 4.4us, 64KiB/op -> 0
  link queue write+read         81.4ns -> 81.5ns
  tsnet TCP, 1 / 8 streams      448 / 580 MB/s -> 492 / 674 MB/s

BenchmarkTCPQueueBottleneck shows that dropping on full queues keeps
throughput at the bottleneck rate (117 MB/s at 100k packets/s, as with
blocking writes), at the cost of more retransmits with many streams.

Fixes #21631
Updates #21638
Updates #21632
Updates #21639

Signed-off-by: James Tucker <james@tailscale.com>

cmd/tailscaled/netstack.go

@@ -32,6 +32,7 @@ func newNetstack(logf logger.Logf, sys *tsd.System, onlyNetstack bool) (tsd.Nets
 	if err != nil {
 		return nil, err
 	}
+	ns.SetMetricsRegistry(sys.UserMetricsRegistry())
 	// Only register debug info if we have a debug mux
 	if debugMux != nil {
 		expvar.Publish("netstack", ns.ExpVar())

cmd/tsconnect/wasm/wasm_js.go

@@ -123,6 +123,7 @@ func newIPN(jsConfig js.Value) map[string]any {
 	if err != nil {
 		log.Fatalf("netstack.Create: %v", err)
 	}
+	ns.SetMetricsRegistry(sys.UserMetricsRegistry())
 	sys.Set(ns)
 	ns.ProcessLocalIPs = true
 	ns.ProcessSubnets = true

feature/conn25/datapath.go

@@ -166,7 +166,7 @@ func (dh *datapathHandler) HandlePacketFromWireGuard(p *packet.Parsed, tun *tstu
 				Proto:  p.IPProto,
 				Reason: packet.RejectedDueToUnknownAppConnectorTransitIP,
 			}
-			if err := tun.InjectOutbound(packet.Generate(rj, nil)); err != nil {
+			if err := tun.TryInjectOutbound(packet.Generate(rj, nil)); err != nil && !errors.Is(err, tstun.ErrClosed) {
 				dh.debugLogf("error sending TSMP flow rejection packet: %v", err)
 			}
 			return filter.Drop

net/tstun/netstack_disabled.go

@@ -9,8 +9,20 @@ type netstack_PacketBuffer struct {
 	GSOOptions netstack_GSO
 }
 
-func (*netstack_PacketBuffer) DecRef()   { panic("unreachable") }
-func (*netstack_PacketBuffer) Size() int { panic("unreachable") }
+func (*netstack_PacketBuffer) DecRef()         { panic("unreachable") }
+func (*netstack_PacketBuffer) Size() int       { panic("unreachable") }
+func (*netstack_PacketBuffer) HeaderSize() int { panic("unreachable") }
+func (*netstack_PacketBuffer) AsViewList() (netstack_ViewList, int) {
+	panic("unreachable")
+}
+
+type netstack_ViewList struct{}
+type netstack_View struct{}
+
+func (netstack_ViewList) Front() *netstack_View { panic("unreachable") }
+func (*netstack_View) Next() *netstack_View     { panic("unreachable") }
+func (*netstack_View) Size() int                { panic("unreachable") }
+func (*netstack_View) AsSlice() []byte          { panic("unreachable") }
 
 type netstack_GSOType int
 
@@ -67,3 +79,4 @@ func (s slicer) Slice() []byte { panic("unreachable") }
 type netstack_Buffer struct{}
 
 func (netstack_Buffer) Flatten() []byte { panic("unreachable") }
+func (netstack_Buffer) Release()        { panic("unreachable") }

net/tstun/wrap.go

@@ -4,6 +4,7 @@
 package tstun
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -67,6 +68,10 @@ var (
 	ErrClosed = errors.New("device closed")
 	// ErrFiltered is returned when the acted-on packet is rejected by a filter.
 	ErrFiltered = errors.New("packet dropped by filter")
+
+	// errInjectionQueueFull is returned by TryInjectOutbound when the packet
+	// cannot be queued immediately.
+	errInjectionQueueFull = errors.New("injection queue full")
 )
 
 var (
@@ -260,6 +265,7 @@ type injectionQueue struct {
 
 	// ch carries injected packets to the reader.
 	// ch is never closed, senders and the reader select on [Wrapper.closed].
+	// Packets left in ch after Close are released by [Wrapper.drainInjected].
 	ch chan tunInjectedRead
 }
 
@@ -381,9 +387,26 @@ func (t *Wrapper) Close() error {
 			t.tunDevStatsCloser.Close()
 		}
 	})
+	// Drain outside closeOnce: packet release callbacks may reenter Close.
+	t.drainInjected()
 	return err
 }
 
+// drainInjected releases packets queued for the injection queue's reader
+// without waiting for more.
+func (t *Wrapper) drainInjected() {
+	for {
+		select {
+		case r := <-t.injectionQueue.ch:
+			if r.packet != nil {
+				r.packet.DecRef()
+			}
+		default:
+			return
+		}
+	}
+}
+
 // isClosed reports whether t is closed.
 func (t *Wrapper) isClosed() bool {
 	select {
@@ -473,16 +496,59 @@ func (t *Wrapper) InjectionQueue() tun.Queue {
 	return t.injectionQueue
 }
 
-// injectOutbound hands r to the injection queue's reader, or releases it if
-// the Wrapper is closed.
-func (t *Wrapper) injectOutbound(r tunInjectedRead) {
-	select {
-	case t.injectionQueue.ch <- r:
-	case <-t.closed:
-		if r.packet != nil {
+// injectOutbound transfers r to the injection queue or releases it on error.
+func (t *Wrapper) injectOutbound(ctx context.Context, r tunInjectedRead) (err error) {
+	defer func() {
+		if err != nil && r.packet != nil {
 			r.packet.DecRef()
 		}
+	}()
+	if err := ctx.Err(); err != nil {
+		return err
 	}
+	if t.isClosed() {
+		return ErrClosed
+	}
+	// Avoid a cancellation select when the queue has space.
+	select {
+	case t.injectionQueue.ch <- r:
+		t.drainIfClosed()
+		return nil
+	default:
+	}
+	select {
+	case t.injectionQueue.ch <- r:
+		t.drainIfClosed()
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-t.closed:
+		return ErrClosed
+	}
+}
+
+// drainIfClosed is called after a send to the injection queue. A send can race
+// with Close; if Close's drain may have missed it, release it here.
+func (t *Wrapper) drainIfClosed() {
+	if t.isClosed() {
+		t.drainInjected()
+	}
+}
+
+// tryInjectOutbound queues data for the injection queue's reader without
+// waiting. It drops data if the queue is full.
+func (t *Wrapper) tryInjectOutbound(data []byte) error {
+	if t.isClosed() {
+		return ErrClosed
+	}
+	select {
+	case t.injectionQueue.ch <- tunInjectedRead{data: data}:
+		t.drainIfClosed()
+		return nil
+	default:
+	}
+	t.metrics.outboundDroppedPacketsTotal.Add(usermetric.DropLabels{Reason: usermetric.ReasonQueueFull}, 1)
+	return errInjectionQueueFull
 }
 
 // snat does SNAT on p if the destination address requires a different source address.
@@ -978,7 +1044,19 @@ func (t *Wrapper) injectedRead(res tunInjectedRead, slab []byte, packets []tun.R
 		}
 		bufN := copy(pkt, res.packet.NetworkHeader().Slice())
 		bufN += copy(pkt[bufN:], res.packet.TransportHeader().Slice())
-		bufN += copy(pkt[bufN:], res.packet.Data().AsRange().ToSlice())
+		// Copy payload views directly to avoid allocating a flattened copy.
+		views, offset := res.packet.AsViewList()
+		// AsViewList's offset points at the first header; skip all headers
+		// (including link/virtio headers) to reach the start of Data.
+		offset += res.packet.HeaderSize()
+		for v := views.Front(); v != nil; v = v.Next() {
+			if offset >= v.Size() {
+				offset -= v.Size()
+				continue
+			}
+			bufN += copy(pkt[bufN:], v.AsSlice()[offset:])
+			offset = 0
+		}
 		gso = res.packet.GSOOptions
 		pkt = pkt[:bufN]
 		defer res.packet.DecRef() // defer DecRef so we may continue to reference it
@@ -1173,8 +1251,7 @@ func (t *Wrapper) filterPacketInboundFromWireGuard(p *packet.Parsed, captHook pa
 			if filt.ShieldsUp() {
 				rj.Reason = packet.RejectedDueToShieldsUp
 			}
-			pkt := packet.Generate(rj, nil)
-			t.InjectOutbound(pkt)
+			t.TryInjectOutbound(packet.Generate(rj, nil))
 
 			// TODO(bradfitz): also send a TCP RST, after the TSMP message.
 
@@ -1435,30 +1512,57 @@ func (t *Wrapper) injectOutboundPong(pp *packet.Parsed, req packet.TSMPPingReque
 		return
 	}
 
-	t.InjectOutbound(packet.Generate(pong, nil))
+	t.TryInjectOutbound(packet.Generate(pong, nil))
 }
 
 // InjectOutbound makes the Wrapper device behave as if a packet
 // with the given contents was sent to the network.
-// It does not block, but takes ownership of the packet.
+// It blocks until the packet can be queued or the Wrapper closes, and takes
+// ownership of the packet. Only reads from InjectionQueue can make space.
 // The injected packet will not pass through outbound filter rules,
 // but UDP/SCTP flow state is recorded so inbound replies are admitted.
 // Injecting an empty packet is a no-op.
 func (t *Wrapper) InjectOutbound(pkt []byte) error {
+	return t.InjectOutboundContext(context.Background(), pkt)
+}
+
+// TryInjectOutbound is InjectOutbound, but drops pkt rather than waiting for
+// queue space. Use it for best-effort replies generated while processing
+// packets, which must not stall behind the injection queue's reader.
+// Dropped packets are counted as queue_full outbound drops.
+func (t *Wrapper) TryInjectOutbound(pkt []byte) error {
 	if len(pkt) > MaxPacketSize {
 		return errPacketTooBig
 	}
 	if len(pkt) == 0 {
 		return nil
 	}
-	t.injectOutbound(tunInjectedRead{data: pkt})
-	return nil
+	return t.tryInjectOutbound(pkt)
 }
 
-// InjectOutboundPacketBuffer logically behaves as InjectOutbound. It takes ownership of one
-// reference count on the packet, and the packet may be mutated. The packet refcount will be
-// decremented after the injected buffer has been read.
+// InjectOutboundContext is InjectOutbound with a cancellable wait for queue
+// space. Cancellation does not retract a packet that has already been queued.
+func (t *Wrapper) InjectOutboundContext(ctx context.Context, pkt []byte) error {
+	if len(pkt) > MaxPacketSize {
+		return errPacketTooBig
+	}
+	if len(pkt) == 0 {
+		return nil
+	}
+	return t.injectOutbound(ctx, tunInjectedRead{data: pkt})
+}
+
+// InjectOutboundPacketBuffer is InjectOutbound for a PacketBuffer. It takes
+// ownership of one reference even on error. The packet may be mutated, and its
+// reference is released after InjectionQueue consumes it or the Wrapper discards it.
 func (t *Wrapper) InjectOutboundPacketBuffer(pkt *netstack_PacketBuffer) error {
+	return t.InjectOutboundPacketBufferContext(context.Background(), pkt)
+}
+
+// InjectOutboundPacketBufferContext is InjectOutboundPacketBuffer with a
+// cancellable wait for queue space. Cancellation does not retract a packet that
+// has already been queued. It takes ownership of one reference even on error.
+func (t *Wrapper) InjectOutboundPacketBufferContext(ctx context.Context, pkt *netstack_PacketBuffer) error {
 	if !buildfeatures.HasNetstack {
 		panic("unreachable")
 	}
@@ -1474,10 +1578,10 @@ func (t *Wrapper) InjectOutboundPacketBuffer(pkt *netstack_PacketBuffer) error {
 	if capt := t.captureHook.Load(); capt != nil {
 		b := pkt.ToBuffer()
 		capt(packet.SynthesizedToPeer, t.now(), b.Flatten(), packet.CaptureMeta{})
+		b.Release()
 	}
 
-	t.injectOutbound(tunInjectedRead{packet: pkt})
-	return nil
+	return t.injectOutbound(ctx, tunInjectedRead{packet: pkt})
 }
 
 func (t *Wrapper) BatchSize() int {

net/tstun/wrap_injection_close_test.go

@@ -0,0 +1,28 @@
+// Copyright (c) Tailscale Inc & contributors
+// SPDX-License-Identifier: BSD-3-Clause
+
+//go:build !ts_omit_netstack
+
+package tstun
+
+import (
+	"sync/atomic"
+	"testing"
+
+	"tailscale.com/util/eventbus/eventbustest"
+)
+
+// A sender can win its select against closed and enqueue after Close has
+// drained the injection queue. The sender must then release the packet.
+func TestInjectionSendAfterCloseDrain(t *testing.T) {
+	_, w := newFakeTUN(t.Logf, eventbustest.NewBus(t), false)
+	if err := w.Close(); err != nil {
+		t.Fatal(err)
+	}
+	var released atomic.Int32
+	w.injectionQueue.ch <- tunInjectedRead{packet: newInjectionTestPacket([]byte{1}, &released)}
+	w.drainIfClosed()
+	if got := released.Load(); got != 1 {
+		t.Fatalf("released %d times, want 1", got)
+	}
+}

net/tstun/wrap_injection_test.go

@@ -0,0 +1,270 @@
+// Copyright (c) Tailscale Inc & contributors
+// SPDX-License-Identifier: BSD-3-Clause
+
+//go:build !ts_omit_netstack
+
+package tstun
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"expvar"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"gvisor.dev/gvisor/pkg/buffer"
+	"gvisor.dev/gvisor/pkg/tcpip/stack"
+	"tailscale.com/net/packet"
+	"tailscale.com/util/eventbus/eventbustest"
+	"tailscale.com/util/usermetric"
+)
+
+func newInjectionTestPacket(data []byte, released *atomic.Int32) *stack.PacketBuffer {
+	return stack.NewPacketBuffer(stack.PacketBufferOptions{
+		Payload: buffer.MakeWithData(data), OnRelease: func() { released.Add(1) },
+	})
+}
+
+func TestInjectOutboundPacketBufferOwnership(t *testing.T) {
+	for _, state := range []string{"read", "buffered-close", "closed", "canceled", "empty", "oversize", "captured"} {
+		t.Run(state, func(t *testing.T) {
+			bus := eventbustest.NewBus(t)
+			_, w := newFakeTUN(t.Logf, bus, false)
+			t.Cleanup(func() { w.Close() })
+			ctx := context.Background()
+			data := udp4("1.2.3.4", "5.6.7.8", 98, 98)
+			wantErr := error(nil)
+			switch state {
+			case "closed":
+				w.Close()
+				wantErr = ErrClosed
+			case "canceled":
+				var cancel context.CancelFunc
+				ctx, cancel = context.WithCancel(ctx)
+				cancel()
+				wantErr = context.Canceled
+			case "empty":
+				data = nil
+			case "oversize":
+				data = make([]byte, MaxPacketSize+1)
+				wantErr = errPacketTooBig
+			}
+			if state == "captured" {
+				w.InstallCaptureHook(func(path packet.CapturePath, _ time.Time, got []byte, _ packet.CaptureMeta) {
+					if path != packet.SynthesizedToPeer || !bytes.Equal(got, data) {
+						t.Errorf("capture = %v, %x", path, got)
+					}
+				})
+			}
+			var released atomic.Int32
+			pkt := newInjectionTestPacket(data, &released)
+			err := w.InjectOutboundPacketBufferContext(ctx, pkt)
+			if !errors.Is(err, wantErr) {
+				t.Fatalf("InjectOutboundPacketBufferContext = %v, want %v", err, wantErr)
+			}
+			if state == "read" || state == "buffered-close" || state == "captured" {
+				if got := released.Load(); got != 0 {
+					t.Fatalf("released %d packets before consuming injection, want 0", got)
+				}
+				if state == "buffered-close" {
+					w.Close()
+				} else {
+					w.Start()
+					slab, packets := getSinglePacketReadArgs()
+					n, err := w.InjectionQueue().Read(slab, packets)
+					if n != 1 || err != nil {
+						t.Fatalf("Read = %d, %v", n, err)
+					}
+					if got := slab[packets[0].Offset : packets[0].Offset+packets[0].Size]; !bytes.Equal(got, data) {
+						t.Errorf("Read = %x, want %x", got, data)
+					}
+				}
+			}
+			if got := released.Load(); got != 1 {
+				t.Errorf("released %d packets, want 1", got)
+			}
+		})
+	}
+}
+
+// Cancel a sender waiting behind another sender on a full queue.
+func TestInjectOutboundCancellationBehindBlockedSender(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		bus := eventbustest.NewBus(t)
+		_, w := newFakeTUN(t.Logf, bus, false)
+		t.Cleanup(func() { w.Close() })
+		var released atomic.Int32
+		data := udp4("1.2.3.4", "5.6.7.8", 98, 98)
+		if err := w.InjectOutboundPacketBuffer(newInjectionTestPacket(data, &released)); err != nil {
+			t.Fatal(err)
+		}
+		blocked := make(chan error, 1)
+		go func() { blocked <- w.InjectOutboundPacketBuffer(newInjectionTestPacket(data, &released)) }()
+		synctest.Wait()
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		canceled := make(chan error, 1)
+		go func() { canceled <- w.InjectOutboundPacketBufferContext(ctx, newInjectionTestPacket(data, &released)) }()
+		synctest.Wait()
+		cancel()
+		synctest.Wait()
+		select {
+		case err := <-canceled:
+			if !errors.Is(err, context.Canceled) {
+				t.Fatalf("injection = %v, want context.Canceled", err)
+			}
+		default:
+			t.Fatal("canceled sender is stuck behind a background sender")
+		}
+		if got := released.Load(); got != 1 {
+			t.Fatalf("released %d packets after cancellation, want 1", got)
+		}
+		select {
+		case err := <-blocked:
+			t.Fatalf("background sender returned before Wrapper.Close: %v", err)
+		default:
+		}
+		w.Close()
+		if err := <-blocked; !errors.Is(err, ErrClosed) {
+			t.Fatalf("background injection = %v, want ErrClosed", err)
+		}
+		if got := released.Load(); got != 3 {
+			t.Errorf("released %d packets after Close, want 3", got)
+		}
+	})
+}
+
+func TestInjectOutboundPacketBufferCancellationWithFullQueue(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		bus := eventbustest.NewBus(t)
+		_, w := newFakeTUN(t.Logf, bus, false)
+		t.Cleanup(func() { w.Close() })
+		data := udp4("1.2.3.4", "5.6.7.8", 98, 98)
+		var released atomic.Int32
+		if err := w.InjectOutboundPacketBuffer(newInjectionTestPacket(data, &released)); err != nil {
+			t.Fatal(err)
+		}
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		done := make(chan error, 1)
+		go func() {
+			done <- w.InjectOutboundPacketBufferContext(ctx, newInjectionTestPacket(data, &released))
+		}()
+		// Cancel while the sender waits for queue space.
+		synctest.Wait()
+		cancel()
+		if err := <-done; !errors.Is(err, context.Canceled) {
+			t.Fatalf("injection = %v, want context.Canceled", err)
+		}
+		if got := released.Load(); got != 1 {
+			t.Fatalf("released %d packets after cancellation, want 1", got)
+		}
+		w.Start()
+		slab, packets := getSinglePacketReadArgs()
+		if n, err := w.InjectionQueue().Read(slab, packets); n != 1 || err != nil {
+			t.Fatalf("Read accepted packet = %d, %v", n, err)
+		}
+		if got := released.Load(); got != 2 {
+			t.Errorf("released %d packets after Read, want 2", got)
+		}
+	})
+}
+
+// Every injected packet is released exactly once, whether it is read or
+// drained by a concurrent Close.
+func TestInjectOutboundConcurrentReadAndClose(t *testing.T) {
+	w := startedWrapper(t, 1)
+	t.Cleanup(func() { w.Close() })
+	const writers, count = 8, 64
+	var released atomic.Int32
+	data := udp4("1.2.3.4", "5.6.7.8", 98, 98)
+	// Start Close only after a reader has processed an accepted packet.
+	firstReleased := make(chan struct{})
+	first := stack.NewPacketBuffer(stack.PacketBufferOptions{
+		Payload: buffer.MakeWithData(data),
+		OnRelease: func() {
+			released.Add(1)
+			close(firstReleased)
+		},
+	})
+	if err := w.InjectOutboundPacketBuffer(first); err != nil {
+		t.Fatal(err)
+	}
+	var wg sync.WaitGroup
+	start := make(chan struct{})
+	for range writers {
+		wg.Go(func() {
+			<-start
+			for range count {
+				err := w.InjectOutboundPacketBuffer(newInjectionTestPacket(data, &released))
+				if err != nil && !errors.Is(err, ErrClosed) {
+					t.Errorf("injection: %v", err)
+				}
+			}
+		})
+	}
+	// Race injection readers with Close's drain.
+	for range 3 {
+		q := w.InjectionQueue()
+		wg.Go(func() {
+			<-start
+			slab, packets := getSinglePacketReadArgs()
+			for {
+				if _, err := q.Read(slab, packets); err != nil {
+					return
+				}
+			}
+		})
+	}
+	wg.Go(func() { <-start; <-firstReleased; w.Close() })
+	close(start)
+	wg.Wait()
+	if got := released.Load(); got != writers*count+1 {
+		t.Errorf("released %d packets, want %d", got, writers*count+1)
+	}
+}
+
+func BenchmarkInjectOutboundPacketBuffer(b *testing.B) {
+	bus := eventbustest.NewBus(b)
+	_, w := newFakeTUN(b.Logf, bus, false)
+	defer w.Close()
+	w.Start()
+	data := udp4("1.2.3.4", "5.6.7.8", 98, 98)
+	pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{Payload: buffer.MakeWithData(data)})
+	defer pkt.DecRef()
+	slab, packets := getSinglePacketReadArgs()
+	b.ReportAllocs()
+	for b.Loop() {
+		if err := w.InjectOutboundPacketBuffer(pkt.IncRef()); err != nil {
+			b.Fatal(err)
+		}
+		if _, err := w.InjectionQueue().Read(slab, packets); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func TestTryInjectOutbound(t *testing.T) {
+	w := startedWrapper(t, 1)
+	t.Cleanup(func() { w.Close() })
+	data := udp4("1.2.3.4", "5.6.7.8", 98, 98)
+	if err := w.TryInjectOutbound(data); err != nil {
+		t.Fatal(err)
+	}
+	// The queue is full and nothing reads it; this must drop, not block.
+	if err := w.TryInjectOutbound(data); err != errInjectionQueueFull {
+		t.Fatalf("TryInjectOutbound on full queue = %v, want errInjectionQueueFull", err)
+	}
+	m, _ := w.metrics.outboundDroppedPacketsTotal.Get(usermetric.DropLabels{Reason: usermetric.ReasonQueueFull}).(*expvar.Int)
+	if m == nil || m.Value() != 1 {
+		t.Errorf("queue_full drops = %v, want 1", m)
+	}
+	w.Close()
+	if err := w.TryInjectOutbound(data); err != ErrClosed {
+		t.Errorf("TryInjectOutbound after Close = %v, want ErrClosed", err)
+	}
+}

net/tstun/wrap_injection_views_test.go

@@ -0,0 +1,157 @@
+// Copyright (c) Tailscale Inc & contributors
+// SPDX-License-Identifier: BSD-3-Clause
+
+//go:build !ts_omit_netstack
+
+package tstun
+
+import (
+	"bytes"
+	"fmt"
+	"net/netip"
+	"testing"
+
+	"github.com/tailscale/wireguard-go/tun"
+	"gvisor.dev/gvisor/pkg/buffer"
+	"gvisor.dev/gvisor/pkg/tcpip"
+	"gvisor.dev/gvisor/pkg/tcpip/header"
+	"gvisor.dev/gvisor/pkg/tcpip/stack"
+	"tailscale.com/net/packet"
+	"tailscale.com/util/eventbus/eventbustest"
+)
+
+func injectionTestPayload(data []byte, viewSize int) buffer.Buffer {
+	var payload buffer.Buffer
+	for len(data) > 0 {
+		n := min(viewSize, len(data))
+		part := buffer.MakeWithData(data[:n])
+		payload.Merge(&part)
+		data = data[n:]
+	}
+	return payload
+}
+
+func TestInjectOutboundPacketBufferViews(t *testing.T) {
+	for _, layout := range []string{"raw", "pushed", "consumed"} {
+		for _, viewSize := range []int{7, 1280} {
+			t.Run(fmt.Sprintf("%s/view=%d", layout, viewSize), func(t *testing.T) {
+				h := &packet.UDP4Header{
+					IP4Header: packet.IP4Header{Src: netip.MustParseAddr("1.2.3.4"), Dst: netip.MustParseAddr("5.6.7.8")},
+					SrcPort:   98, DstPort: 98,
+				}
+				data := packet.Generate(h, bytes.Repeat([]byte{0x55}, 1280-h.Len()))
+				body := data
+				opts := stack.PacketBufferOptions{}
+				if layout == "pushed" {
+					body = data[28:]
+					// Leave unused space before the pushed headers, so the
+					// start of Data is not simply HeaderSize.
+					opts.ReserveHeaderBytes = 128
+				}
+				opts.Payload = injectionTestPayload(body, viewSize)
+				pkt := stack.NewPacketBuffer(opts)
+				defer pkt.DecRef()
+				switch layout {
+				case "consumed":
+					// Header consumption can merge views when headers cross
+					// their boundaries; the payload may remain fragmented.
+					if _, ok := pkt.NetworkHeader().Consume(20); !ok {
+						t.Fatal("consume network header")
+					}
+					if _, ok := pkt.TransportHeader().Consume(8); !ok {
+						t.Fatal("consume transport header")
+					}
+				case "pushed":
+					// Link and virtio headers precede the network header but
+					// are not part of the injected packet.
+					copy(pkt.TransportHeader().Push(8), data[20:28])
+					copy(pkt.NetworkHeader().Push(20), data[:20])
+					copy(pkt.LinkHeader().Push(14), bytes.Repeat([]byte{0xaa}, 14))
+					copy(pkt.VirtioNetHeader().Push(10), bytes.Repeat([]byte{0xbb}, 10))
+				}
+				original := pkt.Data().AsRange().ToSlice()
+				_, w := newFakeTUN(t.Logf, eventbustest.NewBus(t), false)
+				defer w.Close()
+				w.Start()
+				// Reuse the packet to detect accidental consumption or
+				// mutation of its backing views while copying.
+				for range 2 {
+					if err := w.InjectOutboundPacketBuffer(pkt.IncRef()); err != nil {
+						t.Fatal(err)
+					}
+					slab, packets := getSinglePacketReadArgs()
+					n, err := w.InjectionQueue().Read(slab, packets)
+					if n != 1 || err != nil {
+						t.Fatalf("Read = %d, %v", n, err)
+					}
+					got := slab[packets[0].Offset : packets[0].Offset+packets[0].Size]
+					if !bytes.Equal(got, data) {
+						t.Fatalf("Read = %x, want %x", got, data)
+					}
+					if !bytes.Equal(pkt.Data().AsRange().ToSlice(), original) {
+						t.Fatal("copy modified the packet payload")
+					}
+				}
+			})
+		}
+	}
+}
+
+func TestInjectOutboundPacketBufferFragmentedGSO(t *testing.T) {
+	const payloadLen, mss = 384, 64
+	data := make([]byte, 40+payloadLen)
+	src, dst := tcpip.AddrFrom4([4]byte{1, 2, 3, 4}), tcpip.AddrFrom4([4]byte{5, 6, 7, 8})
+	ih := header.IPv4(data[:20])
+	ih.Encode(&header.IPv4Fields{
+		TotalLength: uint16(len(data)), TTL: 64, Protocol: uint8(header.TCPProtocolNumber), SrcAddr: src, DstAddr: dst,
+	})
+	ih.SetChecksum(^ih.CalculateChecksum())
+	th := header.TCP(data[20:40])
+	th.Encode(&header.TCPFields{
+		SrcPort: 1234, DstPort: 5678, SeqNum: 42, AckNum: 1, DataOffset: 20, Flags: header.TCPFlagAck | header.TCPFlagPsh,
+	})
+	th.SetChecksum(^tun.PseudoHeaderChecksum(uint8(header.TCPProtocolNumber), src.AsSlice(), dst.AsSlice(), payloadLen+20))
+	for i := range data[40:] {
+		data[40+i] = byte(i)
+	}
+	pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{
+		ReserveHeaderBytes: 128,
+		Payload:            injectionTestPayload(data[40:], 7),
+	})
+	defer pkt.DecRef()
+	copy(pkt.TransportHeader().Push(20), data[20:40])
+	copy(pkt.NetworkHeader().Push(20), data[:20])
+	pkt.GSOOptions = stack.GSO{Type: stack.GSOTCPv4, NeedsCsum: true, CsumOffset: 16, MSS: mss, L3HdrLen: 20}
+	gso, err := stackGSOToTunGSO(data, pkt.GSOOptions)
+	if err != nil {
+		t.Fatal(err)
+	}
+	wantSlab := make([]byte, MaxPacketSize+2*tun.ReadPacketSpacing)
+	wantPackets := make([]tun.ReadPacket, payloadLen/mss)
+	wantN, err := tun.GSOSplit(data, gso, wantSlab, wantPackets, tun.ReadPacketSpacing)
+	if err != nil || wantN != len(wantPackets) {
+		t.Fatalf("reference GSOSplit = %d, %v", wantN, err)
+	}
+
+	_, w := newFakeTUN(t.Logf, eventbustest.NewBus(t), false)
+	defer w.Close()
+	w.Start()
+	if err := w.InjectOutboundPacketBuffer(pkt.IncRef()); err != nil {
+		t.Fatal(err)
+	}
+	slab := make([]byte, len(wantSlab))
+	packets := make([]tun.ReadPacket, len(wantPackets))
+	n, err := w.InjectionQueue().Read(slab, packets)
+	if err != nil || n != wantN {
+		t.Fatalf("Read = %d, %v; want %d, nil", n, err, wantN)
+	}
+	for i, p := range packets[:n] {
+		want := wantPackets[i]
+		if !bytes.Equal(slab[p.Offset:p.Offset+p.Size], wantSlab[want.Offset:want.Offset+want.Size]) {
+			t.Errorf("segment %d differs from reference GSOSplit", i)
+		}
+	}
+	if !bytes.Equal(pkt.Data().AsRange().ToSlice(), data[40:]) {
+		t.Error("GSO copy modified the packet payload")
+	}
+}

net/tstun/wrap_multiqueue_injection_test.go

@@ -0,0 +1,117 @@
+// Copyright (c) Tailscale Inc & contributors
+// SPDX-License-Identifier: BSD-3-Clause
+
+//go:build !ts_omit_netstack
+
+package tstun
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"os"
+	"sync/atomic"
+	"testing"
+	"testing/synctest"
+
+	"github.com/tailscale/wireguard-go/tun"
+	"tailscale.com/util/eventbus/eventbustest"
+	"tailscale.com/util/usermetric"
+)
+
+// injectionTestQueue is a native TUN queue used by injection tests.
+type injectionTestQueue struct {
+	closed  <-chan struct{}
+	packets chan []byte
+}
+
+func (*injectionTestQueue) File() *os.File { return nil }
+
+func (q *injectionTestQueue) Read(slab []byte, packets []tun.ReadPacket) (int, error) {
+	select {
+	case <-q.closed:
+		return 0, io.EOF
+	case p := <-q.packets:
+		copy(slab[tun.ReadPacketSpacing:], p)
+		packets[0] = tun.ReadPacket{Offset: tun.ReadPacketSpacing, Size: len(p)}
+		return 1, nil
+	}
+}
+
+func newMultiQueueInjectionTestWrapper(t *testing.T) (*fakeTUN, *Wrapper) {
+	t.Helper()
+	d := newFakeMQ(4)
+	for i := range d.queues {
+		d.queues[i] = &injectionTestQueue{closed: d.closechan, packets: make(chan []byte, 1)}
+	}
+	w := Wrap(t.Logf, d, new(usermetric.Registry), eventbustest.NewBus(t))
+	w.disableFilter = true
+	w.Start()
+	t.Cleanup(func() { w.Close() })
+	return d, w
+}
+
+func TestMultiQueueOutboundInjectionProgress(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		d, w := newMultiQueueInjectionTestWrapper(t)
+		queues := w.Queues()[:len(d.queues)]
+		data := udp4("100.64.1.2", "100.64.1.3", 1234, 5678)
+		var released atomic.Int32
+		if err := w.InjectOutboundPacketBuffer(newInjectionTestPacket(data, &released)); err != nil {
+			t.Fatal(err)
+		}
+		blocked := make(chan error, 1)
+		go func() { blocked <- w.InjectOutboundPacketBuffer(newInjectionTestPacket(data, &released)) }()
+		synctest.Wait()
+
+		// The injection queue has no reader yet. Its full handoff and blocked
+		// sender must not stop reads/filtering on any native queue.
+		read := make(chan int, len(queues))
+		closed := make(chan error, len(queues))
+		for i := range queues {
+			go func() {
+				slab, packets := getSinglePacketReadArgs()
+				n, err := queues[i].Read(slab, packets)
+				if n != 1 || err != nil || !bytes.Equal(slab[packets[0].Offset:packets[0].Offset+packets[0].Size], data) {
+					t.Errorf("queue %d read = %d, %v", i, n, err)
+				}
+				read <- i
+				_, err = queues[i].Read(slab, packets)
+				closed <- err
+			}()
+			d.queues[i].(*injectionTestQueue).packets <- data
+		}
+		synctest.Wait()
+		if got := len(read); got != len(queues) {
+			t.Fatalf("%d native queues made progress, want %d", got, len(queues))
+		}
+		select {
+		case err := <-blocked:
+			t.Fatalf("full injection handoff unexpectedly unblocked: %v", err)
+		default:
+		}
+		if got := released.Load(); got != 0 {
+			t.Fatalf("released = %d, want 0", got)
+		}
+
+		slab, packets := getSinglePacketReadArgs()
+		for range 2 {
+			n, err := w.InjectionQueue().Read(slab, packets)
+			if n != 1 || err != nil || !bytes.Equal(slab[packets[0].Offset:packets[0].Offset+packets[0].Size], data) {
+				t.Fatalf("injection queue read = %d, %v", n, err)
+			}
+		}
+		if err := <-blocked; err != nil {
+			t.Fatal(err)
+		}
+		if got := released.Load(); got != 2 {
+			t.Fatalf("released = %d, want 2", got)
+		}
+		w.Close()
+		for range len(queues) {
+			if err := <-closed; !errors.Is(err, io.EOF) {
+				t.Errorf("native read after Close = %v, want EOF", err)
+			}
+		}
+	})
+}

net/tstun/wrap_test.go

@@ -1569,14 +1569,14 @@ func TestInjectAfterClose(t *testing.T) {
 		w := startedWrapper(t, 1)
 		w.Close()
 
-		if err := w.InjectOutbound(udp4("1.2.3.4", "5.6.7.8", 1, 2)); err != nil {
-			t.Errorf("InjectOutbound after Close: %v", err)
+		if err := w.InjectOutbound(udp4("1.2.3.4", "5.6.7.8", 1, 2)); err != ErrClosed {
+			t.Errorf("InjectOutbound after Close = %v, want ErrClosed", err)
 		}
 		pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{
 			Payload: buffer.MakeWithData(udp4("1.2.3.4", "5.6.7.8", 1, 2)),
 		})
-		if err := w.InjectOutboundPacketBuffer(pkt); err != nil {
-			t.Errorf("InjectOutboundPacketBuffer after Close: %v", err)
+		if err := w.InjectOutboundPacketBuffer(pkt); err != ErrClosed {
+			t.Errorf("InjectOutboundPacketBuffer after Close = %v, want ErrClosed", err)
 		}
 	})
 }

tsnet/tsnet.go

@@ -908,6 +908,7 @@ func (s *Server) start() (reterr error) {
 	if err != nil {
 		return fmt.Errorf("netstack.Create: %w", err)
 	}
+	ns.SetMetricsRegistry(sys.UserMetricsRegistry())
 	sys.Tun.Get().Start()
 	sys.Set(ns)
 	if s.Tun == nil {

util/usermetric/metrics.go

@@ -46,6 +46,10 @@ const (
 
 	// ReasonError means that the packet was dropped because of an error.
 	ReasonError DropReason = "error"
+
+	// ReasonQueueFull means that the packet was dropped because an internal
+	// packet queue was full.
+	ReasonQueueFull DropReason = "queue_full"
 )
 
 // DropLabels contains common label(s) for dropped packet counters.

wgengine/netstack/link_endpoint.go

@@ -5,8 +5,10 @@ package netstack
 
 import (
 	"context"
+	"expvar"
 	"fmt"
 	"sync"
+	"sync/atomic"
 
 	"gvisor.dev/gvisor/pkg/buffer"
 	"gvisor.dev/gvisor/pkg/tcpip"
@@ -15,6 +17,7 @@ import (
 	"tailscale.com/feature/buildfeatures"
 	"tailscale.com/net/packet"
 	"tailscale.com/types/ipproto"
+	"tailscale.com/util/usermetric"
 	"tailscale.com/wgengine/netstack/gro"
 )
 
@@ -73,6 +76,12 @@ func (q *queue) Write(pkt *stack.PacketBuffer) tcpip.Error {
 	case <-q.closedCh:
 		pkt.DecRef()
 		return &tcpip.ErrClosedForSend{}
+	default:
+		// Packet processing can synchronously generate a reply on the TUN
+		// reader or loopback consumer. Neither may wait for queue space:
+		// draining the queue can depend on that same goroutine.
+		pkt.DecRef() // balance the IncRef evaluated before selecting a case
+		return &tcpip.ErrNoBufferSpace{}
 	}
 }
 
@@ -139,6 +148,24 @@ type linkEndpoint struct {
 	mtu        uint32
 
 	outboundQueues [outboundQueueLimit]*queue // outbound
+
+	// queueFullDropped counts, per [outboundQueue], packets dropped by
+	// WritePackets because that queue was full. A GSO packet counts as one.
+	queueFullDropped [outboundQueueLimit]expvar.Int
+
+	// outboundDropped, if set, is the user metric for packets dropped while
+	// being sent to peers. It is shared with tstun, so it is added to rather
+	// than replaced with queueFullDropped.
+	outboundDropped atomic.Pointer[usermetric.MultiLabelMap[usermetric.DropLabels]]
+}
+
+// queueFullDroppedTotal returns the sum of queueFullDropped.
+func (ep *linkEndpoint) queueFullDroppedTotal() int64 {
+	var total int64
+	for i := range ep.queueFullDropped {
+		total += ep.queueFullDropped[i].Value()
+	}
+	return total
 }
 
 // newLinkEndpoint constructs a [*linkEndpoint]. size is applied independently
@@ -356,9 +383,12 @@ func (ep *linkEndpoint) SetLinkAddress(addr tcpip.LinkAddress) {
 }
 
 // WritePackets routes outbound packets into the appropriate outbound [queue].
-// Multiple concurrent calls are permitted.
+// Multiple concurrent calls are permitted. Packets whose queue is full are
+// dropped and excluded from the returned count; if any were dropped, the error
+// is [tcpip.ErrNoBufferSpace].
 func (ep *linkEndpoint) WritePackets(pkts stack.PacketBufferList) (int, tcpip.Error) {
 	n := 0
+	var full bool
 	// TODO(jwhited): evaluate writing a stack.PacketBufferList instead of a
 	//  single packet. We can split 2 x 64K GSO across
 	//  wireguard-go/conn.IdealBatchSize (128 slots) @ 1280 MTU, and non-GSO we
@@ -376,14 +406,28 @@ func (ep *linkEndpoint) WritePackets(pkts stack.PacketBufferList) (int, tcpip.Er
 			panic(fmt.Sprintf("linkEndpoint.outboundQueueRouter returned %v which is outside outboundQueueLimit(%v)", q, outboundQueueLimit))
 		}
 		if err := ep.outboundQueues[q].Write(pkt); err != nil {
-			if _, ok := err.(*tcpip.ErrNoBufferSpace); !ok && n == 0 {
+			if _, ok := err.(*tcpip.ErrNoBufferSpace); ok {
+				// Drop only this packet. Later packets may be routed to
+				// queues that have space.
+				ep.queueFullDropped[q].Add(1)
+				if q == outboundToWireGuard {
+					if m := ep.outboundDropped.Load(); m != nil {
+						m.Add(usermetric.DropLabels{Reason: usermetric.ReasonQueueFull}, 1)
+					}
+				}
+				full = true
+				continue
+			}
+			if n == 0 {
 				return 0, err
 			}
 			break
 		}
 		n++
 	}
-
+	if full {
+		return n, &tcpip.ErrNoBufferSpace{}
+	}
 	return n, nil
 }
 

wgengine/netstack/link_endpoint_test.go

@@ -0,0 +1,112 @@
+// Copyright (c) Tailscale Inc & contributors
+// SPDX-License-Identifier: BSD-3-Clause
+
+package netstack
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"gvisor.dev/gvisor/pkg/tcpip"
+	"gvisor.dev/gvisor/pkg/tcpip/stack"
+)
+
+func newTestQueue(t testing.TB, size int) *queue {
+	t.Helper()
+	q := &queue{c: make(chan *stack.PacketBuffer, size), closedCh: make(chan struct{})}
+	t.Cleanup(func() { q.Close(); q.Drain() })
+	return q
+}
+
+func newQueueTestPacket(released *atomic.Int32) *stack.PacketBuffer {
+	return stack.NewPacketBuffer(stack.PacketBufferOptions{OnRelease: func() { released.Add(1) }})
+}
+
+func TestQueueWriteFull(t *testing.T) {
+	q := newTestQueue(t, 1)
+	var released atomic.Int32
+	for i, want := range []tcpip.Error{nil, &tcpip.ErrNoBufferSpace{}} {
+		pkt := newQueueTestPacket(&released)
+		err := q.Write(pkt) // must not block
+		pkt.DecRef()
+		if fmt.Sprintf("%T", err) != fmt.Sprintf("%T", want) {
+			t.Fatalf("Write %d = %v, want %v", i, err, want)
+		}
+	}
+	// Only the rejected packet is released; the queued one is retained.
+	if got := released.Load(); got != 1 {
+		t.Errorf("released %d packets, want 1", got)
+	}
+	if got := q.Num(); got != 1 {
+		t.Errorf("Num = %d, want 1", got)
+	}
+}
+
+func TestQueueWriteClosed(t *testing.T) {
+	for _, state := range []string{"closed", "close-signaled"} {
+		t.Run(state, func(t *testing.T) {
+			q := newTestQueue(t, 0)
+			if state == "closed" {
+				q.Close()
+			} else {
+				q.closeOnce.Do(func() { close(q.closedCh) })
+			}
+			var released atomic.Int32
+			pkt := newQueueTestPacket(&released)
+			err := q.Write(pkt)
+			if got := released.Load(); got != 0 {
+				t.Errorf("released %d packets before caller DecRef, want 0", got)
+			}
+			pkt.DecRef()
+			if _, ok := err.(*tcpip.ErrClosedForSend); !ok {
+				t.Errorf("Write = %v, want ErrClosedForSend", err)
+			}
+			if got := released.Load(); got != 1 {
+				t.Errorf("released %d packets, want 1", got)
+			}
+		})
+	}
+}
+
+func TestQueueConcurrentWriteReadAndClose(t *testing.T) {
+	q := newTestQueue(t, 8)
+	const writers, packets = 8, 64
+	var released atomic.Int32
+	var wg sync.WaitGroup
+	start := make(chan struct{})
+	for range writers {
+		wg.Go(func() {
+			<-start
+			for range packets {
+				pkt := newQueueTestPacket(&released)
+				err := q.Write(pkt)
+				pkt.DecRef()
+				switch err.(type) {
+				case nil, *tcpip.ErrNoBufferSpace, *tcpip.ErrClosedForSend:
+				default:
+					t.Errorf("Write: unexpected error %v", err)
+				}
+			}
+		})
+	}
+	wg.Go(func() {
+		<-start
+		for {
+			pkt := q.ReadContext(context.Background())
+			if pkt == nil {
+				return
+			}
+			pkt.DecRef()
+		}
+	})
+	wg.Go(func() { <-start; q.Close() })
+	close(start)
+	wg.Wait()
+	q.Drain()
+	if got := released.Load(); got != writers*packets {
+		t.Errorf("released %d packets, want %d", got, writers*packets)
+	}
+}

wgengine/netstack/multiqueue_liveness_test.go

@@ -0,0 +1,129 @@
+// Copyright (c) Tailscale Inc & contributors
+// SPDX-License-Identifier: BSD-3-Clause
+
+package netstack
+
+import (
+	"context"
+	"io"
+	"os"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/tailscale/wireguard-go/tun"
+	"gvisor.dev/gvisor/pkg/tcpip/header"
+	"gvisor.dev/gvisor/pkg/tcpip/stack"
+	"tailscale.com/net/tstun"
+	"tailscale.com/util/eventbus/eventbustest"
+	"tailscale.com/util/usermetric"
+)
+
+type queueTestTUN struct {
+	tun.Device
+	queues []tun.Queue
+	closed chan struct{}
+}
+
+func newQueueTestTUN(n int) *queueTestTUN {
+	d := &queueTestTUN{Device: tstun.NewFake(), closed: make(chan struct{})}
+	for range n {
+		d.queues = append(d.queues, &queueTestTUNQueue{closed: d.closed, packets: make(chan []byte, 1)})
+	}
+	return d
+}
+
+func (d *queueTestTUN) Queues() []tun.Queue { return d.queues }
+func (d *queueTestTUN) Read(slab []byte, packets []tun.ReadPacket) (int, error) {
+	return d.queues[0].Read(slab, packets)
+}
+func (d *queueTestTUN) WriteTo(_ int, bufs [][]byte, offset int) (int, error) {
+	return d.Write(bufs, offset)
+}
+func (d *queueTestTUN) Close() error {
+	close(d.closed)
+	return d.Device.Close()
+}
+
+type queueTestTUNQueue struct {
+	closed  <-chan struct{}
+	packets chan []byte
+}
+
+func (*queueTestTUNQueue) File() *os.File { return nil }
+func (q *queueTestTUNQueue) Read(slab []byte, packets []tun.ReadPacket) (int, error) {
+	select {
+	case <-q.closed:
+		return 0, io.EOF
+	case data := <-q.packets:
+		copy(slab[tun.ReadPacketSpacing:], data)
+		packets[0] = tun.ReadPacket{Offset: tun.ReadPacketSpacing, Size: len(data)}
+		return 1, nil
+	}
+}
+
+// Concurrent native TUN reads must not block on synchronous netstack replies
+// when the outbound queue is full.
+func TestMultiQueueNetstackReplyWithFullQueue(t *testing.T) {
+	const queues = 4
+	s, ep := newQueueTestStack(t, 1)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	ns := &Impl{ctx: ctx, ipstack: s, linkEP: ep, logf: t.Logf}
+	ns.ready.Store(true)
+	d := newQueueTestTUN(queues)
+	w := tstun.Wrap(t.Logf, d, new(usermetric.Registry), eventbustest.NewBus(t))
+	defer w.Close()
+	w.PreFilterPacketOutboundToWireGuardNetstackIntercept = ns.handleLocalPackets
+	w.Start()
+
+	var released atomic.Int32
+	filler := newQueueTestPacket(&released)
+	err := ep.outboundQueues[outboundToHost].Write(filler)
+	filler.DecRef()
+	if err != nil {
+		t.Fatal(err)
+	}
+	ack := queueTestACK(queueTestLocalIP, serviceIP)
+	th := header.TCP(ack.TransportHeader().Slice())
+	pseudo := tun.PseudoHeaderChecksum(uint8(header.TCPProtocolNumber), queueTestLocalIP.AsSlice(), serviceIP.AsSlice(), uint16(len(th)))
+	th.SetChecksum(^tun.Checksum(th, pseudo))
+	data := append([]byte(nil), stack.PayloadSince(ack.NetworkHeader()).AsSlice()...)
+	ack.DecRef()
+
+	var wg sync.WaitGroup
+	start := make(chan struct{})
+	for i, q := range w.Queues()[:queues] {
+		d.queues[i].(*queueTestTUNQueue).packets <- data
+		wg.Go(func() {
+			<-start
+			slab := make([]byte, tstun.MaxPacketSize+2*tun.ReadPacketSpacing)
+			n, err := q.Read(slab, make([]tun.ReadPacket, 1))
+			if n != 0 || err != nil {
+				t.Errorf("intercepted read = %d, %v; want 0, nil", n, err)
+			}
+		})
+	}
+	done := make(chan struct{})
+	go func() { wg.Wait(); close(done) }()
+	close(start)
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		ep.Close()
+		w.Close()
+		<-done
+		t.Fatal("a TUN queue blocked generating a reply into a full netstack queue")
+	}
+	if got := ep.queueFullDropped[outboundToHost].Value(); got != queues {
+		t.Fatalf("queue-full drops = %d, want %d", got, queues)
+	}
+	// Releasing capacity must restore delivery without canceling netstack.
+	ep.Read(outboundToHost).DecRef()
+	ep.DeliverLoopback(queueTestACK(queueTestLocalIP, serviceIP))
+	requireQueueTestReset(t, ep, outboundToHost)
+	if got := released.Load(); got != 1 {
+		t.Errorf("filler released %d times, want 1", got)
+	}
+}

wgengine/netstack/netstack.go

@@ -55,6 +55,7 @@ import (
 	"tailscale.com/types/views"
 	"tailscale.com/util/clientmetric"
 	"tailscale.com/util/set"
+	"tailscale.com/util/usermetric"
 	"tailscale.com/version"
 	"tailscale.com/wgengine"
 	"tailscale.com/wgengine/filter"
@@ -442,11 +443,20 @@ func Create(logf logger.Logf, tundev *tstun.Wrapper, e wgengine.Engine, mc *magi
 	return ns, nil
 }
 
+// SetMetricsRegistry registers netstack's user-facing metrics with r, which
+// should be the registry used by ns's [tstun.Wrapper].
+func (ns *Impl) SetMetricsRegistry(r *usermetric.Registry) {
+	// Only WireGuard-bound packets are sent to other peers. Host and loopback
+	// drops are node-local, so they are reported only in debug and client metrics.
+	ns.linkEP.outboundDropped.Store(r.DroppedPacketsOutbound())
+}
+
 func (ns *Impl) Close() error {
 	stacksForMetrics.Delete(ns)
-	// Cancel the injection goroutines before ipstack.Wait closes linkEP's
-	// outbound queues. A nil queue read is expected only after cancellation.
+	// Cancel before closing the queues so consumers expect nil reads.
+	// Close linkEP before aborting endpoints that can generate replies.
 	ns.ctxCancel()
+	ns.linkEP.Close()
 	ns.ipstack.Close()
 	ns.ipstack.Wait()
 	ns.injectWG.Wait()
@@ -469,6 +479,17 @@ func init() {
 	// Please take care to avoid exporting clientmetrics with the same metric
 	// names as the ones used by Impl.ExpVar. Both get exposed via the same HTTP
 	// endpoint, and name collisions will result in Prometheus scraping errors.
+	clientmetric.NewCounterFunc("netstack_outbound_queue_full_dropped_packets", func() int64 {
+		var total int64
+		for ns := range stacksForMetrics.Keys() {
+			v := ns.linkEP.queueFullDroppedTotal()
+			if v > math.MaxInt64-total {
+				return math.MaxInt64
+			}
+			total += v
+		}
+		return total
+	})
 	clientmetric.NewCounterFunc("netstack_tcp_forward_dropped_attempts", func() int64 {
 		var total uint64
 		for ns := range stacksForMetrics.Keys() {
@@ -1053,17 +1074,13 @@ func (ns *Impl) injectToWireGuard() {
 			ns.logf("[v2] injectToWireGuard: % x",
 				stack.PayloadSince(pkt.NetworkHeader()).AsSlice())
 		}
-		if err := ns.tundev.InjectOutboundPacketBuffer(pkt); err != nil {
-			ns.logf("netstack injectToWireGuard err: %v", err)
-			// When failing to inject an outbound packet buffer, log the error, but
-			// continue serving the ReadContext for sending subsequent packets, as
-			// nothing manages or restarts a failed injectToWireGuard. An error here
-			// only applies to the current packet and should not terminate the long-lived
-			// packet pump.
-			// The exception to this is if the context has ended, indicating a shutdown.
-			if ns.ctx.Err() != nil {
+		if err := ns.tundev.InjectOutboundPacketBufferContext(ns.ctx, pkt); err != nil {
+			if ns.ctx.Err() != nil || errors.Is(err, tstun.ErrClosed) {
 				return
 			}
+			ns.logf("netstack injectToWireGuard err: %v", err)
+			// A per-packet error must not terminate this long-lived pump:
+			// nothing manages or restarts a failed injectToWireGuard.
 			continue
 		}
 	}
@@ -1427,7 +1444,7 @@ func (ns *Impl) userPing(dstIP netip.Addr, pingResPkt []byte, direction userPing
 		ns.logf("exec pinged %v in %v", dstIP, time.Since(t0))
 	}
 	if direction == userPingDirectionOutbound {
-		if err := ns.tundev.InjectOutbound(pingResPkt); err != nil {
+		if err := ns.tundev.InjectOutboundContext(ns.ctx, pingResPkt); err != nil && ns.ctx.Err() == nil && !errors.Is(err, tstun.ErrClosed) {
 			ns.logf("InjectOutbound ping response: %v", err)
 		}
 	} else if direction == userPingDirectionInbound {
@@ -2253,6 +2270,14 @@ func readStatCounter(sc *tcpip.StatCounter) int64 {
 func (ns *Impl) ExpVar() expvar.Var {
 	m := new(metrics.Set)
 
+	for q, name := range map[outboundQueue]string{
+		outboundToWireGuard: "wireguard",
+		outboundToHost:      "host",
+		outboundLoopback:    "loopback",
+	} {
+		m.Set("counter_outbound_queue_full_dropped_packets_"+name, &ns.linkEP.queueFullDropped[q])
+	}
+
 	// Global metrics
 	stats := ns.ipstack.Stats()
 	m.Set("counter_dropped_packets", expvar.Func(func() any {

wgengine/netstack/queue_liveness_test.go

@@ -0,0 +1,479 @@
+// Copyright (c) Tailscale Inc & contributors
+// SPDX-License-Identifier: BSD-3-Clause
+
+package netstack
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/netip"
+	"sync/atomic"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"github.com/tailscale/wireguard-go/tun"
+	"gvisor.dev/gvisor/pkg/buffer"
+	"gvisor.dev/gvisor/pkg/tcpip"
+	"gvisor.dev/gvisor/pkg/tcpip/adapters/gonet"
+	"gvisor.dev/gvisor/pkg/tcpip/header"
+	"gvisor.dev/gvisor/pkg/tcpip/network/ipv4"
+	"gvisor.dev/gvisor/pkg/tcpip/stack"
+	"gvisor.dev/gvisor/pkg/tcpip/transport/tcp"
+	"gvisor.dev/gvisor/pkg/tcpip/transport/udp"
+	"tailscale.com/net/tstun"
+	"tailscale.com/tsd"
+)
+
+var queueTestLocalIP = netip.MustParseAddr("100.64.1.2")
+
+// newQueueTestStack uses the production destination classifier, but small
+// queues and no consumers so tests can deterministically force saturation.
+func newQueueTestStack(t testing.TB, size int) (*stack.Stack, *linkEndpoint) {
+	t.Helper()
+	ns := &Impl{}
+	ns.atomicIsLocalIPFunc.Store(func(a netip.Addr) bool { return a == queueTestLocalIP })
+	ns.atomicIsVIPServiceIPFunc.Store(func(netip.Addr) bool { return false })
+	ep := newLinkEndpoint(size, 1280, "", groNotSupported, ns.outboundQueueForPacket)
+	s := stack.New(stack.Options{
+		NetworkProtocols:   []stack.NetworkProtocolFactory{ipv4.NewProtocol},
+		TransportProtocols: []stack.TransportProtocolFactory{tcp.NewProtocol},
+	})
+	t.Cleanup(func() { ep.Close(); s.Close(); s.Wait() })
+	if err := s.CreateNIC(nicID, ep); err != nil {
+		t.Fatal(err)
+	}
+	for _, addr := range []netip.Addr{queueTestLocalIP, serviceIP} {
+		if err := s.AddProtocolAddress(nicID, tcpip.ProtocolAddress{
+			Protocol:          header.IPv4ProtocolNumber,
+			AddressWithPrefix: tcpip.AddrFrom4(addr.As4()).WithPrefix(),
+		}, stack.AddressProperties{}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	subnet, err := tcpip.NewSubnet(tcpip.AddrFrom4([4]byte{}), tcpip.MaskFromBytes([]byte{0, 0, 0, 0}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	s.SetRouteTable([]tcpip.Route{{Destination: subnet, NIC: nicID}})
+	return s, ep
+}
+
+// An ACK without an endpoint bypasses the TCP forwarder and generates a
+// synchronous reset in gVisor. DeliverLoopback marks the checksum validated.
+func queueTestACK(src, dst netip.Addr) *stack.PacketBuffer {
+	pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{ReserveHeaderBytes: 40})
+	header.TCP(pkt.TransportHeader().Push(20)).Encode(&header.TCPFields{
+		SrcPort: 12345, DstPort: 54321, DataOffset: 20, Flags: header.TCPFlagAck, AckNum: 1,
+	})
+	ih := header.IPv4(pkt.NetworkHeader().Push(20))
+	ih.Encode(&header.IPv4Fields{
+		TotalLength: 40, TTL: 64, Protocol: uint8(header.TCPProtocolNumber),
+		SrcAddr: tcpip.AddrFrom4(src.As4()), DstAddr: tcpip.AddrFrom4(dst.As4()),
+	})
+	ih.SetChecksum(^ih.CalculateChecksum())
+	pkt.NetworkProtocolNumber = header.IPv4ProtocolNumber
+	pkt.TransportProtocolNumber = header.TCPProtocolNumber
+	return pkt
+}
+
+func requireQueueTestReset(t *testing.T, ep *linkEndpoint, dest outboundQueue) {
+	t.Helper()
+	pkt := ep.Read(dest)
+	if pkt == nil {
+		t.Fatal("gVisor did not enqueue a reset")
+	}
+	defer pkt.DecRef()
+	if !header.TCP(pkt.TransportHeader().Slice()).Flags().Contains(header.TCPFlagRst) {
+		t.Fatal("reply is not a TCP reset")
+	}
+}
+
+func TestLoopbackReplyWithFullQueue(t *testing.T) {
+	_, ep := newQueueTestStack(t, 1)
+	// Model the consumer dequeuing a packet, then another producer refilling
+	// the queue before DeliverLoopback synchronously produces its reply.
+	ack := queueTestACK(queueTestLocalIP, queueTestLocalIP)
+	var pkts stack.PacketBufferList
+	pkts.PushBack(ack)
+	n, err := ep.WritePackets(pkts)
+	ack.DecRef()
+	if n != 1 || err != nil {
+		t.Fatalf("WritePackets = %d, %v", n, err)
+	}
+	delivering := ep.Read(outboundLoopback)
+	var released atomic.Int32
+	filler := newQueueTestPacket(&released)
+	err = ep.outboundQueues[outboundLoopback].Write(filler)
+	filler.DecRef()
+	if err != nil {
+		t.Fatal(err)
+	}
+	done := make(chan struct{})
+	go func() { ep.DeliverLoopback(delivering); close(done) }()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		ep.Close()
+		<-done
+		t.Fatal("loopback consumer blocked writing its reply into its own full queue")
+	}
+	if got := ep.Read(outboundLoopback); got != filler {
+		if got != nil {
+			got.DecRef()
+		}
+		t.Fatalf("Read = %p, want queued packet %p", got, filler)
+	} else {
+		got.DecRef()
+	}
+	if got := released.Load(); got != 1 {
+		t.Fatalf("queued packet released %d times, want 1", got)
+	}
+	// Recovery must happen before cancellation or closing the endpoint.
+	ep.DeliverLoopback(queueTestACK(queueTestLocalIP, queueTestLocalIP))
+	requireQueueTestReset(t, ep, outboundLoopback)
+}
+
+func TestCloseWithFullOutboundQueues(t *testing.T) {
+	ns := makeNetstack(t, func(ns *Impl) { ns.ProcessLocalIPs = true })
+	if err := ns.ipstack.AddProtocolAddress(nicID, tcpip.ProtocolAddress{
+		Protocol:          header.IPv4ProtocolNumber,
+		AddressWithPrefix: tcpip.AddrFrom4(queueTestLocalIP.As4()).WithPrefix(),
+	}, stack.AddressProperties{}); err != nil {
+		t.Fatal(err)
+	}
+	laddr := tcpip.FullAddress{NIC: nicID, Addr: tcpip.AddrFrom4(queueTestLocalIP.As4()), Port: 8080}
+	ln, err := gonet.ListenTCP(ns.ipstack, laddr, header.IPv4ProtocolNumber)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer ln.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	c, err := gonet.DialContextTCP(ctx, ns.ipstack, laddr, header.IPv4ProtocolNumber)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close()
+	sc, err := ln.Accept()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer sc.Close()
+	// With an established connection, stack shutdown must generate a reset.
+	// Stop all readers and fill all three queues before triggering that abort.
+	ns.ctxCancel()
+	ns.injectWG.Wait()
+	var released atomic.Int32
+	var queued int32
+	for _, q := range ns.linkEP.outboundQueues {
+		if q == nil {
+			continue
+		}
+		for range cap(q.c) {
+			pkt := newQueueTestPacket(&released)
+			err := q.Write(pkt)
+			pkt.DecRef()
+			if err != nil {
+				t.Fatal(err)
+			}
+			queued++
+		}
+	}
+	done := make(chan struct{})
+	go func() { ns.Close(); close(done) }()
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		ns.linkEP.Close()
+		<-done
+		t.Fatal("Close blocked with full outbound queues")
+	}
+	if got := released.Load(); got != queued {
+		t.Errorf("released %d queued packets, want %d", got, queued)
+	}
+}
+
+func TestCloseWithBlockedOutboundInjection(t *testing.T) {
+	for _, queues := range []int{1, 4} {
+		t.Run(fmt.Sprintf("queues=%d", queues), func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				testCloseWithBlockedOutboundInjection(t, queues)
+			})
+		})
+	}
+}
+
+func testCloseWithBlockedOutboundInjection(t *testing.T, queues int) {
+	// Leave InjectionQueue unread; netstack shutdown must cancel the sender
+	// without closing the shared Wrapper.
+	sys := tsd.NewSystem()
+	t.Cleanup(sys.Bus.Get().Close)
+	tw := tstun.Wrap(t.Logf, newQueueTestTUN(queues), sys.UserMetricsRegistry(), sys.Bus.Get())
+	t.Cleanup(func() { tw.Close() })
+	if err := tw.InjectOutbound([]byte{1}); err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	ns := &Impl{ctx: ctx, ctxCancel: cancel, tundev: tw, logf: t.Logf}
+	ns.atomicIsLocalIPFunc.Store(func(netip.Addr) bool { return false })
+	ns.atomicIsVIPServiceIPFunc.Store(func(netip.Addr) bool { return false })
+	ns.linkEP = newLinkEndpoint(1, 1280, "", groNotSupported, ns.outboundQueueForPacket)
+	// This test only runs the injection pump; no protocol workers are needed.
+	ns.ipstack = stack.New(stack.Options{})
+	if err := ns.ipstack.CreateNIC(nicID, ns.linkEP); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { tw.Close(); ns.Close() })
+	q := ns.linkEP.outboundQueues[outboundToWireGuard]
+	var released atomic.Int32
+	pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{
+		Payload: buffer.MakeWithData([]byte{1}), OnRelease: func() { released.Add(1) },
+	})
+	err := q.Write(pkt)
+	pkt.DecRef()
+	if err != nil {
+		t.Fatal(err)
+	}
+	tw.Start()
+	nativeReadDone := make(chan error, queues)
+	for _, q := range tw.Queues()[:queues] {
+		go func() {
+			slab := make([]byte, tstun.MaxPacketSize+2*tun.ReadPacketSpacing)
+			_, err := q.Read(slab, make([]tun.ReadPacket, 1))
+			nativeReadDone <- err
+		}()
+	}
+	ns.injectWG.Go(ns.injectToWireGuard)
+	// With InjectionQueue full and unread, wait until the pump has blocked
+	// in injection, not merely dequeued its packet.
+	synctest.Wait()
+	if q.Num() != 0 {
+		t.Fatal("injector has not dequeued packet")
+	}
+	if got := released.Load(); got != 0 {
+		t.Fatalf("released %d packets before shutdown, want 0", got)
+	}
+	closed := make(chan error, 1)
+	go func() { closed <- ns.Close() }()
+	synctest.Wait()
+	select {
+	case err := <-closed:
+		if err != nil {
+			t.Fatal(err)
+		}
+	default:
+		t.Fatal("netstack Close required closing the Wrapper to cancel injection")
+	}
+	if got := released.Load(); got != 1 {
+		t.Errorf("canceled packet released %d times, want 1", got)
+	}
+	// The shared Wrapper must remain usable after netstack stops injecting.
+	slab := make([]byte, tstun.MaxPacketSize+2*tun.ReadPacketSpacing)
+	if n, err := tw.InjectionQueue().Read(slab, make([]tun.ReadPacket, 1)); n != 1 || err != nil {
+		t.Fatalf("Read after netstack Close = %d, %v", n, err)
+	}
+	if err := tw.InjectOutbound([]byte{2}); err != nil {
+		t.Fatalf("InjectOutbound after netstack Close: %v", err)
+	}
+	select {
+	case err := <-nativeReadDone:
+		t.Fatalf("netstack Close terminated a native queue reader: %v", err)
+	default:
+	}
+	tw.Close()
+	for range queues {
+		if err := <-nativeReadDone; err != io.EOF {
+			t.Errorf("native read after Wrapper Close = %v, want EOF", err)
+		}
+	}
+}
+
+func TestInjectToWireGuardStopsOnClosedWrapper(t *testing.T) {
+	sys := tsd.NewSystem()
+	t.Cleanup(sys.Bus.Get().Close)
+	tw := tstun.Wrap(t.Logf, tstun.NewFake(), sys.UserMetricsRegistry(), sys.Bus.Get())
+	if err := tw.Close(); err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	ep := newLinkEndpoint(1, 1280, "", groNotSupported, nil)
+	defer ep.Close()
+	var released atomic.Int32
+	pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{
+		Payload: buffer.MakeWithData([]byte{1}), OnRelease: func() { released.Add(1) },
+	})
+	err := ep.outboundQueues[outboundToWireGuard].Write(pkt)
+	pkt.DecRef()
+	if err != nil {
+		t.Fatal(err)
+	}
+	ns := &Impl{ctx: ctx, tundev: tw, linkEP: ep, logf: t.Logf}
+	done := make(chan struct{})
+	go func() { ns.injectToWireGuard(); close(done) }()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		cancel()
+		<-done
+		t.Fatal("injectToWireGuard did not stop on a closed Wrapper")
+	}
+	if ctx.Err() != nil {
+		t.Fatal("netstack context should still be active")
+	}
+	if got := released.Load(); got != 1 {
+		t.Errorf("rejected packet released %d times, want 1", got)
+	}
+}
+
+// Queue-full drops are reported to gVisor, which counts them. UDP sockets do
+// not see ENOBUFS (as on Linux without IP_RECVERR), so forwarded UDP flows
+// survive full queues.
+func TestUDPWriteWithFullQueue(t *testing.T) {
+	ns := &Impl{}
+	ns.atomicIsLocalIPFunc.Store(func(a netip.Addr) bool { return a == queueTestLocalIP })
+	ns.atomicIsVIPServiceIPFunc.Store(func(netip.Addr) bool { return false })
+	ep := newLinkEndpoint(1, 1280, "", groNotSupported, ns.outboundQueueForPacket)
+	s := stack.New(stack.Options{
+		NetworkProtocols:   []stack.NetworkProtocolFactory{ipv4.NewProtocol},
+		TransportProtocols: []stack.TransportProtocolFactory{udp.NewProtocol},
+	})
+	t.Cleanup(func() { ep.Close(); s.Close(); s.Wait() })
+	if err := s.CreateNIC(nicID, ep); err != nil {
+		t.Fatal(err)
+	}
+	local := tcpip.AddrFrom4(queueTestLocalIP.As4())
+	if err := s.AddProtocolAddress(nicID, tcpip.ProtocolAddress{
+		Protocol: header.IPv4ProtocolNumber, AddressWithPrefix: local.WithPrefix(),
+	}, stack.AddressProperties{}); err != nil {
+		t.Fatal(err)
+	}
+	s.SetRouteTable([]tcpip.Route{{Destination: header.IPv4EmptySubnet, NIC: nicID}})
+	c, err := gonet.DialUDP(s,
+		&tcpip.FullAddress{NIC: nicID, Addr: local, Port: 1000},
+		&tcpip.FullAddress{NIC: nicID, Addr: tcpip.AddrFrom4([4]byte{100, 64, 9, 9}), Port: 2000},
+		header.IPv4ProtocolNumber)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close()
+	for i := range 2 {
+		if _, err := c.Write([]byte{byte(i)}); err != nil {
+			t.Fatalf("Write %d: %v", i, err)
+		}
+	}
+	if got := ep.queueFullDropped[outboundToWireGuard].Value(); got != 1 {
+		t.Errorf("queue-full drops = %d, want 1", got)
+	}
+	if got := s.Stats().NICs.TxPacketsDroppedNoBufferSpace.Value(); got != 1 {
+		t.Errorf("gVisor TxPacketsDroppedNoBufferSpace = %d, want 1", got)
+	}
+}
+
+// newBottleneckNode returns a stack with production TCP settings whose
+// non-local traffic is queued for WireGuard.
+func newBottleneckNode(b *testing.B, ip netip.Addr) (*stack.Stack, *linkEndpoint) {
+	ns := &Impl{}
+	ns.atomicIsLocalIPFunc.Store(func(a netip.Addr) bool { return a == ip })
+	ns.atomicIsVIPServiceIPFunc.Store(func(netip.Addr) bool { return false })
+	ep := newLinkEndpoint(512, 1280, "", groNotSupported, ns.outboundQueueForPacket)
+	s := stack.New(stack.Options{
+		NetworkProtocols:   []stack.NetworkProtocolFactory{ipv4.NewProtocol},
+		TransportProtocols: []stack.TransportProtocolFactory{tcp.NewProtocol},
+	})
+	b.Cleanup(func() { ep.Close(); s.Close(); s.Wait() })
+	cubic := tcpip.CongestionControlOption("cubic")
+	if err := s.SetTransportProtocolOption(tcp.ProtocolNumber, &cubic); err != nil {
+		b.Fatal(err)
+	}
+	if err := setTCPBufSizes(s); err != nil {
+		b.Fatal(err)
+	}
+	if err := s.CreateNIC(nicID, ep); err != nil {
+		b.Fatal(err)
+	}
+	if err := s.AddProtocolAddress(nicID, tcpip.ProtocolAddress{
+		Protocol: header.IPv4ProtocolNumber, AddressWithPrefix: tcpip.AddrFrom4(ip.As4()).WithPrefix(),
+	}, stack.AddressProperties{}); err != nil {
+		b.Fatal(err)
+	}
+	s.SetRouteTable([]tcpip.Route{{Destination: header.IPv4EmptySubnet, NIC: nicID}})
+	return s, ep
+}
+
+// pumpBottleneck moves packets from src's WireGuard queue into dst, at most
+// pps packets per second if pps > 0.
+func pumpBottleneck(ctx context.Context, src, dst *linkEndpoint, pps int) {
+	start := time.Now()
+	for n := 1; ; n++ {
+		pkt := src.ReadContext(ctx, outboundToWireGuard)
+		if pkt == nil {
+			return
+		}
+		dst.DeliverLoopback(pkt)
+		if pps > 0 && n%32 == 0 {
+			time.Sleep(time.Duration(n)*time.Second/time.Duration(pps) - time.Since(start))
+		}
+	}
+}
+
+type zeroReader struct{}
+
+func (zeroReader) Read(p []byte) (int, error) { clear(p); return len(p), nil }
+
+// BenchmarkTCPQueueBottleneck measures TCP throughput between two stacks when
+// the WireGuard-bound queue drains slower than TCP can send, so full queues
+// drop packets. Throughput should stay at the link rate.
+func BenchmarkTCPQueueBottleneck(b *testing.B) {
+	for _, pps := range []int{100_000, 0} {
+		for _, streams := range []int{1, 8} {
+			b.Run(fmt.Sprintf("pps=%d/streams=%d", pps, streams), func(b *testing.B) {
+				ipA, ipB := netip.MustParseAddr("100.64.1.1"), netip.MustParseAddr("100.64.1.2")
+				sA, epA := newBottleneckNode(b, ipA)
+				sB, epB := newBottleneckNode(b, ipB)
+				ctx, cancel := context.WithCancel(context.Background())
+				b.Cleanup(cancel)
+				go pumpBottleneck(ctx, epA, epB, pps)
+				go pumpBottleneck(ctx, epB, epA, 0)
+				addr := tcpip.FullAddress{NIC: nicID, Addr: tcpip.AddrFrom4(ipB.As4()), Port: 8080}
+				ln, err := gonet.ListenTCP(sB, addr, header.IPv4ProtocolNumber)
+				if err != nil {
+					b.Fatal(err)
+				}
+				defer ln.Close()
+				var conns, sconns []io.ReadWriteCloser
+				for range streams {
+					c, err := gonet.DialContextTCP(ctx, sA, addr, header.IPv4ProtocolNumber)
+					if err != nil {
+						b.Fatal(err)
+					}
+					defer c.Close()
+					sc, err := ln.Accept()
+					if err != nil {
+						b.Fatal(err)
+					}
+					defer sc.Close()
+					conns, sconns = append(conns, c), append(sconns, sc)
+				}
+				const chunk = 1 << 20
+				errc := make(chan error, 2*streams)
+				b.SetBytes(int64(streams * chunk))
+				for b.Loop() {
+					for i := range streams {
+						go func() { _, err := io.CopyN(conns[i], zeroReader{}, chunk); errc <- err }()
+						go func() { _, err := io.CopyN(io.Discard, sconns[i], chunk); errc <- err }()
+					}
+					for range 2 * streams {
+						if err := <-errc; err != nil {
+							b.Fatal(err)
+						}
+					}
+				}
+				b.ReportMetric(float64(sA.Stats().TCP.Retransmits.Value())/float64(b.N), "retransmits/op")
+				b.ReportMetric(float64(epA.queueFullDroppedTotal())/float64(b.N), "drops/op")
+			})
+		}
+	}
+}

wgengine/netstack/queue_metrics_test.go

@@ -0,0 +1,93 @@
+// Copyright (c) Tailscale Inc & contributors
+// SPDX-License-Identifier: BSD-3-Clause
+
+package netstack
+
+import (
+	"bytes"
+	"fmt"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+
+	"gvisor.dev/gvisor/pkg/tcpip"
+	"gvisor.dev/gvisor/pkg/tcpip/stack"
+	"tailscale.com/feature/buildfeatures"
+	"tailscale.com/util/clientmetric"
+	"tailscale.com/util/usermetric"
+)
+
+func TestOutboundQueueFullMetrics(t *testing.T) {
+	route := map[*stack.PacketBuffer]outboundQueue{}
+	ep := newLinkEndpoint(1, 1280, "", groNotSupported, func(pkt *stack.PacketBuffer) outboundQueue { return route[pkt] })
+	t.Cleanup(ep.Close)
+	ns := &Impl{linkEP: ep, ipstack: stack.New(stack.Options{})}
+	t.Cleanup(func() { ns.ipstack.Close(); ns.ipstack.Wait() })
+	stacksForMetrics.Store(ns, struct{}{})
+	t.Cleanup(func() { stacksForMetrics.Delete(ns) })
+	var reg usermetric.Registry
+	ns.SetMetricsRegistry(&reg)
+	debug := ns.ExpVar() // must reflect updates after registration
+
+	// With one slot per queue, the second WireGuard and loopback packets are
+	// dropped. A full queue must not drop later packets bound for other queues.
+	var released atomic.Int32
+	var pkts stack.PacketBufferList
+	for _, q := range []outboundQueue{outboundToWireGuard, outboundToWireGuard, outboundToHost, outboundLoopback, outboundLoopback} {
+		pkt := newQueueTestPacket(&released)
+		route[pkt] = q
+		pkts.PushBack(pkt)
+	}
+	n, err := ep.WritePackets(pkts)
+	pkts.DecRef()
+	if _, ok := err.(*tcpip.ErrNoBufferSpace); !ok || n != 3 {
+		t.Fatalf("WritePackets = %d, %v; want 3, ErrNoBufferSpace", n, err)
+	}
+	if got := released.Load(); got != 2 {
+		t.Errorf("released = %d, want 2", got)
+	}
+	want := map[string]int64{"wireguard": 1, "host": 0, "loopback": 1}
+	for q, name := range map[outboundQueue]string{outboundToWireGuard: "wireguard", outboundToHost: "host", outboundLoopback: "loopback"} {
+		if got := ep.queueFullDropped[q].Value(); got != want[name] {
+			t.Errorf("%s queue-full drops = %d, want %d", name, got, want[name])
+		}
+		if s := fmt.Sprintf("\"counter_outbound_queue_full_dropped_packets_%s\": %d", name, want[name]); !strings.Contains(debug.String(), s) {
+			t.Errorf("debug metrics missing %s: %s", s, debug.String())
+		}
+	}
+	if buildfeatures.HasUserMetrics {
+		rr := httptest.NewRecorder()
+		reg.Handler(rr, httptest.NewRequest("GET", "/metrics", nil))
+		if s := "tailscaled_outbound_dropped_packets_total{reason=\"queue_full\"} 1\n"; !strings.Contains(rr.Body.String(), s) {
+			t.Errorf("user metrics missing %q: %s", s, rr.Body.String())
+		}
+	}
+	if buildfeatures.HasClientMetrics {
+		// Client metrics sum all queues of all instances.
+		other := &Impl{linkEP: newLinkEndpoint(0, 1280, "", groNotSupported, nil), ipstack: ns.ipstack}
+		other.linkEP.queueFullDropped[outboundToHost].Add(2)
+		stacksForMetrics.Store(other, struct{}{})
+		t.Cleanup(func() { stacksForMetrics.Delete(other); other.linkEP.Close() })
+		var b bytes.Buffer
+		clientmetric.WritePrometheusExpositionFormat(&b)
+		if s := "netstack_outbound_queue_full_dropped_packets 4\n"; !strings.Contains(b.String(), s) {
+			t.Errorf("client metrics missing %q: %s", s, b.String())
+		}
+	}
+
+	// Shutdown loss is not queue-full loss.
+	ep.Close()
+	pkt := newQueueTestPacket(&released)
+	route[pkt] = outboundToWireGuard
+	var closedPkts stack.PacketBufferList
+	closedPkts.PushBack(pkt)
+	n, err = ep.WritePackets(closedPkts)
+	closedPkts.DecRef()
+	if _, ok := err.(*tcpip.ErrClosedForSend); !ok || n != 0 {
+		t.Errorf("closed WritePackets = %d, %v", n, err)
+	}
+	if got := ep.queueFullDroppedTotal(); got != 2 {
+		t.Errorf("closed write changed queue-full drops to %d, want 2", got)
+	}
+}

wgengine/userspace.go

@@ -633,7 +633,7 @@ func echoRespondToAll(p *packet.Parsed, t *tstun.Wrapper, gro *gro.GRO) (filter.
 		header := p.ICMP4Header()
 		header.ToResponse()
 		outp := packet.Generate(&header, p.Payload())
-		t.InjectOutbound(outp)
+		t.TryInjectOutbound(outp)
 		// We already responded to it, but it's not an error.
 		// Proceed with regular delivery. (Since this code is only
 		// used in fake mode, regular delivery just means throwing