Don't fail on the batch at the first error (#1826)

control_lifecycle_test.go

@@ -148,8 +148,8 @@ func (c *fakeConn) Rebind() error                             { c.rebinds++; ret
 func (c *fakeConn) LocalAddr() (netip.AddrPort, error)        { return netip.AddrPort{}, nil }
 func (c *fakeConn) ListenOut(_ udp.EncReader, _ func()) error { return nil }
 func (c *fakeConn) WriteTo(_ []byte, _ netip.AddrPort) error  { return nil }
-func (c *fakeConn) WriteBatch(_ [][]byte, _ []netip.AddrPort, _ []byte) error {
+func (c *fakeConn) WriteBatch(bufs [][]byte, _ []netip.AddrPort, _ []byte) (int, error) {
-	return nil
+	return len(bufs), nil
 }
 func (c *fakeConn) ReloadConfig(_ *config.C)      {}
 func (c *fakeConn) SupportsMultipleReaders() bool { return true }

interface.go

@@ -137,6 +137,7 @@ type Interface struct {
 	metricHandshakes    metrics.Histogram
 	messageMetrics      *MessageMetrics
 	cachedPacketMetrics *cachedPacketMetrics
+	metricTxDropped     metrics.Counter
 
 	l *slog.Logger
 }
@@ -237,6 +238,7 @@ func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) {
 		pinThreads:            c.PinThreads,
 
 		metricHandshakes: metrics.GetOrRegisterHistogram("handshakes", nil, metrics.NewExpDecaySample(1028, 0.015)),
+		metricTxDropped:  metrics.GetOrRegisterCounter("udp.tx.dropped", nil),
 		messageMetrics:   c.MessageMetrics,
 		cachedPacketMetrics: &cachedPacketMetrics{
 			sent:    metrics.GetOrRegisterCounter("hostinfo.cached_packets.sent", nil),
@@ -441,19 +443,29 @@ func (f *Interface) listenIn(queue tio.Queue, i int) {
 			// accumulated so the first packets of a deep read drain
 			// hit the wire while the rest are still being encrypted.
 			if sb.Len() >= batch.SendBatchCap {
-				if err := sb.Flush(); err != nil {
+				f.flushSendBatch(sb, i)
-					f.l.Error("Failed to write outgoing batch", "error", err, "writer", i)
-				}
 			}
 		}
-		if err := sb.Flush(); err != nil {
+		f.flushSendBatch(sb, i)
-			f.l.Error("Failed to write outgoing batch", "error", err, "writer", i)
-		}
 	}
 
 	f.l.Debug("overlay reader is done", "reader", i)
 }
 
+// flushSendBatch drains sb to the underlay and accounts for anything it could not deliver. A shortfall means
+// specific destinations were undeliverable (a stale remote, a reject rule), which the backend logs per peer at
+// debug; here it is only a counter, so one unreachable peer cannot spam a log line per batch.
+func (f *Interface) flushSendBatch(sb *batch.SendBatch, q int) {
+	queued := sb.Len()
+	written, err := sb.Flush()
+	if err != nil {
+		f.l.Error("Failed to write outgoing batch", "error", err, "writer", q)
+	}
+	if dropped := queued - written; dropped > 0 {
+		f.metricTxDropped.Inc(int64(dropped))
+	}
+}
+
 func (f *Interface) RegisterConfigChangeCallbacks(c *config.C) {
 	c.RegisterReloadCallback(f.reloadFirewall)
 	c.RegisterReloadCallback(f.reloadSendRecvError)

overlay/batch/batch.go

@@ -19,8 +19,8 @@ type TxBatcher interface {
 	// caller must keep pkt valid until the next Flush. Pass 0 (Not-ECT)
 	// to leave the outer ECN field unset.
 	Commit(pkt []byte, dst netip.AddrPort, outerECN byte)
-	// Flush emits every queued packet via the underlying batch writer in arrival order.
+	// Flush emits every queued packet via the underlying batch writer in arrival order and reports how many were
-	// Returns an errors.Join of one or more errors.
+	// actually written. A short count means some destinations were undeliverable, not that the batch failed.
 	// After Flush returns, borrowed payload slices may be recycled.
-	Flush() error
+	Flush() (int, error)
 }

overlay/batch/tx_batch.go

@@ -6,7 +6,7 @@ const SendBatchCap = 128
 
 // batchWriter is the minimal subset of udp.Conn needed by SendBatch to flush.
 type batchWriter interface {
-	WriteBatch(bufs [][]byte, addrs []netip.AddrPort, outerECNs []byte) error
+	WriteBatch(bufs [][]byte, addrs []netip.AddrPort, outerECNs []byte) (int, error)
 }
 
 // SendBatch accumulates encrypted UDP packets and flushes them via WriteBatch.
@@ -46,15 +46,18 @@ func (b *SendBatch) Commit(pkt []byte, dst netip.AddrPort, outerECN byte) {
 	b.ecns = append(b.ecns, outerECN)
 }
 
-func (b *SendBatch) Flush() error {
+// Flush writes every queued packet and reports how many actually went out. A short count means some destinations
+// were undeliverable; the batch is drained either way.
+func (b *SendBatch) Flush() (int, error) {
 	var err error
+	written := 0
 	if len(b.bufs) > 0 {
-		err = b.out.WriteBatch(b.bufs, b.dsts, b.ecns)
+		written, err = b.out.WriteBatch(b.bufs, b.dsts, b.ecns)
 	}
 	clear(b.bufs)
 	b.bufs = b.bufs[:0]
 	b.dsts = b.dsts[:0]
 	b.ecns = b.ecns[:0]
 	b.arena.Reset()
-	return err
+	return written, err
 }

overlay/batch/tx_batch_test.go

@@ -11,7 +11,7 @@ type fakeBatchWriter struct {
 	ecns  []byte
 }
 
-func (w *fakeBatchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []byte) error {
+func (w *fakeBatchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []byte) (int, error) {
 	// Snapshot — SendBatch.Flush nils its slot pointers right after WriteBatch
 	// returns, so tests must capture data before that happens.
 	w.bufs = make([][]byte, len(bufs))
@@ -22,7 +22,7 @@ func (w *fakeBatchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns
 	}
 	w.addrs = append(w.addrs[:0], addrs...)
 	w.ecns = append(w.ecns[:0], ecns...)
-	return nil
+	return len(bufs), nil
 }
 
 func TestSendBatchReserveCommitFlush(t *testing.T) {
@@ -38,7 +38,7 @@ func TestSendBatchReserveCommitFlush(t *testing.T) {
 		pkt := append(slot[:0], byte(i), byte(i+1), byte(i+2))
 		b.Commit(pkt, ap, 0)
 	}
-	if err := b.Flush(); err != nil {
+	if _, err := b.Flush(); err != nil {
 		t.Fatalf("Flush: %v", err)
 	}
 	if len(fw.bufs) != 4 {
@@ -55,7 +55,7 @@ func TestSendBatchReserveCommitFlush(t *testing.T) {
 
 	// Flush again with nothing committed — should be a no-op.
 	fw.bufs = nil
-	if err := b.Flush(); err != nil {
+	if _, err := b.Flush(); err != nil {
 		t.Fatalf("empty Flush: %v", err)
 	}
 	if fw.bufs != nil {
@@ -79,7 +79,7 @@ func TestSendBatchSlotsDoNotOverlap(t *testing.T) {
 		pkt := append(s[:0], byte(0xA0+i), byte(0xB0+i))
 		b.Commit(pkt, ap, 0)
 	}
-	if err := b.Flush(); err != nil {
+	if _, err := b.Flush(); err != nil {
 		t.Fatalf("Flush: %v", err)
 	}
 
@@ -109,7 +109,7 @@ func TestSendBatchGrowPreservesCommitted(t *testing.T) {
 		t.Fatalf("first packet corrupted by grow: %x", pkt1)
 	}
 
-	if err := b.Flush(); err != nil {
+	if _, err := b.Flush(); err != nil {
 		t.Fatalf("Flush: %v", err)
 	}
 	if len(fw.bufs) != 2 {

udp/conn.go

@@ -48,9 +48,12 @@ type Conn interface {
 	// same length as bufs, and outerECNs[i] is the 2-bit IP-level ECN
 	// codepoint to set on packet i's outer header. Linux uses sendmmsg(2)
 	// for a single syscall and attaches the value as IP_TOS / IPV6_TCLASS
-	// cmsg; other backends ignore it. Returns on the first error; callers
+	// cmsg; other backends ignore it.
-	// may observe a partial send if some packets went out before the error.
+	//
-	WriteBatch(bufs [][]byte, addrs []netip.AddrPort, outerECNs []byte) error
+	// Returns the number of packets successfully written. A destination the kernel rejects costs only
+	// its own packet, so a short count means some peers were undeliverable, not that the batch failed.
+	// Not safe for concurrent use on the same Conn.
+	WriteBatch(bufs [][]byte, addrs []netip.AddrPort, outerECNs []byte) (int, error)
 	ReloadConfig(c *config.C)
 	SupportsMultipleReaders() bool
 	Close() error
@@ -73,8 +76,8 @@ func (NoopConn) SupportsMultipleReaders() bool {
 func (NoopConn) WriteTo(_ []byte, _ netip.AddrPort) error {
 	return nil
 }
-func (NoopConn) WriteBatch(_ [][]byte, _ []netip.AddrPort, _ []byte) error {
+func (NoopConn) WriteBatch(bufs [][]byte, _ []netip.AddrPort, _ []byte) (int, error) {
-	return nil
+	return len(bufs), nil
 }
 func (NoopConn) ReloadConfig(_ *config.C) {
 	return

udp/udp_darwin.go

@@ -140,13 +140,17 @@ func (u *StdConn) WriteTo(b []byte, ap netip.AddrPort) error {
 	}
 }
 
-func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, _ []byte) error {
+func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, _ []byte) (int, error) {
+	// An un-sendable destination costs its own packet, never the ones behind it in the batch.
+	written := 0
 	for i, b := range bufs {
-		if err := u.WriteTo(b, addrs[i]); err != nil {
+		if err := u.WriteTo(b, addrs[i]); err == nil {
-			return err
+			written++
+		} else {
+			u.l.Debug("failed to write packet in batch", "udpAddr", addrs[i], "error", err)
 		}
 	}
-	return nil
+	return written, nil
 }
 
 func (u *StdConn) LocalAddr() (netip.AddrPort, error) {

udp/udp_generic.go

@@ -44,13 +44,17 @@ func (u *GenericConn) WriteTo(b []byte, addr netip.AddrPort) error {
 	return err
 }
 
-func (u *GenericConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, _ []byte) error {
+func (u *GenericConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, _ []byte) (int, error) {
+	// An un-sendable destination costs its own packet, never the ones behind it in the batch.
+	written := 0
 	for i, b := range bufs {
-		if _, err := u.UDPConn.WriteToUDPAddrPort(b, addrs[i]); err != nil {
+		if _, err := u.UDPConn.WriteToUDPAddrPort(b, addrs[i]); err == nil {
-			return err
+			written++
+		} else {
+			u.l.Debug("failed to write packet in batch", "udpAddr", addrs[i], "error", err)
 		}
 	}
-	return nil
+	return written, nil
 }
 
 func (u *GenericConn) LocalAddr() (netip.AddrPort, error) {

udp/udp_linux.go

@@ -430,7 +430,7 @@ func sendto(fd int, b []byte, addr netip.AddrPort, isV4 bool) error {
 // WriteBatch sends bufs via sendmmsg(2), coalescing same-destination runs
 // into UDP-GSO superpackets when supported. See batchWriter in
 // udp_linux_writebatch.go for the mechanics.
-func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []byte) error {
+func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []byte) (int, error) {
 	return u.bw.WriteBatch(bufs, addrs, ecns)
 }
 

udp/udp_linux_fixes_test.go

@@ -123,9 +123,13 @@ func TestWriteBatchBadFamilyDeliversOthers(t *testing.T) {
 	bufs := [][]byte{[]byte("AAA"), []byte("BBB"), []byte("CCC")}
 	addrs := []netip.AddrPort{good, bad, good}
 
-	if err := sender.WriteBatch(bufs, addrs, nil); err != nil {
+	n, err := sender.WriteBatch(bufs, addrs, nil)
+	if err != nil {
 		t.Fatalf("WriteBatch returned error, want nil (bad dest should be isolated): %v", err)
 	}
+	if n != 2 {
+		t.Errorf("WriteBatch wrote %d packets, want 2 of 3 (the bad-family dest is the only casualty)", n)
+	}
 
 	got := map[string]bool{}
 	rx.SetReadDeadline(time.Now().Add(2 * time.Second))
@@ -184,7 +188,7 @@ func TestWriteBatchOuterTOSToV4Mapped(t *testing.T) {
 	dst := netip.AddrPortFrom(netip.AddrFrom4([4]byte{127, 0, 0, 1}), uint16(rxPort))
 	const wantECN = byte(0x02) // ECT(0)
 
-	if err := sender.WriteBatch([][]byte{[]byte("tos-probe")}, []netip.AddrPort{dst}, []byte{wantECN}); err != nil {
+	if _, err := sender.WriteBatch([][]byte{[]byte("tos-probe")}, []netip.AddrPort{dst}, []byte{wantECN}); err != nil {
 		t.Fatalf("WriteBatch: %v", err)
 	}
 
@@ -231,3 +235,50 @@ func TestWriteBatchOuterTOSToV4Mapped(t *testing.T) {
 		t.Logf("verified: v4 receiver saw outer TOS 0x%02x (ECN=0x%02x) from dual-stack sender", gotTOS, gotTOS&0x03)
 	}
 }
+
+// TestWriteBatchUnreachableDestDeliversOthers is the sendmmsg-fallback twin of
+// TestWriteBatchBadFamilyDeliversOthers. A destination the kernel refuses outright (240.0.0.0/4 is reserved, so
+// sendto returns EINVAL) makes sendmmsg fail for the whole chunk; the per-packet replay must then still deliver
+// every other packet rather than abandoning the batch at the first failure.
+func TestWriteBatchUnreachableDestDeliversOthers(t *testing.T) {
+	rx, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)})
+	if err != nil {
+		t.Skipf("cannot open v4 receiver (sandbox?): %v", err)
+	}
+	defer rx.Close()
+	rxPort := rx.LocalAddr().(*net.UDPAddr).Port
+
+	c, err := NewListener(testLogger(), netip.MustParseAddr("127.0.0.1"), 0, false, 1)
+	if err != nil {
+		t.Skipf("cannot open v4 sender (sandbox?): %v", err)
+	}
+	defer c.Close()
+	sender := c.(*StdConn)
+
+	good := netip.AddrPortFrom(netip.AddrFrom4([4]byte{127, 0, 0, 1}), uint16(rxPort))
+	bad := netip.MustParseAddrPort("240.0.0.1:9999") // reserved space, the kernel refuses it
+
+	bufs := [][]byte{[]byte("P0"), []byte("P1"), []byte("BAD"), []byte("P3"), []byte("P4")}
+	addrs := []netip.AddrPort{good, good, bad, good, good}
+
+	// The bad destination is reported, but only after every other packet has been attempted.
+	if _, err := sender.WriteBatch(bufs, addrs, nil); err == nil {
+		t.Log("WriteBatch returned nil; kernel accepted the reserved address, delivery assertions still apply")
+	}
+
+	got := map[string]bool{}
+	rx.SetReadDeadline(time.Now().Add(2 * time.Second))
+	buf := make([]byte, 64)
+	for i := 0; i < 4; i++ {
+		n, _, rerr := rx.ReadFromUDPAddrPort(buf)
+		if rerr != nil {
+			t.Fatalf("expected 4 delivered packets, read #%d failed: %v (got so far: %v)", i+1, rerr, got)
+		}
+		got[string(buf[:n])] = true
+	}
+	for _, want := range []string{"P0", "P1", "P3", "P4"} {
+		if !got[want] {
+			t.Errorf("packet %s was not delivered; delivered set = %v", want, got)
+		}
+	}
+}

udp/udp_linux_writebatch.go

@@ -194,12 +194,15 @@ func parseRelease(r string) (major, minor int) {
 // sendmmsg returns an error AND zero entries went out we fall back to
 // per-packet sendto for that chunk so the caller still gets best-effort
 // delivery; on a partial-success error we just replay the remainder.
-func (w *batchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []byte) error {
+//
+// Returns the number of packets that reached the wire. An error means the call
+// itself failed; a short count means specific destinations were undeliverable.
+func (w *batchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []byte) (int, error) {
 	if len(bufs) != len(addrs) {
-		return fmt.Errorf("WriteBatch: len(bufs)=%d != len(addrs)=%d", len(bufs), len(addrs))
+		return 0, fmt.Errorf("WriteBatch: len(bufs)=%d != len(addrs)=%d", len(bufs), len(addrs))
 	}
 	if ecns != nil && len(ecns) != len(bufs) {
-		return fmt.Errorf("WriteBatch: len(ecns)=%d != len(bufs)=%d", len(ecns), len(bufs))
+		return 0, fmt.Errorf("WriteBatch: len(ecns)=%d != len(bufs)=%d", len(ecns), len(bufs))
 	}
 
 	// Callers deliver same-destination packets contiguously and in counter
@@ -207,6 +210,10 @@ func (w *batchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []b
 	// sorting pass measurably hurt throughput in microbenchmarks while
 	// providing no observed reordering benefit.
 
+	// A destination the kernel rejects costs its own packet, never the ones around it. We count what actually made
+	// it out rather than returning an error, since the caller is the only one that knows whether a shortfall matters.
+	written := 0
+
 	i := 0
 sendChunks:
 	for i < len(bufs) {
@@ -248,8 +255,10 @@ sendChunks:
 				// never the batch. (Same fallback the zero-sent sendmmsg path
 				// below uses, extended to cover the misaddressed packet.)
 				for k := baseI; k <= i; k++ {
-					if werr := sendto(w.fd, bufs[k], addrs[k], w.isV4); werr != nil && k != i {
+					if werr := sendto(w.fd, bufs[k], addrs[k], w.isV4); werr == nil {
-						return werr
+						written++
+					} else {
+						w.l.Debug("failed to write packet in batch", "udpAddr", addrs[k], "error", werr)
 					}
 				}
 				i++
@@ -278,7 +287,7 @@ sendChunks:
 		}
 
 		if entry == 0 {
-			return fmt.Errorf("sendmmsg: no progress")
+			return written, fmt.Errorf("sendmmsg: no progress")
 		}
 
 		sent, serr := w.sendmmsg(entry)
@@ -313,21 +322,25 @@ sendChunks:
 				"gso", w.gsoSupported,
 			)
 			for k := baseI; k < i; k++ {
-				if werr := sendto(w.fd, bufs[k], addrs[k], w.isV4); werr != nil {
+				if werr := sendto(w.fd, bufs[k], addrs[k], w.isV4); werr == nil {
-					return werr
+					written++
+				} else {
+					w.l.Debug("failed to write packet in batch", "udpAddr", addrs[k], "error", werr)
 				}
 			}
 			continue
 		}
 		if sent == 0 {
-			return fmt.Errorf("sendmmsg made no progress")
+			return written, fmt.Errorf("sendmmsg made no progress")
 		}
 		// Rewind i to the end of the last successfully sent entry. For a
 		// full-success send this leaves i unchanged; for a partial send it
-		// replays the remainder on the next outer-loop iteration.
+		// replays the remainder on the next outer-loop iteration. A single
+		// entry can carry a whole GSO run, so count packets, not entries.
+		written += w.entryEnd[sent-1] - baseI
 		i = w.entryEnd[sent-1]
 	}
-	return nil
+	return written, nil
 }
 
 // planRun groups consecutive packets starting at `start` that can be sent as

udp/udp_linux_writebatch_alloc_test.go

@@ -79,11 +79,11 @@ func TestWriteBatchNoAllocs(t *testing.T) {
 				t.Helper()
 				var werr error
 				// Warm-up outside the measured runs.
-				if err := tx.WriteBatch(bufs, addrs, ecns); err != nil {
+				if _, err := tx.WriteBatch(bufs, addrs, ecns); err != nil {
 					t.Fatalf("WriteBatch warm-up: %v", err)
 				}
 				allocs := testing.AllocsPerRun(100, func() {
-					if err := tx.WriteBatch(bufs, addrs, ecns); err != nil {
+					if _, err := tx.WriteBatch(bufs, addrs, ecns); err != nil {
 						werr = err
 					}
 				})

udp/udp_rio_windows.go

@@ -317,13 +317,17 @@ func (u *RIOConn) WriteTo(buf []byte, ip netip.AddrPort) error {
 	return winrio.SendEx(u.rq, dataBuffer, 1, nil, addressBuffer, nil, nil, 0, 0)
 }
 
-func (u *RIOConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, _ []byte) error {
+func (u *RIOConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, _ []byte) (int, error) {
+	// An un-sendable destination costs its own packet, never the ones behind it in the batch.
+	written := 0
 	for i, b := range bufs {
-		if err := u.WriteTo(b, addrs[i]); err != nil {
+		if err := u.WriteTo(b, addrs[i]); err == nil {
-			return err
+			written++
+		} else {
+			u.l.Debug("failed to write packet in batch", "udpAddr", addrs[i], "error", err)
 		}
 	}
-	return nil
+	return written, nil
 }
 
 func (u *RIOConn) LocalAddr() (netip.AddrPort, error) {

udp/udp_tester.go

@@ -171,13 +171,16 @@ func (u *TesterConn) WriteTo(b []byte, addr netip.AddrPort) error {
 		return nil
 	}
 }
-func (u *TesterConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, _ []byte) error {
+func (u *TesterConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, _ []byte) (int, error) {
+	written := 0
 	for i, b := range bufs {
-		if err := u.WriteTo(b, addrs[i]); err != nil {
+		if err := u.WriteTo(b, addrs[i]); err == nil {
-			return err
+			written++
+		} else {
+			u.l.Debug("failed to write packet in batch", "udpAddr", addrs[i], "error", err)
 		}
 	}
-	return nil
+	return written, nil
 }
 
 func (u *TesterConn) ListenOut(r EncReader, flush func()) error {