diff --git a/control_lifecycle_test.go b/control_lifecycle_test.go index 6c0387b6..7ffbf18a 100644 --- a/control_lifecycle_test.go +++ b/control_lifecycle_test.go @@ -148,7 +148,7 @@ 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(bufs [][]byte, _ []netip.AddrPort) (int, error) { +func (c *fakeConn) WriteBatch(bufs [][]byte, _ []netip.AddrPort, _ func(int, error)) (int, error) { return len(bufs), nil } func (c *fakeConn) ReloadConfig(_ *config.C, _ *config.Reload) {} diff --git a/e2e/dual_stack_roam_test.go b/e2e/dual_stack_roam_test.go index 9511d589..9dedec2f 100644 --- a/e2e/dual_stack_roam_test.go +++ b/e2e/dual_stack_roam_test.go @@ -23,15 +23,10 @@ import ( // // go test -tags e2e_testing -v -run TestDualStackRoamToV4Only ./e2e/ func TestDualStackRoamToV4Only(t *testing.T) { - for _, tc := range []struct { - name string - meSends bool - themSends bool - rebind bool - lhOverV4 bool - }{ + for _, tc := range []dualStackRoamCase{ {name: "both send", meSends: true, themSends: true}, {name: "both send, rebind", meSends: true, themSends: true, rebind: true}, + {name: "both send, two tunnels", meSends: true, themSends: true, simultaneous: true}, {name: "laptop sends", meSends: true}, {name: "peer sends", themSends: true}, {name: "peer sends, rebind", themSends: true, rebind: true}, @@ -39,12 +34,25 @@ func TestDualStackRoamToV4Only(t *testing.T) { } { t.Run(tc.name, func(t *testing.T) { t.Parallel() - fwd, rev := measureDualStackRoam(t, tc.meSends, tc.themSends, tc.rebind, tc.lhOverV4) + fwd, rev := measureDualStackRoam(t, tc) t.Logf("RESULT %-24s laptop->peer %-9v peer->laptop %-9v", tc.name, fmtRecovery(fwd), fmtRecovery(rev)) }) } } +type dualStackRoamCase struct { + name string + // meSends and themSends say who keeps traffic flowing after the move + meSends, themSends bool + // rebind fires RebindUDPServer on the move, as the darwin network change monitor would + rebind bool + // lhOverV4 forms the lighthouse tunnel over IPv4 at the office, so the move does not break it + lhOverV4 bool + // simultaneous has both sides initiate at once at the office, which can leave each side a different primary + // tunnel for the other + simultaneous bool +} + func fmtRecovery(d time.Duration) string { if d == 0 { return "never" @@ -84,7 +92,7 @@ func newDualStackNode(ca cert.Certificate, caKey []byte, name, vpnNet string, v4 // measureDualStackRoam returns how long after the move the first packet made it laptop->peer and peer->laptop, zero // for a direction that never recovered. -func measureDualStackRoam(t *testing.T, meSends, themSends, rebind, lhOverV4 bool) (time.Duration, time.Duration) { +func measureDualStackRoam(t *testing.T, tc dualStackRoamCase) (time.Duration, time.Duration) { t.Helper() ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version2, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{}) @@ -131,16 +139,24 @@ func measureDualStackRoam(t *testing.T, meSends, themSends, rebind, lhOverV4 boo // for a static_host_map that only lists the lighthouse's IPv4 address. officeV6 := func(p *udp.Packet, _ *nebula.Control) router.ExitType { lhTraffic := p.To == lhV4 || p.To == lhV6 || p.From == lhV4 || p.From == lhV6 - if p.To.Addr().Is4() == (lhOverV4 && lhTraffic) { + if p.To.Addr().Is4() == (tc.lhOverV4 && lhTraffic) { return router.KeepRouting } return router.Drop } r.RouteForAllExitFuncOrTimeout(time.Second, officeV6) + // One side at a time, so both sides hold the same single tunnel, unless the case wants them racing me.c.InjectTunPacket(BuildTunUDPPacket(them.vpnAddr, 80, me.vpnAddr, 80, []byte("establish"))) + if !tc.simultaneous { + r.RouteForAllExitFuncOrTimeout(time.Second, officeV6) + } them.c.InjectTunPacket(BuildTunUDPPacket(me.vpnAddr, 80, them.vpnAddr, 80, []byte("establish"))) - r.RouteForAllExitFuncOrTimeout(time.Second*2, officeV6) + r.RouteForAllExitFuncOrTimeout(time.Second, officeV6) + if tc.simultaneous && me.c.GetHostmapIndexCount() < 3 { + // A tunnel to the lighthouse and two to the peer, or the race did not happen + t.Logf("simultaneous handshakes settled on a single tunnel this run") + } drainTun(me.c) drainTun(them.c) @@ -152,7 +168,7 @@ func measureDualStackRoam(t *testing.T, meSends, themSends, rebind, lhOverV4 boo }{ {"laptop->peer", me.c, them.vpnAddr, true}, {"peer->laptop", them.c, me.vpnAddr, true}, - {"laptop->lh", me.c, lh.vpnAddr, !lhOverV4}, + {"laptop->lh", me.c, lh.vpnAddr, !tc.lhOverV4}, } { hi := want.c.GetHostInfoByVpnAddr(want.vpnAddr, false) if hi == nil { @@ -170,7 +186,7 @@ func measureDualStackRoam(t *testing.T, meSends, themSends, rebind, lhOverV4 boo r.AddRoute(hotel.Addr(), hotel.Port(), me.c) start := time.Now() - if rebind { + if tc.rebind { me.c.RebindUDPServer() } @@ -188,10 +204,10 @@ func measureDualStackRoam(t *testing.T, meSends, themSends, rebind, lhOverV4 boo case <-stop: return case <-tick.C: - if meSends { + if tc.meSends { me.c.InjectTunPacket(BuildTunUDPPacket(them.vpnAddr, 80, me.vpnAddr, 80, []byte("retry"))) } - if themSends { + if tc.themSends { them.c.InjectTunPacket(BuildTunUDPPacket(me.vpnAddr, 80, them.vpnAddr, 80, []byte("retry"))) } } @@ -200,7 +216,7 @@ func measureDualStackRoam(t *testing.T, meSends, themSends, rebind, lhOverV4 boo // When only one side has traffic, the other direction recovering is a bonus we still report but do not wait for done := func() bool { - return (!meSends || fwd.Load() != 0) && (!themSends || rev.Load() != 0) + return (!tc.meSends || fwd.Load() != 0) && (!tc.themSends || rev.Load() != 0) } r.RouteForAllExitFuncOrTimeout(time.Second*90, func(p *udp.Packet, _ *nebula.Control) router.ExitType { if p.To == officeV4 || p.To == officeV6Addr { diff --git a/hostmap.go b/hostmap.go index 60505cc2..5f0838c2 100644 --- a/hostmap.go +++ b/hostmap.go @@ -776,7 +776,8 @@ const unreachableProbeInterval = time.Second // after moving to a network without that address family. Sends a test packet to every other remote we know for this // host: the first one that arrives roams the peer over to wherever we are now, and its reply roams us onto that path. func (i *HostInfo) remoteUnreachable(ifce *Interface, dead netip.AddrPort) { - if i.remotes == nil || i.ConnectionState == nil { + if i.remotes == nil || i.ConnectionState == nil || i.GetRemote() != dead { + // Nothing to probe with, or we already moved on from the remote that failed return } diff --git a/inside.go b/inside.go index f79702e6..a26e0031 100644 --- a/inside.go +++ b/inside.go @@ -215,7 +215,7 @@ func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []b return nil } - sendBatch.Commit(toSend, relayHostInfo.GetRemote()) + sendBatch.Commit(toSend, relayHostInfo.GetRemote(), relayHostInfo.localIndexId) return nil }) if err != nil { @@ -233,7 +233,7 @@ func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []b return nil } - sendBatch.Commit(out, remote) + sendBatch.Commit(out, remote, hostinfo.localIndexId) return nil }) if err != nil { diff --git a/interface.go b/interface.go index e229f52a..6ed68ec5 100644 --- a/interface.go +++ b/interface.go @@ -495,7 +495,8 @@ func (f *Interface) listenIn(queue tio.Queue, i int) { // 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. +// debug; here it is only a counter, so one unreachable peer cannot spam a log line per batch. Tunnels whose remote +// we no longer have a route to go looking for another path. func (f *Interface) flushSendBatch(sb *batch.SendBatch, q int) { queued := sb.Len() written, err := sb.Flush() @@ -505,6 +506,23 @@ func (f *Interface) flushSendBatch(sb *batch.SendBatch, q int) { if dropped := queued - written; dropped > 0 { f.metricTxDropped.Inc(int64(dropped)) } + + // Packets are tagged with the local index of the tunnel they belong to. A tunnel's packets tend to fail together, + // so skip straight past a repeat of the one before. + var last batch.Failure + for i, fail := range sb.Failures() { + if i > 0 && fail.Tag == last.Tag && fail.Dst == last.Dst { + continue + } + last = fail + + if !udp.IsUnreachable(fail.Err) { + continue + } + if hostinfo := f.hostMap.QueryIndex(fail.Tag); hostinfo != nil { + hostinfo.remoteUnreachable(f, fail.Dst) + } + } } func (f *Interface) RegisterConfigChangeCallbacks(c *config.C) { diff --git a/overlay/batch/tx_batch.go b/overlay/batch/tx_batch.go index 4f6f7da2..23dfca13 100644 --- a/overlay/batch/tx_batch.go +++ b/overlay/batch/tx_batch.go @@ -6,7 +6,14 @@ 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) (int, error) + WriteBatch(bufs [][]byte, addrs []netip.AddrPort, failed func(i int, err error)) (int, error) +} + +// Failure is a packet the underlay would not send, with the tag it was committed under +type Failure struct { + Tag uint32 + Dst netip.AddrPort + Err error } // SendBatch accumulates encrypted UDP packets and flushes them via WriteBatch. @@ -16,17 +23,29 @@ type SendBatch struct { out batchWriter bufs [][]byte dsts []netip.AddrPort + tags []uint32 arena *Arena + + failures []Failure + // recordFailure is bound once so handing it to WriteBatch does not allocate on every Flush + recordFailure func(i int, err error) } // NewSendBatch makes a SendBatch with batchCap slots and an arenaSize byte buffer for slices to back those slots func NewSendBatch(out batchWriter, batchCap, arenaSize int) *SendBatch { - return &SendBatch{ + b := &SendBatch{ out: out, bufs: make([][]byte, 0, batchCap), dsts: make([]netip.AddrPort, 0, batchCap), + tags: make([]uint32, 0, batchCap), arena: NewArena(arenaSize), } + b.recordFailure = b.failed + return b +} + +func (b *SendBatch) failed(i int, err error) { + b.failures = append(b.failures, Failure{Tag: b.tags[i], Dst: b.dsts[i], Err: err}) } func (b *SendBatch) Reserve(sz int) []byte { @@ -38,22 +57,35 @@ func (b *SendBatch) Reserve(sz int) []byte { // bounding how long the first packet of a large read batch waits. func (b *SendBatch) Len() int { return len(b.bufs) } -func (b *SendBatch) Commit(pkt []byte, dst netip.AddrPort) { +// Commit queues pkt for dst. tag comes back in Failures if the underlay refuses the packet, so the caller can tell +// whose packet it was. +func (b *SendBatch) Commit(pkt []byte, dst netip.AddrPort, tag uint32) { b.bufs = append(b.bufs, pkt) b.dsts = append(b.dsts, dst) + b.tags = append(b.tags, tag) } // 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. +// were undeliverable, Failures says which; the batch is drained either way. func (b *SendBatch) Flush() (int, error) { + clear(b.failures) + b.failures = b.failures[:0] + var err error written := 0 if len(b.bufs) > 0 { - written, err = b.out.WriteBatch(b.bufs, b.dsts) + written, err = b.out.WriteBatch(b.bufs, b.dsts, b.recordFailure) } clear(b.bufs) b.bufs = b.bufs[:0] b.dsts = b.dsts[:0] + b.tags = b.tags[:0] b.arena.Reset() return written, err } + +// Failures returns every packet the last Flush could not send, in the order they were committed. Valid until the +// next Flush. +func (b *SendBatch) Failures() []Failure { + return b.failures +} diff --git a/overlay/batch/tx_batch_test.go b/overlay/batch/tx_batch_test.go index aad16e9c..7e5aa024 100644 --- a/overlay/batch/tx_batch_test.go +++ b/overlay/batch/tx_batch_test.go @@ -1,7 +1,9 @@ package batch import ( + "errors" "net/netip" + "slices" "testing" ) @@ -10,7 +12,7 @@ type fakeBatchWriter struct { addrs []netip.AddrPort } -func (w *fakeBatchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) { +func (w *fakeBatchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, _ func(int, error)) (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)) @@ -34,7 +36,7 @@ func TestSendBatchReserveCommitFlush(t *testing.T) { t.Fatalf("slot %d: cap=%d want 32", i, cap(slot)) } pkt := append(slot[:0], byte(i), byte(i+1), byte(i+2)) - b.Commit(pkt, ap) + b.Commit(pkt, ap, 0) } if _, err := b.Flush(); err != nil { t.Fatalf("Flush: %v", err) @@ -75,7 +77,7 @@ func TestSendBatchSlotsDoNotOverlap(t *testing.T) { for i := range 3 { s := b.Reserve(8) pkt := append(s[:0], byte(0xA0+i), byte(0xB0+i)) - b.Commit(pkt, ap) + b.Commit(pkt, ap, 0) } if _, err := b.Flush(); err != nil { t.Fatalf("Flush: %v", err) @@ -96,11 +98,11 @@ func TestSendBatchGrowPreservesCommitted(t *testing.T) { s1 := b.Reserve(4) pkt1 := append(s1[:0], 0x11, 0x22, 0x33, 0x44) - b.Commit(pkt1, ap) + b.Commit(pkt1, ap, 0) s2 := b.Reserve(8) // exceeds remaining cap, triggers grow pkt2 := append(s2[:0], 0xA, 0xB, 0xC, 0xD, 0xE) - b.Commit(pkt2, ap) + b.Commit(pkt2, ap, 0) // pkt1 must still be intact even though backing reallocated. if pkt1[0] != 0x11 || pkt1[3] != 0x44 { @@ -120,3 +122,68 @@ func TestSendBatchGrowPreservesCommitted(t *testing.T) { t.Errorf("second packet on the wire: %x", fw.bufs[1]) } } + +// failingBatchWriter refuses every packet whose first byte is in refuse +type failingBatchWriter struct { + refuse map[byte]error +} + +func (w *failingBatchWriter) WriteBatch(bufs [][]byte, _ []netip.AddrPort, failed func(int, error)) (int, error) { + written := 0 + for i, b := range bufs { + if err, ok := w.refuse[b[0]]; ok { + failed(i, err) + continue + } + written++ + } + return written, nil +} + +func TestSendBatchReportsFailedTags(t *testing.T) { + errNoRoute := errors.New("no route") + fw := &failingBatchWriter{refuse: map[byte]error{2: errNoRoute, 3: errNoRoute}} + + b := NewSendBatch(fw, 4, 64) + + dsts := []netip.AddrPort{ + netip.MustParseAddrPort("10.0.0.1:4242"), + netip.MustParseAddrPort("[fd00::2]:4242"), + netip.MustParseAddrPort("[fd00::3]:4242"), + netip.MustParseAddrPort("10.0.0.4:4242"), + } + for i, dst := range dsts { + pkt := append(b.Reserve(8)[:0], byte(i+1)) + b.Commit(pkt, dst, uint32(100+i)) + } + + written, err := b.Flush() + if err != nil { + t.Fatalf("Flush: %v", err) + } + if written != 2 { + t.Errorf("written = %d, want 2", written) + } + want := []Failure{{101, dsts[1], errNoRoute}, {102, dsts[2], errNoRoute}} + if got := b.Failures(); !slices.Equal(got, want) { + t.Errorf("failures = %v, want %v", got, want) + } + + // Tags do not leak into the next flush + b.Commit(append(b.Reserve(8)[:0], 2), dsts[0], 7) + if _, err := b.Flush(); err != nil { + t.Fatalf("Flush: %v", err) + } + if want := []Failure{{7, dsts[0], errNoRoute}}; !slices.Equal(b.Failures(), want) { + t.Errorf("second flush failures = %v, want %v", b.Failures(), want) + } + + // A clean flush forgets the last one's failures + b.Commit(append(b.Reserve(8)[:0], 1), dsts[0], 8) + if _, err := b.Flush(); err != nil { + t.Fatalf("Flush: %v", err) + } + if got := b.Failures(); len(got) != 0 { + t.Errorf("clean flush failures = %v, want none", got) + } +} diff --git a/udp/conn.go b/udp/conn.go index 62806569..dfee6253 100644 --- a/udp/conn.go +++ b/udp/conn.go @@ -41,8 +41,9 @@ type Conn interface { // // 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. + // failed, when not nil, is called with the index of every packet a destination cost us and why. // Not safe for concurrent use on the same Conn. - WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) + WriteBatch(bufs [][]byte, addrs []netip.AddrPort, failed func(i int, err error)) (int, error) // ReloadConfig applies the listen settings that can change, r is nil at startup ReloadConfig(c *config.C, r *config.Reload) SupportsMultipleReaders() bool @@ -66,7 +67,7 @@ func (NoopConn) SupportsMultipleReaders() bool { func (NoopConn) WriteTo(_ []byte, _ netip.AddrPort) error { return nil } -func (NoopConn) WriteBatch(bufs [][]byte, _ []netip.AddrPort) (int, error) { +func (NoopConn) WriteBatch(bufs [][]byte, _ []netip.AddrPort, _ func(int, error)) (int, error) { return len(bufs), nil } func (NoopConn) ReloadConfig(_ *config.C, _ *config.Reload) { diff --git a/udp/udp_darwin.go b/udp/udp_darwin.go index 516e5d7b..090217e3 100644 --- a/udp/udp_darwin.go +++ b/udp/udp_darwin.go @@ -139,7 +139,7 @@ func (u *StdConn) WriteTo(b []byte, ap netip.AddrPort) error { } } -func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) { +func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, failed func(i int, err error)) (int, error) { // An un-sendable destination costs its own packet, never the ones behind it in the batch. // TODO: WriteTo maps EWOULDBLOCK to an error, so a full send buffer // silently drops the rest of a burst (linux blocks instead). Poll for @@ -150,6 +150,9 @@ func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) written++ } else { u.l.Debug("failed to write packet in batch", "udpAddr", addrs[i], "error", err) + if failed != nil { + failed(i, err) + } } } return written, nil diff --git a/udp/udp_generic.go b/udp/udp_generic.go index 54f00ad2..e5d82f29 100644 --- a/udp/udp_generic.go +++ b/udp/udp_generic.go @@ -44,7 +44,7 @@ func (u *GenericConn) WriteTo(b []byte, addr netip.AddrPort) error { return err } -func (u *GenericConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) { +func (u *GenericConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, failed func(i int, err error)) (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 { @@ -52,6 +52,9 @@ func (u *GenericConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, er written++ } else { u.l.Debug("failed to write packet in batch", "udpAddr", addrs[i], "error", err) + if failed != nil { + failed(i, err) + } } } return written, nil diff --git a/udp/udp_linux.go b/udp/udp_linux.go index f9987341..93bcb33e 100644 --- a/udp/udp_linux.go +++ b/udp/udp_linux.go @@ -358,8 +358,8 @@ 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) (int, error) { - return u.bw.WriteBatch(bufs, addrs) +func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, failed func(i int, err error)) (int, error) { + return u.bw.WriteBatch(bufs, addrs, failed) } // writeSockaddr encodes addr into buf (which must be at least SizeofSockaddrInet6 bytes). diff --git a/udp/udp_linux_fixes_test.go b/udp/udp_linux_fixes_test.go index ab8ae18f..ac5d7d95 100644 --- a/udp/udp_linux_fixes_test.go +++ b/udp/udp_linux_fixes_test.go @@ -3,6 +3,7 @@ package udp import ( + "errors" "fmt" "log/slog" "net" @@ -96,13 +97,22 @@ func TestWriteBatchBadFamilyDeliversOthers(t *testing.T) { bufs := [][]byte{[]byte("AAA"), []byte("BBB"), []byte("CCC")} addrs := []netip.AddrPort{good, bad, good} - n, err := sender.WriteBatch(bufs, addrs) + var failed []int + n, err := sender.WriteBatch(bufs, addrs, func(i int, err error) { + if !IsUnreachable(err) { + t.Errorf("packet %d failed with %v, want an unreachable error", i, err) + } + failed = append(failed, i) + }) 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) } + if want := []int{1}; !slices.Equal(failed, want) { + t.Errorf("failed reported %v, want %v", failed, want) + } got := map[string]bool{} rx.SetReadDeadline(time.Now().Add(2 * time.Second)) @@ -154,7 +164,7 @@ func TestWriteBatchUnreachableDestDeliversOthers(t *testing.T) { 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); err == nil { + if _, err := sender.WriteBatch(bufs, addrs, nil); err == nil { t.Log("WriteBatch returned nil; kernel accepted the reserved address, delivery assertions still apply") } @@ -376,7 +386,7 @@ func TestWriteBatchPartialSendRewind(t *testing.T) { return accept, nil } - written, err := w.WriteBatch(bufs, addrs) + written, err := w.WriteBatch(bufs, addrs, nil) if err != nil { t.Fatalf("WriteBatch: %v", err) } @@ -429,13 +439,22 @@ func TestWriteBatchSkipUnroutableRunAccounting(t *testing.T) { return accept, nil } - written, err := w.WriteBatch(bufs, addrs) + var failed []int + written, err := w.WriteBatch(bufs, addrs, func(i int, err error) { + if !IsUnreachable(err) { + t.Errorf("packet %d failed with %v, want an unreachable error", i, err) + } + failed = append(failed, i) + }) if err != nil { t.Fatalf("WriteBatch: %v", err) } if written != 4 { t.Errorf("written = %d, want 4 (the unroutable run is the only casualty)", written) } + if want := []int{2}; !slices.Equal(failed, want) { + t.Errorf("failed reported %v, want %v", failed, want) + } wantTags := []byte{1, 2, 4, 5} if len(wire) != len(wantTags) { t.Fatalf("wire got %d packets, want %d (dup or loss around the skip)", len(wire), len(wantTags)) @@ -486,13 +505,23 @@ func TestWriteBatchMidChunkRejectResumes(t *testing.T) { } } - written, err := w.WriteBatch(bufs, addrs) + var failed []int + written, err := w.WriteBatch(bufs, addrs, func(i int, err error) { + if !errors.Is(err, unix.EPERM) { + t.Errorf("packet %d failed with %v, want EPERM", i, err) + } + failed = append(failed, i) + }) if err != nil { t.Fatalf("WriteBatch: %v", err) } if written != 3 { t.Errorf("written = %d, want 3 (B's rejected run is the only casualty)", written) } + // Every packet in B's run is reported, not just the entry + if want := []int{2, 3}; !slices.Equal(failed, want) { + t.Errorf("failed reported %v, want %v", failed, want) + } wantTags := []byte{1, 2, 5} if len(wire) != len(wantTags) { t.Fatalf("wire got %d packets, want %d (dup or loss around the mid-chunk reject)", len(wire), len(wantTags)) @@ -543,7 +572,7 @@ func TestWriteBatchMidChunkEIODisablesGSOWithoutDup(t *testing.T) { } } - written, err := w.WriteBatch(bufs, addrs) + written, err := w.WriteBatch(bufs, addrs, nil) if err != nil { t.Fatalf("WriteBatch: %v", err) } @@ -571,7 +600,7 @@ func TestWriteBatchZeroProgress(t *testing.T) { w.sendFn = func(start, n int) (int, error) { return 0, nil } bufs := [][]byte{make([]byte, 100)} addrs := []netip.AddrPort{netip.MustParseAddrPort("127.0.0.1:4242")} - if _, err := w.WriteBatch(bufs, addrs); err == nil { + if _, err := w.WriteBatch(bufs, addrs, nil); err == nil { t.Fatal("WriteBatch = nil error on zero progress, want error") } } @@ -597,7 +626,7 @@ func TestWriteBatchEIODisablesGSOAndReplays(t *testing.T) { return n, nil } - written, err := w.WriteBatch(bufs, addrs) + written, err := w.WriteBatch(bufs, addrs, nil) if err != nil { t.Fatalf("WriteBatch: %v", err) } @@ -674,7 +703,7 @@ func TestGSOEngagesOnLoopback(t *testing.T) { addrs[i] = dst } - written, err := sc.WriteBatch(bufs, addrs) + written, err := sc.WriteBatch(bufs, addrs, nil) if err != nil { t.Fatalf("WriteBatch: %v", err) } diff --git a/udp/udp_linux_offloads_test.go b/udp/udp_linux_offloads_test.go index 80648ca0..53e4659a 100644 --- a/udp/udp_linux_offloads_test.go +++ b/udp/udp_linux_offloads_test.go @@ -86,7 +86,7 @@ func TestWriteBatchOffloadsDisabledScripted(t *testing.T) { addrs[i] = dst } - written, err := w.WriteBatch(bufs, addrs) + written, err := w.WriteBatch(bufs, addrs, nil) if err != nil { t.Fatalf("WriteBatch: %v", err) } @@ -163,7 +163,7 @@ func TestOffloadsDisabledOnLoopback(t *testing.T) { addrs[i] = dst } - written, err := sc.WriteBatch(bufs, addrs) + written, err := sc.WriteBatch(bufs, addrs, nil) if err != nil { t.Fatalf("WriteBatch: %v", err) } diff --git a/udp/udp_linux_writebatch.go b/udp/udp_linux_writebatch.go index bd846629..a34900d5 100644 --- a/udp/udp_linux_writebatch.go +++ b/udp/udp_linux_writebatch.go @@ -190,8 +190,8 @@ func parseRelease(r string) (major, minor int) { // its packets are dropped and the rest of the chunk resumes in place. // // Returns the number of packets sent. An error means the call itself failed. -// A short count means some destinations were undeliverable. -func (w *batchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) { +// A short count means some destinations were undeliverable, failed hears about each of their packets. +func (w *batchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, failed func(i int, err error)) (int, error) { if len(bufs) != len(addrs) { return 0, fmt.Errorf("WriteBatch: len(bufs)=%d != len(addrs)=%d", len(bufs), len(addrs)) } @@ -233,6 +233,7 @@ func (w *batchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, er if w.l.Enabled(context.Background(), slog.LevelDebug) { w.l.Debug("skipping unroutable batch destination", "udpAddr", addrs[i], "packets", runLen, "error", err) } + reportFailed(failed, i, i+runLen, err) i += runLen continue } @@ -300,12 +301,23 @@ func (w *batchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, er "gso", w.gsoSupported, ) } + reportFailed(failed, w.entryEnd[done]-w.entryPkts[done], w.entryEnd[done], serr) done++ } } return written, nil } +// reportFailed tells failed, if anyone is listening, that every packet in bufs[start:end] was lost to err +func reportFailed(failed func(i int, err error), start, end int, err error) { + if failed == nil { + return + } + for i := start; i < end; i++ { + failed(i, err) + } +} + // planRun returns the length of the run starting at start and its segment // size (len(bufs[start])). A run of length 1 carries no UDP_SEGMENT cmsg // and is sent as a plain datagram; without GSO support planRun always returns 1. diff --git a/udp/udp_linux_writebatch_alloc_test.go b/udp/udp_linux_writebatch_alloc_test.go index 190fd90b..a8a1a6fc 100644 --- a/udp/udp_linux_writebatch_alloc_test.go +++ b/udp/udp_linux_writebatch_alloc_test.go @@ -80,11 +80,11 @@ func TestWriteBatchNoAllocs(t *testing.T) { var werr error // Warm-up outside the measured runs. - if _, err := tx.WriteBatch(bufs, addrs); err != nil { + if _, err := tx.WriteBatch(bufs, addrs, nil); err != nil { t.Fatalf("WriteBatch warm-up: %v", err) } allocs := testing.AllocsPerRun(100, func() { - if _, err := tx.WriteBatch(bufs, addrs); err != nil { + if _, err := tx.WriteBatch(bufs, addrs, nil); err != nil { werr = err } }) diff --git a/udp/udp_rio_windows.go b/udp/udp_rio_windows.go index 36ae2040..3c0b0a38 100644 --- a/udp/udp_rio_windows.go +++ b/udp/udp_rio_windows.go @@ -319,7 +319,7 @@ 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) (int, error) { +func (u *RIOConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, failed func(i int, err error)) (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 { @@ -327,6 +327,9 @@ func (u *RIOConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) written++ } else { u.l.Debug("failed to write packet in batch", "udpAddr", addrs[i], "error", err) + if failed != nil { + failed(i, err) + } } } return written, nil diff --git a/udp/udp_tester.go b/udp/udp_tester.go index e556ce9e..2a7340cf 100644 --- a/udp/udp_tester.go +++ b/udp/udp_tester.go @@ -217,13 +217,16 @@ func (u *TesterConn) WriteTo(b []byte, addr netip.AddrPort) error { return nil } } -func (u *TesterConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) { +func (u *TesterConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, failed func(i int, err error)) (int, error) { written := 0 for i, b := range bufs { 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) + if failed != nil { + failed(i, err) + } } } return written, nil