From 88872a8433f2fd132925f4e1030e4f33670500b4 Mon Sep 17 00:00:00 2001 From: Nate Brown Date: Mon, 27 Jul 2026 14:43:30 -0500 Subject: [PATCH] Don't fail on the batch at the first error (#1826) --- control_lifecycle_test.go | 4 +- interface.go | 24 ++++++++--- overlay/batch/batch.go | 6 +-- overlay/batch/tx_batch.go | 11 ++++-- overlay/batch/tx_batch_test.go | 12 +++--- udp/conn.go | 13 +++--- udp/udp_darwin.go | 12 ++++-- udp/udp_generic.go | 12 ++++-- udp/udp_linux.go | 2 +- udp/udp_linux_fixes_test.go | 55 +++++++++++++++++++++++++- udp/udp_linux_writebatch.go | 35 ++++++++++------ udp/udp_linux_writebatch_alloc_test.go | 4 +- udp/udp_rio_windows.go | 12 ++++-- udp/udp_tester.go | 11 ++++-- 14 files changed, 155 insertions(+), 58 deletions(-) diff --git a/control_lifecycle_test.go b/control_lifecycle_test.go index 67cae925..73b14b46 100644 --- a/control_lifecycle_test.go +++ b/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 { - return nil +func (c *fakeConn) WriteBatch(bufs [][]byte, _ []netip.AddrPort, _ []byte) (int, error) { + return len(bufs), nil } func (c *fakeConn) ReloadConfig(_ *config.C) {} func (c *fakeConn) SupportsMultipleReaders() bool { return true } diff --git a/interface.go b/interface.go index 06778587..4508e6ad 100644 --- a/interface.go +++ b/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.l.Error("Failed to write outgoing batch", "error", err, "writer", i) - } + f.flushSendBatch(sb, i) } } - if err := sb.Flush(); err != nil { - f.l.Error("Failed to write outgoing batch", "error", err, "writer", i) - } + f.flushSendBatch(sb, 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) diff --git a/overlay/batch/batch.go b/overlay/batch/batch.go index 4247033b..217c62f1 100644 --- a/overlay/batch/batch.go +++ b/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. - // Returns an errors.Join of one or more errors. + // Flush emits every queued packet via the underlying batch writer in arrival order and reports how many were + // 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) } diff --git a/overlay/batch/tx_batch.go b/overlay/batch/tx_batch.go index 074a6c1e..a78943a5 100644 --- a/overlay/batch/tx_batch.go +++ b/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 } diff --git a/overlay/batch/tx_batch_test.go b/overlay/batch/tx_batch_test.go index 454011dc..d314784a 100644 --- a/overlay/batch/tx_batch_test.go +++ b/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 { diff --git a/udp/conn.go b/udp/conn.go index 37277054..dba84c0a 100644 --- a/udp/conn.go +++ b/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 - // may observe a partial send if some packets went out before the error. - WriteBatch(bufs [][]byte, addrs []netip.AddrPort, outerECNs []byte) error + // cmsg; other backends ignore it. + // + // 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 { - return nil +func (NoopConn) WriteBatch(bufs [][]byte, _ []netip.AddrPort, _ []byte) (int, error) { + return len(bufs), nil } func (NoopConn) ReloadConfig(_ *config.C) { return diff --git a/udp/udp_darwin.go b/udp/udp_darwin.go index 67f4b11e..1ef23d31 100644 --- a/udp/udp_darwin.go +++ b/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 { - return err + if err := u.WriteTo(b, addrs[i]); err == nil { + 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) { diff --git a/udp/udp_generic.go b/udp/udp_generic.go index 0c254906..a7c83e92 100644 --- a/udp/udp_generic.go +++ b/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 { - return err + if _, err := u.UDPConn.WriteToUDPAddrPort(b, addrs[i]); err == nil { + 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) { diff --git a/udp/udp_linux.go b/udp/udp_linux.go index 9ee8e5fc..d47e4b23 100644 --- a/udp/udp_linux.go +++ b/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) } diff --git a/udp/udp_linux_fixes_test.go b/udp/udp_linux_fixes_test.go index b3a59d2d..8fc5f92a 100644 --- a/udp/udp_linux_fixes_test.go +++ b/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) + } + } +} diff --git a/udp/udp_linux_writebatch.go b/udp/udp_linux_writebatch.go index d56e2362..008ed044 100644 --- a/udp/udp_linux_writebatch.go +++ b/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 { - return werr + if werr := sendto(w.fd, bufs[k], addrs[k], w.isV4); werr == nil { + 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 { - return werr + if werr := sendto(w.fd, bufs[k], addrs[k], w.isV4); werr == nil { + 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 diff --git a/udp/udp_linux_writebatch_alloc_test.go b/udp/udp_linux_writebatch_alloc_test.go index 072eced9..d0b5dc4a 100644 --- a/udp/udp_linux_writebatch_alloc_test.go +++ b/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 } }) diff --git a/udp/udp_rio_windows.go b/udp/udp_rio_windows.go index a95ad3d0..a6f097dc 100644 --- a/udp/udp_rio_windows.go +++ b/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 { - return err + if err := u.WriteTo(b, addrs[i]); err == nil { + 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) { diff --git a/udp/udp_tester.go b/udp/udp_tester.go index b3f83116..aa6cd570 100644 --- a/udp/udp_tester.go +++ b/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 { - return err + if err := u.WriteTo(b, addrs[i]); err == nil { + 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 {