mirror of
https://github.com/slackhq/nebula.git
synced 2026-11-15 12:36:52 +01:00
Ryan HuberandClaude Opus 5.5
64bec56404
On darwin, listenIn encrypts into one SendBatch while a sender goroutine writes the other (batch.SendPipeline). The sender takes whatever built up during its last write, so writes overlap tun reads and packets queued behind a write leave in one WriteBatch, which a connected socket sends as one sendmsg_x. Two batches bound memory; a full producer waits for the swap rather than dropping. One producer and one sender keep per-peer order equal to message-counter order. Everywhere else the branch is a false constant and listenIn is unchanged. -tags send_pipeline builds the pipeline on any platform, so its tests and the e2e suite can run through it on linux. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
90 lines
2.3 KiB
Go
90 lines
2.3 KiB
Go
//go:build darwin || send_pipeline
|
|
|
|
package nebula
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"net/netip"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/slackhq/nebula/overlay/tio"
|
|
"github.com/slackhq/nebula/test"
|
|
"github.com/slackhq/nebula/udp"
|
|
"github.com/stretchr/testify/assert"
|
|
"go.uber.org/goleak"
|
|
)
|
|
|
|
// scriptedQueue hands out reads from a script, then fails every Read with err.
|
|
type scriptedQueue struct {
|
|
reads [][]tio.Packet
|
|
err error
|
|
}
|
|
|
|
func (q *scriptedQueue) Read() ([]tio.Packet, error) {
|
|
if len(q.reads) == 0 {
|
|
return nil, q.err
|
|
}
|
|
r := q.reads[0]
|
|
q.reads = q.reads[1:]
|
|
return r, nil
|
|
}
|
|
func (q *scriptedQueue) Write(b []byte) (int, error) { return len(b), nil }
|
|
func (q *scriptedQueue) Close() error { return nil }
|
|
|
|
type countingConn struct {
|
|
udp.NoopConn
|
|
mu sync.Mutex
|
|
batches int
|
|
}
|
|
|
|
func (c *countingConn) WriteBatch(bufs [][]byte, _ []netip.AddrPort) (int, error) {
|
|
c.mu.Lock()
|
|
c.batches++
|
|
c.mu.Unlock()
|
|
return len(bufs), nil
|
|
}
|
|
|
|
// TestListenInPipelinedStops pins that the pipelined reader returns when its queue fails, stopping its sender, and
|
|
// reports the error as fatal only when the interface isn't shutting down.
|
|
func TestListenInPipelinedStops(t *testing.T) {
|
|
for _, closing := range []bool{false, true} {
|
|
t.Run(map[bool]string{false: "fatal", true: "shutdown"}[closing], func(t *testing.T) {
|
|
defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
|
|
readErr := errors.New("tun gone")
|
|
conn := &countingConn{}
|
|
// Packets too short to parse are dropped before the firewall, so this reads without a hostmap.
|
|
junk := tio.Packet{Bytes: []byte{0x45, 0}}
|
|
q := &scriptedQueue{reads: [][]tio.Packet{{junk}, {junk, junk, junk}, {}}, err: readErr}
|
|
shutdown := make(chan struct{})
|
|
f := &Interface{
|
|
ctx: context.Background(),
|
|
l: test.NewLogger(),
|
|
writers: []udp.Conn{conn},
|
|
triggerShutdown: func() { close(shutdown) },
|
|
}
|
|
f.closed.Store(closing)
|
|
|
|
done := make(chan struct{})
|
|
go func() {
|
|
f.listenIn(q, 0)
|
|
close(done)
|
|
}()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("listenIn did not return after its queue failed")
|
|
}
|
|
if closing {
|
|
assert.Nil(t, f.fatalErr.Load())
|
|
} else {
|
|
<-shutdown
|
|
assert.Equal(t, readErr, *f.fatalErr.Load())
|
|
}
|
|
assert.Zero(t, conn.batches, "dropped packets must not be written")
|
|
})
|
|
}
|
|
}
|