datapath: fix 12 correctness findings from tun/UDP offload review

Multi-disciplinary correctness review of the batched tun / GSO-GRO / sendmmsg
rework. Each fix has a regression test; the merged tree builds on
linux/darwin/openbsd/windows/freebsd/netbsd, vets clean, passes the unit and
e2e suites, and is -race clean.

Critical:
- C1 zero-length inner UDP datagram no longer panics the process (remote DoS):
  the UDP coalescer routes payLen==0 to passthrough instead of seeding a GSO
  slot, and WriteGSO skips empty payload iovecs as defense in depth.
- C2 segmenter no longer corrupts inner headers when gsoSize < headerLen: the
  L3+L4 header is snapshotted once and each segment stamped from the copy,
  replacing the destructive overlapping in-place slide (SegmentTCP + SegmentUDP).

High:
- H1 applyOuterECN updates the IPv4 header checksum (RFC 1624 incremental) when
  folding outer CE into the inner ToS, so passthrough packets are no longer
  dropped by the peer stack.
- H2 the GRO reject path caps the borrowed RX segment ([:n:n]) so a reject can
  no longer overrun into the next coalesced segment's Nebula header. Note:
  oversized ICMPv6 rejects that need >16B beyond the segment are now refused
  rather than sent under GRO (safe; see TOFIX.md for the scratch-buffer follow-up).
- H3 WriteBatch falls back to per-packet WriteTo for a chunk when writeSockaddr
  fails, so one bad-family destination costs only its own packet, not the batch.
- H4 UserDevice.Readers returns N distinct queue wrappers with private buffers
  (sharing the pipes) so concurrent readers no longer race/overwrite borrowed
  packet bytes.
- H5 Poll.Close / Offload.Close no longer null t.fd (matching master's
  tunFile.Close), removing the data race with a concurrent readOne load.

Medium/Low:
- M1 the UDP GSO 127-segment gate moved from kernel >=5.5 to >=6.9 (the real
  UDP_MAX_SEGMENTS 64->128 threshold), avoiding EINVAL + per-packet fallback on
  5.5-6.8 kernels.
- M2 NewMultiQueueReader replays the offload mask newTun actually negotiated
  instead of the TSO-only mask, so adding a queue no longer disables USO
  device-wide; the advertised USO capability derives from the same mask.
- M3 the shutdown eventfd is closed in pollQueueSet.Close / offloadQueueSet.Close
  (double-close guarded), fixing the per-lifecycle fd leak.
- M4 dual-stack ECN selects the cmsg by address family, not socket family: RX
  parseRecvCmsg reads both IP_TOS and IPV6_TCLASS; TX writeEntryCmsg stamps
  IP_TOS for v4/v4-mapped dests and IPV6_TCLASS for v6 (on-host verified).
- L1 newPoll no longer closes the fd on failure (matching newOffload), removing
  the double-close on QueueSet.Add error.

overlay/user.go

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