more fixes!

Makefile

@@ -161,6 +161,10 @@ bin-pkcs11: BUILD_ARGS += -tags pkcs11
 bin-pkcs11: CGO_ENABLED = 1
 bin-pkcs11: bin
 
+# Build with the pprof debug server (serves on :6060). See startPprofServer.
+debug: BUILD_ARGS += -tags debug
+debug: bin
+
 bin:
 	go build $(BUILD_ARGS) -ldflags "$(LDFLAGS)" -o ./nebula${NEBULA_CMD_SUFFIX} ${NEBULA_CMD_PATH}
 	go build $(BUILD_ARGS) -ldflags "$(LDFLAGS)" -o ./nebula-cert${NEBULA_CMD_SUFFIX} ./cmd/nebula-cert
@@ -280,5 +284,5 @@ smoke-vagrant/%: bin-docker build/%/nebula
 	cd .github/workflows/smoke/ && ./smoke-vagrant.sh $*
 
 .FORCE:
-.PHONY: all all-linux all-freebsd all-openbsd all-netbsd all-darwin all-windows all-cross-linux all-cross-linux-arm all-cross-linux-mips all-cross-linux-other all-cross-darwin all-cross-windows bench bench-cpu bench-cpu-long bin build-test-mobile e2e e2ev e2evv e2evvv e2evvvv proto release service smoke-docker smoke-docker-race test test-cov-html smoke-vagrant/%
+.PHONY: all all-linux all-freebsd all-openbsd all-netbsd all-darwin all-windows all-cross-linux all-cross-linux-arm all-cross-linux-mips all-cross-linux-other all-cross-darwin all-cross-windows bench bench-cpu bench-cpu-long bin debug build-test-mobile e2e e2ev e2evv e2evvv e2evvvv proto release service smoke-docker smoke-docker-race test test-cov-html smoke-vagrant/%
 .DEFAULT_GOAL := bin

examples/config.yml

@@ -254,6 +254,20 @@ tun:
   # Default MTU for every packet, safe setting is (and the default) 1300 for internet based traffic
   mtu: 1300
 
+  # Linux only. pin_threads pins each tun reader/encrypt OS thread to a single CPU. This keeps every goroutine's
+  # batched sends flowing through one XPS-selected NIC TX ring, so packets within a flow stay ordered on the wire
+  # instead of being sprayed across multiple TX rings and reordered. Not reloadable.
+  #pin_threads: true
+
+  # Linux only. cpu_affinity overrides which CPUs the tun reader threads pin to: a list of CPU IDs, one per routine
+  # (see the top-level `routines` setting). Lists shorter than `routines` are modulo-cycled across the queues; extra
+  # entries are ignored. IDs must be within the process's allowed CPU set, so this respects taskset / cgroup cpusets;
+  # a non-integer or not-allowed entry disables the override and falls back to spreading queues across the allowed
+  # CPUs. Only meaningful while pin_threads is true. Not reloadable.
+  #cpu_affinity:
+  #  - 2
+  #  - 4
+
   # Route based MTU overrides, you have known vpn ip paths that can support larger MTUs you can increase/decrease them here
   routes:
     #- mtu: 8800
@@ -390,6 +404,16 @@ logging:
   # This setting is reloadable
   #inactivity_timeout: 10m
 
+  # ecn (default true) propagates ECN (Explicit Congestion Notification) across the tunnel per RFC 6040: the inner
+  # packet's ECN codepoint is copied onto the outer carrier header on encapsulation, and an outer CE ("congestion
+  # experienced") mark is folded back into the inner header on decapsulation. On linux it additionally stamps
+  # RTAX_FEATURE_ECN on the routes nebula installs, so the kernel actively negotiates ECN for connections to mesh
+  # prefixes. Disable this only when an underlay middlebox mangles or clears ECN bits unpredictably.
+  # This setting is reloadable, BUT flipping it at runtime only updates the datapath (the inner<->outer copy/combine).
+  # The RTAX_FEATURE_ECN flag on already-installed routes is NOT revisited on reload, so nebula must be restarted for
+  # the route half of this setting to take effect.
+  #ecn: true
+
 # Nebula security group configuration
 firewall:
   # Action to take when a packet is not allowed by the firewall rules.

interface.go

@@ -56,8 +56,13 @@ type InterfaceConfig struct {
 	// CpuAffinity, when non-empty, names the CPUs each TUN reader goroutine
 	// should pin to. Queue i pins to CpuAffinity[i % len(CpuAffinity)] —
 	// shorter lists than `routines` cycle. Empty list keeps the default
-	// pin-to-(i % NumCPU) behavior.
+	// pin-to-(i % NumCPU) behavior. Only consulted when PinThreads is true.
 	CpuAffinity []int
+	// PinThreads controls whether each TUN reader OS thread is pinned to a
+	// single CPU (via tun.pin_threads, default true). Pinning keeps each
+	// goroutine's sendmmsg on one XPS-selected NIC TX ring so per-flow
+	// packets stay ordered on the wire.
+	PinThreads bool
 
 	l *slog.Logger
 }
@@ -85,8 +90,13 @@ type Interface struct {
 	closed                atomic.Bool
 	// cpuAffinity, when non-empty, names the CPUs each TUN reader goroutine
 	// should pin to. Queue i pins to cpuAffinity[i % len(cpuAffinity)].
-	// Empty falls back to the default pin-to-(i % NumCPU) behavior.
+	// Empty falls back to the default pin-to-(allowed CPU) behavior.
+	// Only consulted when pinThreads is true.
 	cpuAffinity []int
+	// pinThreads controls whether listenIn pins each TUN reader OS thread to
+	// a CPU at all (tun.pin_threads, default true). When false, threads are
+	// left free to migrate as on stock nebula.
+	pinThreads bool
 	// ecnEnabled gates RFC 6040 underlay ECN propagation. When true,
 	// inside.go copies the inner ECN onto the outer carrier on encap and
 	// decryptToTun folds outer CE into the inner header on decap. Toggle
@@ -223,6 +233,7 @@ func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) {
 		connectionManager:     c.connectionManager,
 		conntrackCacheTimeout: c.ConntrackCacheTimeout,
 		cpuAffinity:           c.CpuAffinity,
+		pinThreads:            c.PinThreads,
 
 		metricHandshakes: metrics.GetOrRegisterHistogram("handshakes", nil, metrics.NewExpDecaySample(1028, 0.015)),
 		messageMetrics:   c.MessageMetrics,
@@ -383,13 +394,24 @@ func (f *Interface) listenOut(i int) {
 
 func (f *Interface) listenIn(reader tio.Queue, i int) {
 	// Pinning this thread (and goroutine) to a single CPU keeps every sendmmsg from this goroutine going through the
-	// same TX ring on the nic, so the wire sees per-flow order.
-	cpu := i % runtime.NumCPU()
-	if n := len(f.cpuAffinity); n > 0 {
-		cpu = f.cpuAffinity[i%n]
-	}
-	if err := util.PinThreadToCPU(cpu); err != nil {
-		f.l.Warn("failed to pin tun reader to CPU", "queue", i, "cpu", cpu, "err", err)
+	// same TX ring on the nic, so the wire sees per-flow order. Skip entirely when tun.pin_threads is false.
+	if f.pinThreads {
+		var cpu int
+		if n := len(f.cpuAffinity); n > 0 {
+			// Explicit tun.cpu_affinity list wins; parseCpuAffinity already
+			// validated the entries against the allowed CPU set.
+			cpu = f.cpuAffinity[i%n]
+		} else if allowed, err := util.AllowedCPUs(); err == nil && len(allowed) > 0 {
+			// Default: spread queues across the CPUs we're actually allowed to
+			// run on. Under a cpuset/taskset mask these aren't 0..NumCPU-1, so
+			// i % NumCPU would pick unrunnable IDs and every pin would fail.
+			cpu = allowed[i%len(allowed)]
+		} else {
+			cpu = i % runtime.NumCPU()
+		}
+		if err := util.PinThreadToCPU(cpu); err != nil {
+			f.l.Warn("failed to pin tun reader to CPU", "queue", i, "cpu", cpu, "err", err)
+		}
 	}
 
 	rejectBuf := make([]byte, mtu)
@@ -568,9 +590,12 @@ func (f *Interface) reloadEcn(c *config.C) {
 	initial := c.InitialLoad()
 	if initial || c.HasChanged("tunnels.ecn") {
 		v := c.GetBool("tunnels.ecn", true)
-		f.ecnEnabled.Store(v)
+		changed := f.ecnEnabled.Swap(v) != v
 		if !initial {
 			f.l.Info("tunnels.ecn changed", "enabled", v)
+			if changed {
+				f.l.Warn("tunnels.ecn datapath toggled, but route-level ECN negotiation (RTAX_FEATURE_ECN) retains its previous state until nebula is restarted", "enabled", v)
+			}
 		}
 	}
 }

main.go

@@ -5,11 +5,9 @@ import (
 	"fmt"
 	"log/slog"
 	"net"
-	"net/http"
-	_ "net/http/pprof"
 	"net/netip"
-	"runtime"
 	"runtime/debug"
+	"slices"
 	"strings"
 	"time"
 
@@ -36,17 +34,8 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev
 		buildVersion = moduleVersion()
 	}
 
-	//todo no merge
-	pprofServer := &http.Server{Addr: ":6060", Handler: nil}
-	go func() {
-		pprofServer.ListenAndServe()
-	}()
-
-	// Shut down the server when context is cancelled
-	go func() {
-		<-ctx.Done()
-		pprofServer.Shutdown(context.Background())
-	}()
+	// Debug builds (-tags debug) serve pprof on :6060; a no-op otherwise.
+	startPprofServer(ctx, l)
 
 	// Print the config if in test, the exit comes later
 	if configTest {
@@ -247,6 +236,7 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev
 		punchy:                punchy,
 		ConntrackCacheTimeout: conntrackCacheTimeout,
 		CpuAffinity:           parseCpuAffinity(c, l, routines),
+		PinThreads:            c.GetBool("tun.pin_threads", true),
 		l:                     l,
 	}
 
@@ -301,11 +291,16 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev
 
 // parseCpuAffinity reads `tun.cpu_affinity` from the config — a list of
 // integer CPU IDs, one per TUN reader goroutine. Empty / unset returns nil
-// (listenIn falls back to its default `i % NumCPU` pinning). Length
-// mismatch with `routines` is a warning, not an error: shorter lists are
-// modulo-cycled across queues, longer lists' tail is ignored. Invalid
-// entries (non-integer, out of range) are also a warning and disable the
-// override entirely so we don't silently pin to the wrong CPU.
+// (listenIn falls back to spreading queues across the allowed CPU set).
+// Length mismatch with `routines` is a warning, not an error: shorter lists
+// are modulo-cycled across queues, longer lists' tail is ignored. Invalid
+// entries (non-integer, or a CPU ID we're not allowed to run on) are also a
+// warning and disable the override entirely so we don't silently pin to the
+// wrong CPU. Entries are validated against the process's current affinity
+// mask (util.AllowedCPUs) rather than 0..NumCPU-1: under a cgroup cpuset or
+// taskset the runnable IDs are frequently not that contiguous range, and
+// pinning to an unrunnable ID always fails. If the allowed set can't be
+// determined we fall back to a plain non-negative check.
 func parseCpuAffinity(c *config.C, l *slog.Logger, routines int) []int {
 	raw := c.Get("tun.cpu_affinity")
 	if raw == nil {
@@ -316,7 +311,14 @@ func parseCpuAffinity(c *config.C, l *slog.Logger, routines int) []int {
 		l.Warn("tun.cpu_affinity must be a list of integers; ignoring", "value", raw)
 		return nil
 	}
-	nCPU := runtime.NumCPU()
+	// allowed is the set of CPU IDs we're actually permitted to run on. A nil
+	// slice (unsupported platform or lookup error) means "can't tell", so we
+	// only apply the weaker non-negative check in that case.
+	allowed, err := util.AllowedCPUs()
+	if err != nil {
+		l.Warn("could not determine allowed CPUs; validating tun.cpu_affinity against non-negative only", "error", err)
+		allowed = nil
+	}
 	cpus := make([]int, 0, len(rv))
 	for i, e := range rv {
 		var cpu int
@@ -332,9 +334,14 @@ func parseCpuAffinity(c *config.C, l *slog.Logger, routines int) []int {
 				"index", i, "value", e)
 			return nil
 		}
-		if cpu < 0 || cpu >= nCPU {
+		if cpu < 0 {
 			l.Warn("tun.cpu_affinity entry out of range; ignoring affinity",
-				"index", i, "cpu", cpu, "num_cpu", nCPU)
+				"index", i, "cpu", cpu)
+			return nil
+		}
+		if len(allowed) > 0 && !slices.Contains(allowed, cpu) {
+			l.Warn("tun.cpu_affinity entry not in allowed CPU set; ignoring affinity",
+				"index", i, "cpu", cpu, "allowed", allowed)
 			return nil
 		}
 		cpus = append(cpus, cpu)

overlay/batch/coalesce_core.go

@@ -93,8 +93,11 @@ func parseIPPrologue(pkt []byte, wantProto byte) (parsedIP, bool) {
 
 // ipHeadersMatch compares the IP portion of two packet header prefixes for
 // byte-for-byte equality on every field that must be identical across
-// coalesced segments. Size/IPID/IPCsum and the 2-bit IP-level ECN field are
-// masked out — the appendPayload step merges CE into the seed.
+// coalesced segments. Size/IPID/IPCsum are masked out. The full DSCP/ECN
+// byte (IPv4 ToS / IPv6 traffic class) is compared, matching Linux kernel
+// GRO: segments with differing ECN codepoints must not coalesce, otherwise
+// ORing e.g. ECT(0) with ECT(1) would fabricate a false CE (congestion)
+// mark or mark a Not-ECT flow as ECN-capable.
 //
 // The transport (L4) portion of the header is checked separately by the
 // per-protocol matcher.
@@ -102,11 +105,11 @@ func ipHeadersMatch(a, b []byte, isV6 bool) bool {
 	if isV6 {
 		// IPv6: byte 0 = version/TC[7:4], byte 1 = TC[3:0]/flow[19:16],
 		// bytes [2:4] = flow[15:0], [6:8] = next_hdr/hop, [8:40] = src+dst.
-		// ECN lives in TC[1:0] = byte 1 mask 0x30. Skip [4:6] payload_len.
+		// Compare byte 1 fully so ECN (TC[1:0]) must match. Skip [4:6] payload_len.
 		if a[0] != b[0] {
 			return false
 		}
-		if a[1]&^0x30 != b[1]&^0x30 {
+		if a[1] != b[1] {
 			return false
 		}
 		if !bytes.Equal(a[2:4], b[2:4]) {
@@ -119,11 +122,12 @@ func ipHeadersMatch(a, b []byte, isV6 bool) bool {
 	}
 	// IPv4: byte 0 = version/IHL, byte 1 = DSCP(6)|ECN(2),
 	// [6:10] flags/fragoff/TTL/proto, [12:20] src+dst.
+	// Compare byte 1 fully so ECN must match.
 	// Skip [2:4] total len, [4:6] id, [10:12] csum.
 	if a[0] != b[0] {
 		return false
 	}
-	if a[1]&^0x03 != b[1]&^0x03 {
+	if a[1] != b[1] {
 		return false
 	}
 	if !bytes.Equal(a[6:10], b[6:10]) {
@@ -135,19 +139,6 @@ func ipHeadersMatch(a, b []byte, isV6 bool) bool {
 	return true
 }
 
-// mergeECNIntoSeed ORs the 2-bit IP-level ECN field of pkt's IP header
-// onto the seed's IP header, so a CE mark on any coalesced segment
-// propagates to the final superpacket. (CE is 0b11; ORing yields CE if
-// any segment carried it.) Used by both TCP and UDP coalescers, so the
-// invariant lives in one place.
-func mergeECNIntoSeed(seedHdr, pktHdr []byte, isV6 bool) {
-	if isV6 {
-		seedHdr[1] |= pktHdr[1] & 0x30
-	} else {
-		seedHdr[1] |= pktHdr[1] & 0x03
-	}
-}
-
 // Arena is an injectable byte-slab that hands out non-overlapping borrowed
 // slices via Reserve and releases them in bulk via Reset. Coalescers take
 // an *Arena at construction so the caller controls the slab lifetime and

overlay/batch/tcp_coalesce.go

@@ -365,9 +365,6 @@ func (c *TCPCoalescer) appendPayload(s *coalesceSlot, pkt []byte, info parsedTCP
 		// last segment. Without this the sender's push signal is dropped.
 		s.hdrBuf[s.ipHdrLen+13] |= tcpFlagPsh
 	}
-	// Merge IP-level CE marks into the seed: headersMatch ignores ECN, so
-	// this is the one place the signal is preserved.
-	mergeECNIntoSeed(s.hdrBuf[:s.ipHdrLen], pkt[:s.ipHdrLen], s.isV6)
 	if info.payLen < s.gsoSize || info.flags&tcpFlagPsh != 0 {
 		s.psh = true
 	}
@@ -424,8 +421,9 @@ func (c *TCPCoalescer) flushSlot(s *coalesceSlot) error {
 
 // headersMatch compares two IP+TCP header prefixes for byte-for-byte
 // equality on every field that must be identical across coalesced
-// segments. Size/IPID/IPCsum/seq/flags/tcpCsum are masked out, as is the
-// 2-bit IP-level ECN field — appendPayload merges CE into the seed.
+// segments. Size/IPID/IPCsum/seq/flags/tcpCsum are masked out. The IP-level
+// ECN codepoint is compared (via ipHeadersMatch) so segments with differing
+// ECN don't coalesce, matching kernel GRO.
 func headersMatch(a, b []byte, isV6 bool, ipHdrLen int) bool {
 	if len(a) != len(b) {
 		return false
@@ -632,6 +630,9 @@ func flowKeyCompare(a, b flowKey) int {
 // ECE state must agree across both slots: PSH is a semantic delimiter
 // (preserving the sender's push boundary) and ECE state must be uniform
 // across a window (the same rule canAppend enforces for in-flow appends).
+// The IP-level ECN codepoint must also match: this check calls headersMatch
+// → ipHeadersMatch, which compares the full DSCP/ECN byte, so two slots with
+// differing ECN marks stay separate superpackets, each keeping its own mark.
 //
 // Note: a slot sealed by reorder (canAppend returned false on seq
 // mismatch) keeps psh=false, so this restriction does not block the
@@ -670,10 +671,9 @@ func canMergeSlots(prev, s *coalesceSlot) bool {
 }
 
 // mergeSlots folds src into dst in place: payIovs concatenated, counters
-// and totals updated, PSH and IP-level CE bits OR'd into the seed header
-// so neither the push signal nor a CE mark is lost. The seed header's
-// seq, gsoSize, and fk are unchanged. Caller is responsible for releasing
-// src (it's no longer in c.slots after this call).
+// and totals updated, PSH OR'd into the seed header so the push signal is
+// not lost. The seed header's seq, gsoSize, and fk are unchanged. Caller
+// is responsible for releasing src (it's no longer in c.slots after this call).
 func mergeSlots(dst, src *coalesceSlot) {
 	dst.payIovs = append(dst.payIovs, src.payIovs...)
 	dst.numSeg += src.numSeg
@@ -683,7 +683,6 @@ func mergeSlots(dst, src *coalesceSlot) {
 		dst.psh = true
 		dst.hdrBuf[dst.ipHdrLen+13] |= tcpFlagPsh
 	}
-	mergeECNIntoSeed(dst.hdrBuf[:dst.ipHdrLen], src.hdrBuf[:src.ipHdrLen], dst.isV6)
 }
 
 // ipv4HdrChecksum computes the IPv4 header checksum over hdr (which must

overlay/batch/tcp_coalesce_test.go

@@ -762,39 +762,88 @@ func TestCoalescerEceMismatchReseeds(t *testing.T) {
 	}
 }
 
-// TestCoalescerMergesCEMark confirms that an ECT(0) burst with a single
-// CE-marked packet still coalesces, and the merged superpacket carries CE.
-func TestCoalescerMergesCEMark(t *testing.T) {
+// TestCoalescerDifferingECNReseeds confirms that segments with differing IP
+// ECN codepoints do NOT coalesce: headersMatch compares the full ToS byte,
+// matching kernel GRO. Two ECT(0) segments merge; a CE stamp mid-run seals
+// the ECT(0) chain and starts a fresh superpacket that keeps CE; a trailing
+// ECT(0) starts yet another. Each superpacket keeps its own codepoint —
+// ORing the marks (the old buggy behavior) would have fabricated a false CE
+// across the whole burst.
+func TestCoalescerDifferingECNReseeds(t *testing.T) {
 	w := &fakeTunWriter{gsoEnabled: true}
 	c := NewTCPCoalescer(w, test.NewLogger(), NewArena(0))
 	pay := make([]byte, 1200)
 	if err := c.Commit(buildTCPv4WithToS(ecnECT0, 1000, tcpAck, pay)); err != nil {
 		t.Fatal(err)
 	}
-	// Router along the path stamped CE on this one.
-	if err := c.Commit(buildTCPv4WithToS(ecnCE, 2200, tcpAck, pay)); err != nil {
+	if err := c.Commit(buildTCPv4WithToS(ecnECT0, 2200, tcpAck, pay)); err != nil {
 		t.Fatal(err)
 	}
-	if err := c.Commit(buildTCPv4WithToS(ecnECT0, 3400, tcpAck, pay)); err != nil {
+	// Router along the path stamped CE on this one.
+	if err := c.Commit(buildTCPv4WithToS(ecnCE, 3400, tcpAck, pay)); err != nil {
+		t.Fatal(err)
+	}
+	if err := c.Commit(buildTCPv4WithToS(ecnECT0, 4600, tcpAck, pay)); err != nil {
 		t.Fatal(err)
 	}
 	if err := c.Flush(); err != nil {
 		t.Fatal(err)
 	}
-	if len(w.gsoWrites) != 1 {
-		t.Fatalf("want 1 merged gso write, got %d (plain=%d)", len(w.gsoWrites), len(w.writes))
+	if len(w.gsoWrites) != 3 {
+		t.Fatalf("want 3 superpackets (ECN split), got %d (plain=%d)", len(w.gsoWrites), len(w.writes))
 	}
-	g := w.gsoWrites[0]
-	if len(g.pays) != 3 {
-		t.Errorf("pay count=%d want 3", len(g.pays))
+	// gso[0]: the two ECT(0) segments merged; gso[1]: CE alone; gso[2]:
+	// trailing ECT(0) alone. Emitted in seq order.
+	type want struct {
+		pays int
+		ecn  byte
 	}
-	if got := g.hdr[1] & 0x03; got != ecnCE {
-		t.Errorf("seed ECN=0x%02x want CE 0x%02x", got, ecnCE)
+	wants := []want{{2, ecnECT0}, {1, ecnCE}, {1, ecnECT0}}
+	for i, wnt := range wants {
+		g := w.gsoWrites[i]
+		if len(g.pays) != wnt.pays {
+			t.Errorf("gso %d pay count=%d want %d", i, len(g.pays), wnt.pays)
+		}
+		if got := g.hdr[1] & 0x03; got != wnt.ecn {
+			t.Errorf("gso %d ECN=0x%02x want 0x%02x", i, got, wnt.ecn)
+		}
 	}
 }
 
-// TestCoalescerDscpMismatchReseeds confirms that the new ECN-mask in
-// headersMatch did not also relax DSCP — different DSCP must still split.
+// TestCoalescerECT0ThenECT1NoCE is the core regression for the ECN merge
+// bug: ORing ECT(0)=0b10 with ECT(1)=0b01 fabricates CE=0b11. The two
+// segments must land in separate superpackets, each preserving its own
+// codepoint, and neither may end up CE-marked.
+func TestCoalescerECT0ThenECT1NoCE(t *testing.T) {
+	w := &fakeTunWriter{gsoEnabled: true}
+	c := NewTCPCoalescer(w, test.NewLogger(), NewArena(0))
+	pay := make([]byte, 1200)
+	if err := c.Commit(buildTCPv4WithToS(ecnECT0, 1000, tcpAck, pay)); err != nil {
+		t.Fatal(err)
+	}
+	if err := c.Commit(buildTCPv4WithToS(ecnECT1, 2200, tcpAck, pay)); err != nil {
+		t.Fatal(err)
+	}
+	if err := c.Flush(); err != nil {
+		t.Fatal(err)
+	}
+	if len(w.gsoWrites) != 2 {
+		t.Fatalf("want 2 separate superpackets (ECT0 vs ECT1), got %d", len(w.gsoWrites))
+	}
+	wantECN := []byte{ecnECT0, ecnECT1}
+	for i, g := range w.gsoWrites {
+		if got := g.hdr[1] & 0x03; got != wantECN[i] {
+			t.Errorf("gso %d ECN=0x%02x want 0x%02x", i, got, wantECN[i])
+		}
+		if got := g.hdr[1] & 0x03; got == ecnCE {
+			t.Errorf("gso %d fabricated CE from ECT merge", i)
+		}
+	}
+}
+
+// TestCoalescerDscpMismatchReseeds confirms that a DSCP difference (same
+// ECN) still splits — headersMatch compares the full ToS byte, so the upper
+// six DSCP bits must match too.
 func TestCoalescerDscpMismatchReseeds(t *testing.T) {
 	w := &fakeTunWriter{gsoEnabled: true}
 	c := NewTCPCoalescer(w, test.NewLogger(), NewArena(0))
@@ -995,9 +1044,10 @@ func TestCoalescerSortKeepsPassthroughBarrier(t *testing.T) {
 	}
 }
 
-// TestCoalescerIPv6MergesCEMark is the IPv6 analogue of
-// TestCoalescerMergesCEMark. ECN bits live in TC[1:0] = byte 1 mask 0x30.
-func TestCoalescerIPv6MergesCEMark(t *testing.T) {
+// TestCoalescerIPv6DifferingECNReseeds is the IPv6 analogue of
+// TestCoalescerDifferingECNReseeds. ECN bits live in TC[1:0] = byte 1 mask
+// 0x30, so ipHeadersMatch (comparing byte 1 fully) still splits them.
+func TestCoalescerIPv6DifferingECNReseeds(t *testing.T) {
 	w := &fakeTunWriter{gsoEnabled: true}
 	c := NewTCPCoalescer(w, test.NewLogger(), NewArena(0))
 	pay := make([]byte, 1200)
@@ -1005,20 +1055,36 @@ func TestCoalescerIPv6MergesCEMark(t *testing.T) {
 	if err := c.Commit(buildTCPv6(ecnECT0, 1000, tcpAck, pay)); err != nil {
 		t.Fatal(err)
 	}
-	if err := c.Commit(buildTCPv6(ecnCE, 2200, tcpAck, pay)); err != nil {
+	if err := c.Commit(buildTCPv6(ecnECT0, 2200, tcpAck, pay)); err != nil {
+		t.Fatal(err)
+	}
+	if err := c.Commit(buildTCPv6(ecnCE, 3400, tcpAck, pay)); err != nil {
+		t.Fatal(err)
+	}
+	if err := c.Commit(buildTCPv6(ecnECT0, 4600, tcpAck, pay)); err != nil {
 		t.Fatal(err)
 	}
 	if err := c.Flush(); err != nil {
 		t.Fatal(err)
 	}
-	if len(w.gsoWrites) != 1 {
-		t.Fatalf("want 1 merged gso write, got %d", len(w.gsoWrites))
+	if len(w.gsoWrites) != 3 {
+		t.Fatalf("want 3 superpackets (ECN split), got %d", len(w.gsoWrites))
 	}
-	g := w.gsoWrites[0]
 	// Byte 1 high nibble holds TC[3:0]; ECN is the low 2 bits of that nibble,
 	// which appears in byte 1 mask 0x30 (>>4 to read the codepoint value).
-	if got := (g.hdr[1] >> 4) & 0x03; got != ecnCE {
-		t.Errorf("seed v6 ECN=0x%02x want CE 0x%02x", got, ecnCE)
+	type want struct {
+		pays int
+		ecn  byte
+	}
+	wants := []want{{2, ecnECT0}, {1, ecnCE}, {1, ecnECT0}}
+	for i, wnt := range wants {
+		g := w.gsoWrites[i]
+		if len(g.pays) != wnt.pays {
+			t.Errorf("gso %d pay count=%d want %d", i, len(g.pays), wnt.pays)
+		}
+		if got := (g.hdr[1] >> 4) & 0x03; got != wnt.ecn {
+			t.Errorf("gso %d v6 ECN=0x%02x want 0x%02x", i, got, wnt.ecn)
+		}
 	}
 }
 

overlay/batch/udp_coalesce.go

@@ -257,8 +257,6 @@ func (c *UDPCoalescer) appendPayload(s *udpSlot, pkt []byte, info parsedUDP) {
 	s.payIovs = append(s.payIovs, pkt[info.hdrLen:info.hdrLen+info.payLen])
 	s.numSeg++
 	s.totalPay += info.payLen
-	// Merge IP-level CE marks into the seed (same trick TCP coalescer uses).
-	mergeECNIntoSeed(s.hdrBuf[:s.ipHdrLen], pkt[:s.ipHdrLen], s.isV6)
 	if info.payLen < s.gsoSize {
 		// Last-segment-can-be-shorter: this seals the chain.
 		s.sealed = true
@@ -329,8 +327,9 @@ func (c *UDPCoalescer) flushSlot(s *udpSlot) error {
 
 // udpHeadersMatch compares two IP+UDP header prefixes for byte-equality on
 // every field that must be identical across coalesced segments. Length
-// fields and the ECN bits in IP TOS/TC are masked out — appendPayload
-// merges CE into the seed; flushSlot rewrites lengths.
+// fields are masked out (flushSlot rewrites them), but the IP-level ECN
+// codepoint is compared (via ipHeadersMatch) so segments with differing ECN
+// don't coalesce, matching kernel GRO.
 func udpHeadersMatch(a, b []byte, isV6 bool, ipHdrLen int) bool {
 	if len(a) != len(b) {
 		return false

overlay/batch/udp_coalesce_test.go

@@ -261,32 +261,37 @@ func TestUDPCoalescerCapsAtMaxSegs(t *testing.T) {
 	}
 }
 
-// CE marks on appended segments must be merged into the seed's IP TOS.
-func TestUDPCoalescerMergesCEMark(t *testing.T) {
+// Differing IP ECN codepoints must not coalesce: udpHeadersMatch compares
+// the full ToS byte (matching kernel GRO). A CE-marked datagram mid-run
+// seals the Not-ECT chain and seeds a fresh superpacket that keeps CE; the
+// trailing Not-ECT datagram seeds another.
+func TestUDPCoalescerDifferingECNReseeds(t *testing.T) {
 	w := &fakeTunWriter{gsoEnabled: true}
 	c := NewUDPCoalescer(w, NewArena(0))
 	pay := make([]byte, 800)
-	pkt0 := buildUDPv4(1000, 53, pay) // ECN=00
+	pkt0 := buildUDPv4(1000, 53, pay) // ECN=00 (Not-ECT)
 	pkt1 := buildUDPv4(1000, 53, pay)
-	pkt1[1] = 0x03 // CE
-	pkt2 := buildUDPv4(1000, 53, pay)
-	if err := c.Commit(pkt0); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Commit(pkt1); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Commit(pkt2); err != nil {
-		t.Fatal(err)
+	pkt1[1] = 0x03                    // CE
+	pkt2 := buildUDPv4(1000, 53, pay) // ECN=00 again
+	for _, p := range [][]byte{pkt0, pkt1, pkt2} {
+		if err := c.Commit(p); err != nil {
+			t.Fatal(err)
+		}
 	}
 	if err := c.Flush(); err != nil {
 		t.Fatal(err)
 	}
-	if len(w.gsoWrites) != 1 {
-		t.Fatalf("want 1 merged gso write, got %d (plain=%d)", len(w.gsoWrites), len(w.writes))
+	if len(w.gsoWrites) != 3 {
+		t.Fatalf("want 3 separate seeds (differing ECN), got %d (plain=%d)", len(w.gsoWrites), len(w.writes))
 	}
-	if w.gsoWrites[0].hdr[1]&0x03 != 0x03 {
-		t.Errorf("CE not merged into seed (tos=%#x)", w.gsoWrites[0].hdr[1])
+	wantECN := []byte{0x00, 0x03, 0x00}
+	for i, g := range w.gsoWrites {
+		if len(g.pays) != 1 {
+			t.Errorf("gso %d pay count=%d want 1", i, len(g.pays))
+		}
+		if got := g.hdr[1] & 0x03; got != wantECN[i] {
+			t.Errorf("gso %d ECN=%#x want %#x", i, got, wantECN[i])
+		}
 	}
 }
 
@@ -326,7 +331,7 @@ func TestUDPCoalescerIPv6Coalesces(t *testing.T) {
 	}
 }
 
-// DSCP differences must reseed (headers don't match outside ECN).
+// DSCP differences must reseed: udpHeadersMatch compares the full ToS byte.
 func TestUDPCoalescerDSCPMismatchReseeds(t *testing.T) {
 	w := &fakeTunWriter{gsoEnabled: true}
 	c := NewUDPCoalescer(w, NewArena(0))

overlay/tio/blockon_linux.go

@@ -0,0 +1,50 @@
+//go:build linux && !android
+// +build linux,!android
+
+package tio
+
+import (
+	"os"
+
+	"golang.org/x/sys/unix"
+)
+
+// blockOn parks the calling goroutine until fd is ready (events is POLLIN for
+// reads, POLLOUT for writes) or shutdownFd signals teardown. It builds the
+// pollfd array on the stack every call, so concurrent callers on the same
+// Queue never share Revents storage: the previous shared-array implementation
+// was a genuine Go data race when two writers parked in poll(2) at once (the
+// kernel writing Revents while another goroutine zeroed it). Level-triggered
+// events kept it from deadlocking, but it was still a race.
+//
+// Poll(2) is looped over EINTR. err is checked before the Revents bits are
+// trusted, since a failed poll may leave them bogus. Returns os.ErrClosed when
+// shutdown was signaled (POLLIN on shutdownFd) or either fd reported a problem
+// condition (POLLHUP|POLLNVAL|POLLERR).
+func blockOn(fd, shutdownFd int32, events int16) error {
+	const problemFlags = unix.POLLHUP | unix.POLLNVAL | unix.POLLERR
+	pfds := [2]unix.PollFd{
+		{Fd: fd, Events: events},
+		{Fd: shutdownFd, Events: unix.POLLIN},
+	}
+	var err error
+	for {
+		_, err = unix.Poll(pfds[:], -1)
+		if err != unix.EINTR {
+			break
+		}
+	}
+	tunEvents := pfds[0].Revents
+	shutdownEvents := pfds[1].Revents
+	// Check err before trusting the potentially bogus bits we just got.
+	if err != nil {
+		return err
+	}
+	if shutdownEvents&(unix.POLLIN|problemFlags) != 0 {
+		return os.ErrClosed
+	}
+	if tunEvents&problemFlags != 0 {
+		return os.ErrClosed
+	}
+	return nil
+}

overlay/tio/tio.go

@@ -26,8 +26,9 @@ type Capabilities struct {
 	USO bool
 }
 
-// Queue is a readable/writable Poll queue. One Queue is driven by a single
-// read goroutine plus a single writer (see Write below).
+// Queue is a readable/writable Poll queue. Concurrency contract: a single
+// read goroutine drives Read; plain Write is safe for concurrent callers;
+// WriteGSO (on Queues that implement GSOWriter) is single-writer per queue.
 type Queue interface {
 	io.Closer
 
@@ -37,11 +38,12 @@ type Queue interface {
 	// or copy each slice before the next call. A Packet may carry a
 	// GSO/USO superpacket (see GSOInfo); when GSO.IsSuperpacket() is
 	// true the caller must segment Bytes before treating it as a single
-	// IP datagram. Not safe for concurrent Reads.
+	// IP datagram. Single-reader only: not safe for concurrent Reads (it
+	// reuses per-queue rx scratch each call).
 	Read() ([]Packet, error)
 
 	// Write emits a single packet on the plaintext (outside→inside)
-	// delivery path. Not safe for concurrent Writes.
+	// delivery path. Safe for concurrent use.
 	Write(p []byte) (int, error)
 }
 

overlay/tio/tio_gso_linux.go

@@ -8,7 +8,6 @@ import (
 	"io"
 	"log/slog"
 	"os"
-	"sync"
 	"sync/atomic"
 	"syscall"
 	"unsafe"
@@ -62,17 +61,10 @@ var validVnetHdr = [virtio.Size]byte{unix.VIRTIO_NET_HDR_F_DATA_VALID}
 type Offload struct {
 	fd         int
 	shutdownFd int
-	readPoll   [2]unix.PollFd
-	writePoll  [2]unix.PollFd
-	// writeLock serializes blockOnWrite's read+clear of writePoll[*].Revents.
-	// Any goroutine that calls Write may end up parked in poll(2); without
-	// the lock concurrent waiters could race the Revents reset and lose
-	// events.
-	writeLock sync.Mutex
-	closed    atomic.Bool
-	rxBuf     []byte   // backing store for kernel-handed packets read this drain
-	rxOff     int      // cursor into rxBuf for the current Read drain
-	pending   []Packet // packets returned from the most recent Read
+	closed     atomic.Bool
+	rxBuf      []byte   // backing store for kernel-handed packets read this drain
+	rxOff      int      // cursor into rxBuf for the current Read drain
+	pending    []Packet // packets returned from the most recent Read
 
 	// readVnetScratch holds the 10-byte virtio_net_hdr split off the front of
 	// every TUN read via readv(2). Decoupling the header from the packet body
@@ -109,15 +101,6 @@ func newOffload(fd int, shutdownFd int, usoEnabled bool) (*Offload, error) {
 		shutdownFd: shutdownFd,
 		usoEnabled: usoEnabled,
 		closed:     atomic.Bool{},
-		readPoll: [2]unix.PollFd{
-			{Fd: int32(fd), Events: unix.POLLIN},
-			{Fd: int32(shutdownFd), Events: unix.POLLIN},
-		},
-		writePoll: [2]unix.PollFd{
-			{Fd: int32(fd), Events: unix.POLLOUT},
-			{Fd: int32(shutdownFd), Events: unix.POLLIN},
-		},
-		writeLock: sync.Mutex{},
 
 		rxBuf:   make([]byte, tunRxBufCap),
 		gsoIovs: make([]unix.Iovec, 2, gsoMaxIovs),
@@ -135,57 +118,11 @@ func newOffload(fd int, shutdownFd int, usoEnabled bool) (*Offload, error) {
 }
 
 func (r *Offload) blockOnRead() error {
-	const problemFlags = unix.POLLHUP | unix.POLLNVAL | unix.POLLERR
-	var err error
-	for {
-		_, err = unix.Poll(r.readPoll[:], -1)
-		if err != unix.EINTR {
-			break
-		}
-	}
-	//always reset these!
-	tunEvents := r.readPoll[0].Revents
-	shutdownEvents := r.readPoll[1].Revents
-	r.readPoll[0].Revents = 0
-	r.readPoll[1].Revents = 0
-	//do the err check before trusting the potentially bogus bits we just got
-	if err != nil {
-		return err
-	}
-	if shutdownEvents&(unix.POLLIN|problemFlags) != 0 {
-		return os.ErrClosed
-	} else if tunEvents&problemFlags != 0 {
-		return os.ErrClosed
-	}
-	return nil
+	return blockOn(int32(r.fd), int32(r.shutdownFd), unix.POLLIN)
 }
 
 func (r *Offload) blockOnWrite() error {
-	const problemFlags = unix.POLLHUP | unix.POLLNVAL | unix.POLLERR
-	var err error
-	for {
-		_, err = unix.Poll(r.writePoll[:], -1)
-		if err != unix.EINTR {
-			break
-		}
-	}
-	//always reset these!
-	r.writeLock.Lock()
-	tunEvents := r.writePoll[0].Revents
-	shutdownEvents := r.writePoll[1].Revents
-	r.writePoll[0].Revents = 0
-	r.writePoll[1].Revents = 0
-	r.writeLock.Unlock()
-	//do the err check before trusting the potentially bogus bits we just got
-	if err != nil {
-		return err
-	}
-	if shutdownEvents&(unix.POLLIN|problemFlags) != 0 {
-		return os.ErrClosed
-	} else if tunEvents&problemFlags != 0 {
-		return os.ErrClosed
-	}
-	return nil
+	return blockOn(int32(r.fd), int32(r.shutdownFd), unix.POLLOUT)
 }
 
 // readPacket issues a single readv(2) splitting the virtio_net_hdr off

overlay/tio/tio_poll_linux.go

@@ -17,11 +17,9 @@ import (
 const tunReadBufSize = 65535
 
 type Poll struct {
-	fd int
-
-	readPoll  [2]unix.PollFd
-	writePoll [2]unix.PollFd
-	closed    atomic.Bool
+	fd         int
+	shutdownFd int
+	closed     atomic.Bool
 
 	readBuf  []byte
 	batchRet [1]Packet
@@ -37,16 +35,9 @@ func newPoll(fd int, shutdownFd int) (*Poll, error) {
 	}
 
 	out := &Poll{
-		fd:      fd,
-		readBuf: make([]byte, tunReadBufSize),
-		readPoll: [2]unix.PollFd{
-			{Fd: int32(fd), Events: unix.POLLIN},
-			{Fd: int32(shutdownFd), Events: unix.POLLIN},
-		},
-		writePoll: [2]unix.PollFd{
-			{Fd: int32(fd), Events: unix.POLLOUT},
-			{Fd: int32(shutdownFd), Events: unix.POLLIN},
-		},
+		fd:         fd,
+		shutdownFd: shutdownFd,
+		readBuf:    make([]byte, tunReadBufSize),
 	}
 	return out, nil
 }
@@ -54,53 +45,11 @@ func newPoll(fd int, shutdownFd int) (*Poll, error) {
 // blockOnRead waits until the Poll fd is readable or shutdown has been signaled.
 // Returns os.ErrClosed if Close was called.
 func (t *Poll) blockOnRead() error {
-	const problemFlags = unix.POLLHUP | unix.POLLNVAL | unix.POLLERR
-	var err error
-	for {
-		_, err = unix.Poll(t.readPoll[:], -1)
-		if err != unix.EINTR {
-			break
-		}
-	}
-	tunEvents := t.readPoll[0].Revents
-	shutdownEvents := t.readPoll[1].Revents
-	t.readPoll[0].Revents = 0
-	t.readPoll[1].Revents = 0
-	if err != nil {
-		return err
-	}
-	if shutdownEvents&(unix.POLLIN|problemFlags) != 0 {
-		return os.ErrClosed
-	}
-	if tunEvents&problemFlags != 0 {
-		return os.ErrClosed
-	}
-	return nil
+	return blockOn(int32(t.fd), int32(t.shutdownFd), unix.POLLIN)
 }
 
 func (t *Poll) blockOnWrite() error {
-	const problemFlags = unix.POLLHUP | unix.POLLNVAL | unix.POLLERR
-	var err error
-	for {
-		_, err = unix.Poll(t.writePoll[:], -1)
-		if err != unix.EINTR {
-			break
-		}
-	}
-	tunEvents := t.writePoll[0].Revents
-	shutdownEvents := t.writePoll[1].Revents
-	t.writePoll[0].Revents = 0
-	t.writePoll[1].Revents = 0
-	if err != nil {
-		return err
-	}
-	if shutdownEvents&(unix.POLLIN|problemFlags) != 0 {
-		return os.ErrClosed
-	}
-	if tunEvents&problemFlags != 0 {
-		return os.ErrClosed
-	}
-	return nil
+	return blockOn(int32(t.fd), int32(t.shutdownFd), unix.POLLOUT)
 }
 
 func (t *Poll) Read() ([]Packet, error) {
@@ -133,7 +82,7 @@ func (t *Poll) readOne(to []byte) (int, error) {
 	}
 }
 
-// Write is only valid for single threaded use
+// Write is safe for concurrent use
 func (t *Poll) Write(from []byte) (int, error) {
 	for {
 		n, errno := unix.Write(t.fd, from)

overlay/tio/tun_file_linux_test.go

@@ -70,6 +70,76 @@ func TestPoll_WakeForShutdown_WakesFriends(t *testing.T) {
 	}
 }
 
+// TestPoll_ConcurrentWrite_NoRace hammers a single Poll queue from two writer
+// goroutines while a reader drains the other end of the pipe. The writers
+// overflow the pipe buffer, so both repeatedly park in blockOnWrite at the same
+// time — the exact scenario that raced on the old shared writePoll member
+// array. Run under -race; a shared-array regression trips the detector here.
+func TestPoll_ConcurrentWrite_NoRace(t *testing.T) {
+	var fds [2]int
+	require.NoError(t, unix.Pipe2(fds[:], unix.O_CLOEXEC))
+	readFd, writeFd := fds[0], fds[1]
+
+	shutdownFd, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = unix.Close(shutdownFd) })
+
+	p, err := newPoll(writeFd, shutdownFd)
+	require.NoError(t, err)
+
+	const writers = 2
+	const perWriter = 4000
+	payload := make([]byte, 100)
+	total := writers * perWriter * len(payload)
+
+	// Reader: drain the read end (blocking) until every writer's bytes are
+	// consumed, so the writers keep making progress rather than wedging on a
+	// permanently full pipe.
+	readDone := make(chan struct{})
+	go func() {
+		defer close(readDone)
+		buf := make([]byte, 4096)
+		got := 0
+		for got < total {
+			n, rerr := unix.Read(readFd, buf)
+			got += n
+			if rerr != nil {
+				if rerr == unix.EINTR {
+					continue
+				}
+				return
+			}
+			if n == 0 { // EOF
+				return
+			}
+		}
+	}()
+
+	var wg sync.WaitGroup
+	for w := 0; w < writers; w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for i := 0; i < perWriter; i++ {
+				if _, werr := p.Write(payload); werr != nil {
+					t.Errorf("write: %v", werr)
+					return
+				}
+			}
+		}()
+	}
+	wg.Wait()
+
+	select {
+	case <-readDone:
+	case <-time.After(10 * time.Second):
+		t.Fatal("reader did not drain")
+	}
+
+	require.NoError(t, p.Close())
+	_ = unix.Close(readFd)
+}
+
 // TestPoll_NewPoll_DoesNotCloseFdOnFailure pins the ownership rule: when
 // newPoll fails, it must leave fd open so the caller (pollQueueSet.Add's
 // callers in tun_linux.go) is the sole closer. If newPoll also closed fd,

overlay/tio/virtio/segment_linux.go

@@ -143,7 +143,7 @@ func CorrectHdrLen(pkt []byte, hdr *Hdr) error {
 	if hdr.HdrLen < hdr.CsumStart {
 		return fmt.Errorf("virtioNetHdr.HdrLen (%d) < virtioNetHdr.CsumStart (%d)", hdr.HdrLen, hdr.CsumStart)
 	}
-	cSumAt := int(hdr.CsumStart + hdr.CsumStart)
+	cSumAt := int(hdr.CsumStart + hdr.CsumOffset)
 	if cSumAt+1 >= len(pkt) {
 		return fmt.Errorf("end of checksum offset (%d) exceeds packet length (%d)", cSumAt+1, len(pkt))
 	}

overlay/tio/virtio/segment_linux_test.go

@@ -211,6 +211,55 @@ func TestSegmentTCPHeaderNotCorrupted(t *testing.T) {
 	}
 }
 
+// TestCorrectHdrLenChecksumBound guards the checksum-field bounds check in
+// CorrectHdrLen. The checksum field sits at CsumStart+CsumOffset, so the check
+// must be computed from CsumStart+CsumOffset — NOT CsumStart+CsumStart, a
+// regression that doubled CsumStart and thus over-tightened the bound (since
+// CsumOffset, 6 for UDP / 16 for TCP, is always < CsumStart >= 20). That bogus
+// bound spuriously rejected valid small USO superpackets in decodeRead.
+func TestCorrectHdrLenChecksumBound(t *testing.T) {
+	// A valid IPv4 USO superpacket: 20B IPv4 + 8B UDP + two 6-byte segments
+	// (payload 12) = 40 bytes total. CsumStart=20, CsumOffset=6, so the UDP
+	// checksum field lives at bytes 26..27, comfortably inside the 40-byte
+	// packet. The OLD formula computed cSumAt = CsumStart+CsumStart = 40 and
+	// rejected on cSumAt+1 (41) >= len(pkt) (40); the fix (CsumStart+CsumOffset
+	// = 26) accepts. This case FAILS against the CsumStart+CsumStart regression.
+	t.Run("valid-small-uso-accepted", func(t *testing.T) {
+		pkt, _, csumStart := buildUDPv4Super(12) // total len 40
+		hdr := Hdr{
+			Flags:      unix.VIRTIO_NET_HDR_F_NEEDS_CSUM,
+			GSOType:    unix.VIRTIO_NET_HDR_GSO_UDP_L4,
+			GSOSize:    6, // two 6-byte segments
+			CsumStart:  csumStart,
+			CsumOffset: 6,
+		}
+		if err := CorrectHdrLen(pkt, &hdr); err != nil {
+			t.Fatalf("CorrectHdrLen rejected a valid 40-byte USO superpacket: %v", err)
+		}
+		if hdr.HdrLen != csumStart+udpHeaderLen {
+			t.Errorf("HdrLen = %d, want %d", hdr.HdrLen, csumStart+udpHeaderLen)
+		}
+	})
+
+	// A genuinely-too-short packet: CsumStart=20, CsumOffset=6 means the
+	// checksum field would end at byte 27, but the packet is only 25 bytes
+	// (CsumStart+CsumOffset+2 = 28 > 25). CorrectHdrLen must still reject it.
+	t.Run("too-short-rejected", func(t *testing.T) {
+		pkt := make([]byte, 25)
+		pkt[0] = 0x45 // IPv4, IHL 5
+		hdr := Hdr{
+			Flags:      unix.VIRTIO_NET_HDR_F_NEEDS_CSUM,
+			GSOType:    unix.VIRTIO_NET_HDR_GSO_UDP_L4,
+			GSOSize:    6,
+			CsumStart:  20,
+			CsumOffset: 6,
+		}
+		if err := CorrectHdrLen(pkt, &hdr); err == nil {
+			t.Fatalf("CorrectHdrLen accepted a too-short (25-byte) packet")
+		}
+	})
+}
+
 // TestSegmentUDPHeaderNotCorrupted is the USO counterpart: SegmentUDP performs
 // the same header stamp and must be correct when gsoSize < headerLen.
 func TestSegmentUDPHeaderNotCorrupted(t *testing.T) {

pprof_debug.go

@@ -0,0 +1,33 @@
+//go:build debug
+
+package nebula
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"net/http"
+	_ "net/http/pprof" // registers pprof handlers on http.DefaultServeMux
+)
+
+// startPprofServer serves net/http/pprof on :6060 for the life of ctx. It is
+// only compiled into debug builds (`-tags debug`, `make debug`), so a debug
+// build announces itself with the Info line below.
+func startPprofServer(ctx context.Context, l *slog.Logger) {
+	server := &http.Server{Addr: ":6060", Handler: nil}
+	l.Info("Starting pprof debug server (debug build)", "addr", server.Addr)
+
+	go func() {
+		if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			l.Error("pprof debug server stopped", "error", err)
+		}
+	}()
+
+	// Shut down the server when the context is cancelled.
+	go func() {
+		<-ctx.Done()
+		if err := server.Shutdown(context.Background()); err != nil {
+			l.Debug("Error shutting down pprof debug server", "error", err)
+		}
+	}()
+}

pprof_nodebug.go

@@ -0,0 +1,11 @@
+//go:build !debug
+
+package nebula
+
+import (
+	"context"
+	"log/slog"
+)
+
+// startPprofServer is a no-op unless built with `-tags debug` (see make debug).
+func startPprofServer(_ context.Context, _ *slog.Logger) {}

util/cpupin_linux.go

@@ -21,3 +21,23 @@ func PinThreadToCPU(cpu int) error {
 	set.Set(cpu)
 	return unix.SchedSetaffinity(0, &set)
 }
+
+// AllowedCPUs returns the CPU IDs the calling process is currently allowed to
+// run on, as reported by sched_getaffinity(2). Under a cgroup cpuset or a
+// `taskset` mask the allowed IDs are frequently not the contiguous range
+// 0..NumCPU-1 (e.g. pinned to CPUs 4-7: NumCPU reports 4 while the valid IDs
+// are 4,5,6,7). Callers that need a real CPU to pin to must choose from this
+// set rather than assuming i % NumCPU is runnable, or every pin fails.
+func AllowedCPUs() ([]int, error) {
+	var set unix.CPUSet
+	if err := unix.SchedGetaffinity(0, &set); err != nil {
+		return nil, err
+	}
+	cpus := make([]int, 0, set.Count())
+	for cpu := 0; cpu < len(set)*64; cpu++ {
+		if set.IsSet(cpu) {
+			cpus = append(cpus, cpu)
+		}
+	}
+	return cpus, nil
+}

util/cpupin_other.go

@@ -9,3 +9,10 @@ package util
 func PinThreadToCPU(_ int) error {
 	return nil
 }
+
+// AllowedCPUs has no meaningful answer off Linux (no sched_getaffinity), so it
+// reports "unknown" by returning a nil slice and nil error. Callers treat an
+// empty result as "fall back to the default CPU choice".
+func AllowedCPUs() ([]int, error) {
+	return nil, nil
+}