derp/client_test.go
Rama main · réplica de lectura del repositorio original.
// Copyright (c) Tailscale Inc & contributors
// SPDX-License-Identifier: BSD-3-Clause
package derp
import (
"bufio"
"bytes"
"encoding/binary"
"errors"
"fmt"
"io"
"net"
"net/netip"
"os"
"reflect"
"runtime"
"strings"
"sync"
"sync/atomic"
"testing"
"testing/iotest"
"testing/synctest"
"time"
"go4.org/mem"
"tailscale.com/feature/buildfeatures"
"tailscale.com/tstest"
"tailscale.com/tstime"
"tailscale.com/types/key"
"tailscale.com/types/logger"
"tailscale.com/util/clientmetric"
)
type dummyNetConn struct {
net.Conn
}
func (dummyNetConn) SetReadDeadline(time.Time) error { return nil }
func TestClientRecv(t *testing.T) {
tests := []struct {
name string
input []byte
want any
}{
{
name: "ping",
input: []byte{
byte(FramePing), 0, 0, 0, 8,
1, 2, 3, 4, 5, 6, 7, 8,
},
want: PingMessage{1, 2, 3, 4, 5, 6, 7, 8},
},
{
name: "pong",
input: []byte{
byte(FramePong), 0, 0, 0, 8,
1, 2, 3, 4, 5, 6, 7, 8,
},
want: PongMessage{1, 2, 3, 4, 5, 6, 7, 8},
},
{
name: "health_bad",
input: []byte{
byte(FrameHealth), 0, 0, 0, 3,
byte('B'), byte('A'), byte('D'),
},
want: HealthMessage{Problem: "BAD"},
},
{
name: "health_ok",
input: []byte{
byte(FrameHealth), 0, 0, 0, 0,
},
want: HealthMessage{},
},
{
name: "server_restarting",
input: []byte{
byte(FrameRestarting), 0, 0, 0, 8,
0, 0, 0, 1,
0, 0, 0, 2,
},
want: ServerRestartingMessage{
ReconnectIn: 1 * time.Millisecond,
TryFor: 2 * time.Millisecond,
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
c := &Client{
nc: dummyNetConn{},
br: bufio.NewReader(bytes.NewReader(tt.input)),
logf: t.Logf,
clock: &tstest.Clock{},
}
got, err := c.Recv()
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(got, tt.want) {
t.Errorf("got %#v; want %#v", got, tt.want)
}
})
}
}
func TestClientRecvMetrics(t *testing.T) {
if !buildfeatures.HasClientMetrics {
t.Skip("client metrics disabled")
}
metrics := []*clientmetric.Metric{metricClientRecvKeepAlive, metricClientRecvPing, metricClientRecvPong}
for _, tt := range []struct {
name string
frame FrameType
payload []byte
want []int64
wantErr bool
}{
{"keepalive", FrameKeepAlive, nil, []int64{1, 0, 0}, false},
{"ping", FramePing, make([]byte, 8), []int64{0, 1, 0}, false},
{"pong", FramePong, make([]byte, 8), []int64{0, 0, 1}, false},
{"short_ping", FramePing, make([]byte, 7), []int64{0, 0, 0}, true},
{"short_pong", FramePong, make([]byte, 7), []int64{0, 0, 0}, true},
{"truncated_ping", FramePing, nil, []int64{0, 0, 0}, true},
} {
t.Run(tt.name, func(t *testing.T) {
before := make([]int64, len(metrics))
for i, m := range metrics {
before[i] = m.Value()
}
var buf bytes.Buffer
bw := bufio.NewWriter(&buf)
if err := WriteFrame(bw, tt.frame, tt.payload); err != nil {
t.Fatal(err)
}
if tt.name == "truncated_ping" {
buf.Bytes()[4] = 8 // header claims a payload that never arrives
}
c := &Client{nc: dummyNetConn{}, br: bufio.NewReader(&buf), logf: t.Logf}
if _, err := c.Recv(); (err != nil) != tt.wantErr {
t.Fatalf("Recv error = %v; wantErr %v", err, tt.wantErr)
}
for i, m := range metrics {
if got := m.Value() - before[i]; got != tt.want[i] {
t.Errorf("receive metric %d delta = %d; want %d", i, got, tt.want[i])
}
}
})
}
}
func TestClientRecvReadTimeoutMetric(t *testing.T) {
if !buildfeatures.HasClientMetrics {
t.Skip("client metrics disabled")
}
t.Run("read_deadline", func(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
before := metricClientRecvReadTimeout.Value()
nc, peer := net.Pipe()
defer nc.Close()
defer peer.Close()
c := &Client{nc: nc, br: bufio.NewReader(nc)}
start := time.Now()
if _, err := c.Recv(); !errors.Is(err, os.ErrDeadlineExceeded) {
t.Fatalf("Recv error = %v; want read deadline", err)
}
if got := time.Since(start); got != 120*time.Second {
t.Errorf("Recv elapsed = %v; want 120s", got)
}
if got := metricClientRecvReadTimeout.Value() - before; got != 1 {
t.Errorf("read timeout delta = %d; want 1", got)
}
})
})
for _, tt := range []struct {
name string
err error
want int64
}{
{"deadline", os.ErrDeadlineExceeded, 1},
{"wrapped_deadline", fmt.Errorf("read: %w", os.ErrDeadlineExceeded), 1},
{"eof", io.EOF, 0},
{"closed", net.ErrClosed, 0},
{"other_timeout", recvTestTimeoutError{}, 0},
} {
t.Run(tt.name, func(t *testing.T) {
before := metricClientRecvReadTimeout.Value()
c := &Client{nc: dummyNetConn{}, br: bufio.NewReader(iotest.ErrReader(tt.err))}
_, firstErr := c.Recv()
if !errors.Is(firstErr, tt.err) {
t.Fatalf("Recv error = %v; want wrapping %v", firstErr, tt.err)
}
if _, err := c.Recv(); err != firstErr {
t.Fatalf("second Recv error = %v; want sticky error %v", err, firstErr)
}
if got := metricClientRecvReadTimeout.Value() - before; got != tt.want {
t.Errorf("read timeout delta = %d; want %d", got, tt.want)
}
})
}
}
// A generic net.Error timeout must not be attributed to the read deadline.
// In particular, a transport failure is not necessarily a receive-idle expiry.
type recvTestTimeoutError struct{}
func (recvTestTimeoutError) Error() string { return "other timeout" }
func (recvTestTimeoutError) Timeout() bool { return true }
func (recvTestTimeoutError) Temporary() bool { return true }
func TestClientSendPingPongMetrics(t *testing.T) {
if !buildfeatures.HasClientMetrics {
t.Skip("client metrics disabled")
}
for _, frame := range []FrameType{FramePing, FramePong} {
for _, fail := range []bool{false, true} {
t.Run(fmt.Sprintf("frame_%x/fail_%v", frame, fail), func(t *testing.T) {
beforePing, beforePong := metricClientSendPing.Value(), metricClientSendPong.Value()
var buf bytes.Buffer
c := &Client{bw: bufio.NewWriter(&buf)}
if fail {
// A buffered writer retains its first error. Neither a failed
// flush nor an already broken writer should count a send.
pr, pw := io.Pipe()
pr.CloseWithError(io.ErrClosedPipe)
defer pw.Close()
c.bw = bufio.NewWriter(pw)
}
for range 2 {
if err := c.sendPingOrPong(frame, [8]byte{}); (err != nil) != fail {
t.Fatalf("send error = %v; want failure %v", err, fail)
}
}
var wantPing, wantPong int64
if !fail {
if frame == FramePing {
wantPing = 2
} else {
wantPong = 2
}
}
if got := metricClientSendPing.Value() - beforePing; got != wantPing {
t.Errorf("sent ping delta = %d; want %d", got, wantPing)
}
if got := metricClientSendPong.Value() - beforePong; got != wantPong {
t.Errorf("sent pong delta = %d; want %d", got, wantPong)
}
})
}
}
}
// TestClientRecvPeerPresent tests that the client can parse peerPresent
// frames from servers of various eras: old servers that send fewer fields
// than the client knows about, and newer servers that send trailing fields
// the client doesn't know about, which it must ignore. This matters during
// rollouts of new DERP servers, when a region's meshed nodes and watchers
// run a mix of versions.
func TestClientRecvPeerPresent(t *testing.T) {
keyb := bytes.Repeat([]byte{1}, KeyLen)
k := key.NodePublicFromRaw32(mem.B(keyb))
ipPort := []byte{
0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff, 1, 2, 3, 4, // ::ffff:1.2.3.4
0x12, 0x34, // port 4660
}
wantIPPort := netip.MustParseAddrPort("1.2.3.4:4660")
frame := func(fields ...[]byte) []byte {
b := []byte{byte(FramePeerPresent), 0, 0, 0, 0}
for _, f := range fields {
b = append(b, f...)
}
b[4] = byte(len(b) - FrameHeaderLen)
return b
}
tests := []struct {
name string
input []byte
want PeerPresentMessage
}{
{
name: "key_only_from_ancient_server",
input: frame(keyb),
want: PeerPresentMessage{Key: k},
},
{
name: "ip_port_from_old_server",
input: frame(keyb, ipPort),
want: PeerPresentMessage{Key: k, IPPort: wantIPPort},
},
{
name: "flags_from_current_server",
input: frame(keyb, ipPort, []byte{PeerPresentIsRegular}),
want: PeerPresentMessage{Key: k, IPPort: wantIPPort, Flags: PeerPresentIsRegular},
},
{
name: "app_name_from_current_server",
input: frame(keyb, ipPort, []byte{PeerPresentIsRegular}, []byte{3, 'a', 'b', 'c'}),
want: PeerPresentMessage{Key: k, IPPort: wantIPPort, Flags: PeerPresentIsRegular, AppName: "abc"},
},
{
name: "extra_fields_from_newer_server",
// A hypothetical newer server sending fields this client
// doesn't know about. They must be ignored.
input: frame(keyb, ipPort, []byte{PeerPresentIsRegular},
[]byte{3, 'a', 'b', 'c'}, []byte{0xde, 0xad}),
want: PeerPresentMessage{Key: k, IPPort: wantIPPort, Flags: PeerPresentIsRegular, AppName: "abc"},
},
{
name: "truncated_app_name_ignored",
// A buggy or malicious server sending an app name length
// that exceeds the frame.
input: frame(keyb, ipPort, []byte{PeerPresentIsRegular}, []byte{200, 'a', 'b', 'c'}),
want: PeerPresentMessage{Key: k, IPPort: wantIPPort, Flags: PeerPresentIsRegular},
},
{
name: "invalid_app_name_ignored",
input: frame(keyb, ipPort, []byte{PeerPresentIsRegular}, []byte{3, 0x01, 0x02, 0x03}),
want: PeerPresentMessage{Key: k, IPPort: wantIPPort, Flags: PeerPresentIsRegular},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
c := &Client{
nc: dummyNetConn{},
br: bufio.NewReader(bytes.NewReader(tt.input)),
logf: t.Logf,
clock: &tstest.Clock{},
}
m, err := c.Recv()
if err != nil {
t.Fatal(err)
}
got, ok := m.(PeerPresentMessage)
if !ok {
t.Fatalf("message type = %T; want PeerPresentMessage", m)
}
if got != tt.want {
t.Errorf("got %+v; want %+v", got, tt.want)
}
})
}
}
func TestClientSendPing(t *testing.T) {
var buf bytes.Buffer
c := &Client{
bw: bufio.NewWriter(&buf),
}
if err := c.SendPing([8]byte{1, 2, 3, 4, 5, 6, 7, 8}); err != nil {
t.Fatal(err)
}
want := []byte{
byte(FramePing), 0, 0, 0, 8,
1, 2, 3, 4, 5, 6, 7, 8,
}
if !bytes.Equal(buf.Bytes(), want) {
t.Errorf("unexpected output\nwrote: % 02x\n want: % 02x", buf.Bytes(), want)
}
}
func TestClientSendPong(t *testing.T) {
var buf bytes.Buffer
c := &Client{
bw: bufio.NewWriter(&buf),
}
if err := c.SendPong([8]byte{1, 2, 3, 4, 5, 6, 7, 8}); err != nil {
t.Fatal(err)
}
want := []byte{
byte(FramePong), 0, 0, 0, 8,
1, 2, 3, 4, 5, 6, 7, 8,
}
if !bytes.Equal(buf.Bytes(), want) {
t.Errorf("unexpected output\nwrote: % 02x\n want: % 02x", buf.Bytes(), want)
}
}
type countWriter struct {
mu sync.Mutex
writes int
bytes int64
}
func (w *countWriter) Write(p []byte) (n int, err error) {
w.mu.Lock()
defer w.mu.Unlock()
w.writes++
w.bytes += int64(len(p))
return len(p), nil
}
func (w *countWriter) Stats() (writes int, bytes int64) {
w.mu.Lock()
defer w.mu.Unlock()
return w.writes, w.bytes
}
func (w *countWriter) ResetStats() {
w.mu.Lock()
defer w.mu.Unlock()
w.writes, w.bytes = 0, 0
}
func TestClientSendRateLimiting(t *testing.T) {
cw := new(countWriter)
c := &Client{
bw: bufio.NewWriter(cw),
clock: &tstest.Clock{},
}
c.setSendRateLimiter(ServerInfoMessage{})
pkt := make([]byte, 1000)
if err := c.send(key.NodePublic{}, pkt); err != nil {
t.Fatal(err)
}
writes1, bytes1 := cw.Stats()
if writes1 != 1 {
t.Errorf("writes = %v, want 1", writes1)
}
// Flood should all succeed.
cw.ResetStats()
for range 1000 {
if err := c.send(key.NodePublic{}, pkt); err != nil {
t.Fatal(err)
}
}
writes1K, bytes1K := cw.Stats()
if writes1K != 1000 {
t.Logf("writes = %v; want 1000", writes1K)
}
if got, want := bytes1K, bytes1*1000; got != want {
t.Logf("bytes = %v; want %v", got, want)
}
// Set a rate limiter
cw.ResetStats()
c.setSendRateLimiter(ServerInfoMessage{
TokenBucketBytesPerSecond: 1,
TokenBucketBytesBurst: int(bytes1 * 2),
})
for range 1000 {
if err := c.send(key.NodePublic{}, pkt); err != nil {
t.Fatal(err)
}
}
writesLimited, bytesLimited := cw.Stats()
if writesLimited == 0 || writesLimited == writes1K {
t.Errorf("limited conn's write count = %v; want non-zero, less than 1k", writesLimited)
}
if bytesLimited < bytes1*2 || bytesLimited >= bytes1K {
t.Errorf("limited conn's bytes count = %v; want >=%v, <%v", bytesLimited, bytes1K*2, bytes1K)
}
}
// fwdTestConn is a derp.Conn for ForwardPacket tests. Writes go to w,
// Close is counted, and overlapping Write calls are reported, as some
// Conn implementations (net/wsconn) panic on them.
type fwdTestConn struct {
dummyNetConn
t testing.TB
w io.Writer
closes atomic.Int32
writing atomic.Bool
}
func (c *fwdTestConn) Write(p []byte) (int, error) {
if !c.writing.CompareAndSwap(false, true) {
c.t.Error("concurrent writes to the connection")
}
defer c.writing.Store(false)
return c.w.Write(p)
}
func (c *fwdTestConn) Close() error { c.closes.Add(1); return nil }
func (c *fwdTestConn) SetWriteDeadline(time.Time) error { return nil }
// newForwardTestClient returns a Client whose connection writes go to
// w, along with the connection.
func newForwardTestClient(t testing.TB, w io.Writer) (*Client, *fwdTestConn) {
nc := &fwdTestConn{t: t, w: w}
c := &Client{
nc: nc,
bw: bufio.NewWriter(nc),
clock: tstime.StdClock{},
logf: logger.Discard,
}
return c, nc
}
// newForwardTestClientLogging is newForwardTestClient with the
// client's log output captured.
func newForwardTestClientLogging(t testing.TB, w io.Writer) (*Client, *fwdTestConn, *syncBuffer) {
c, nc := newForwardTestClient(t, w)
logs := new(syncBuffer)
c.logf = func(format string, args ...any) {
logs.mu.Lock()
defer logs.mu.Unlock()
fmt.Fprintf(&logs.b, format+"\n", args...)
}
return c, nc, logs
}
type syncBuffer struct {
mu sync.Mutex
b bytes.Buffer
}
func (b *syncBuffer) String() string {
b.mu.Lock()
defer b.mu.Unlock()
return b.b.String()
}
// forwardFrame is the wire encoding of a forward packet frame.
func forwardFrame(src, dst key.NodePublic, payload []byte) []byte {
b := appendFrameHeader(nil, FrameForwardPacket, uint32(KeyLen*2+len(payload)))
b = src.AppendTo(b)
b = dst.AppendTo(b)
return append(b, payload...)
}
// gatedWriter is an io.Writer that records what's written and, while
// gated, blocks each Write until the test sends on gate, so the test
// controls exactly when each write completes. pending reports whether
// a Write is blocked on the gate. The gate is created in the synctest
// bubble that uses it, so blocking on it is durable.
type gatedWriter struct {
mu sync.Mutex
out bytes.Buffer
writes int
gated atomic.Bool
pending atomic.Bool
gate chan struct{}
err error // returned by every write once set
}
func newGatedWriter() *gatedWriter {
return &gatedWriter{gate: make(chan struct{})}
}
func (w *gatedWriter) Write(p []byte) (int, error) {
if w.gated.Load() {
w.pending.Store(true)
<-w.gate
w.pending.Store(false)
}
w.mu.Lock()
defer w.mu.Unlock()
if w.err != nil {
return 0, w.err
}
w.writes++
return w.out.Write(p)
}
// releaseAll lets gated writes complete one at a time until the
// bubble is quiescent with no write pending, and returns how many it
// released. Between releases it waits for the bubble to settle, so
// callers waiting for room have refilled the buffer before the next
// write is let through.
func (w *gatedWriter) releaseAll() (released int) {
for {
synctest.Wait()
if !w.pending.Load() {
return released
}
w.gate <- struct{}{}
released++
}
}
func (w *gatedWriter) bytes() []byte {
w.mu.Lock()
defer w.mu.Unlock()
return bytes.Clone(w.out.Bytes())
}
func (w *gatedWriter) writeCount() int {
w.mu.Lock()
defer w.mu.Unlock()
return w.writes
}
func (w *gatedWriter) setErr(err error) {
w.mu.Lock()
defer w.mu.Unlock()
w.err = err
}
// forwardBufState returns the length and capacity of c's pending
// forward buffer and the capacity of its idle one, which is 0 while
// the flusher holds it.
func forwardBufState(c *Client) (pendingLen, pendingCap, idleCap int) {
c.fwd.mu.Lock()
defer c.fwd.mu.Unlock()
return len(c.fwd.pending), cap(c.fwd.pending), cap(c.fwd.idle)
}
// readForwardFrames decodes b as a sequence of forward packet frames
// from src to dst and returns their payloads.
func readForwardFrames(t *testing.T, b []byte, src, dst key.NodePublic) (payloads [][]byte) {
t.Helper()
br := bufio.NewReader(bytes.NewReader(b))
for {
if _, err := br.Peek(1); err == io.EOF {
break
}
ft, fl, err := ReadFrameHeader(br)
if err != nil {
t.Fatal(err)
}
if ft != FrameForwardPacket {
t.Fatalf("frame type %d; want %d", ft, FrameForwardPacket)
}
body := make([]byte, fl)
if _, err := io.ReadFull(br, body); err != nil {
t.Fatal(err)
}
if !bytes.Equal(body[:KeyLen], src.AppendTo(nil)) || !bytes.Equal(body[KeyLen:2*KeyLen], dst.AppendTo(nil)) {
t.Fatal("wrong keys in frame")
}
payloads = append(payloads, body[2*KeyLen:])
}
return payloads
}
// TestClientForwardPacketDeferredWrite checks that a lone forward's
// frame reaches the connection intact without any further call, and
// that the caller didn't write it itself: ForwardPacket returns while
// the write is still held open.
func TestClientForwardPacketDeferredWrite(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
w := newGatedWriter()
c, nc := newForwardTestClient(t, w)
src, dst := key.NewNode().Public(), key.NewNode().Public()
w.gated.Store(true)
payload := []byte("hello")
if err := c.ForwardPacket(src, dst, LoanBytes(payload)); err != nil {
t.Fatal(err)
}
synctest.Wait()
if !w.pending.Load() {
t.Fatal("no write in progress after ForwardPacket returned")
}
if got := w.releaseAll(); got != 1 {
t.Fatalf("flusher made %d writes; want 1", got)
}
if got, want := w.bytes(), forwardFrame(src, dst, payload); !bytes.Equal(got, want) {
t.Errorf("single frame\n got: % 02x\nwant: % 02x", got, want)
}
if nc.closes.Load() != 0 {
t.Errorf("connection closed %d times", nc.closes.Load())
}
})
}
// TestClientForwardPacketBatching checks that frames forwarded by many
// goroutines while a write is in progress share the following writes,
// and all arrive intact. Writes are gated and let through one at a
// time, with the bubble settled before each, so every write after the
// first carries a full buffer: the number of writes is a function of
// the bytes forwarded and the buffer size, not of scheduling.
func TestClientForwardPacketBatching(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
w := newGatedWriter()
w.gated.Store(true)
c, _ := newForwardTestClient(t, w)
src, dst := key.NewNode().Public(), key.NewNode().Public()
if err := c.ForwardPacket(src, dst, LoanBytes([]byte("held"))); err != nil {
t.Fatal(err)
}
synctest.Wait() // the flusher is in the gated write of that frame alone
if !w.pending.Load() {
t.Fatal("first forward is not in a write")
}
const n = 1000
var wg sync.WaitGroup
for i := range n {
wg.Go(func() {
p := binary.BigEndian.AppendUint32(nil, uint32(i))
if err := c.ForwardPacket(src, dst, LoanBytes(p)); err != nil {
t.Error(err)
}
})
}
// Every goroutine has now either buffered its frame or is
// waiting for room, and nothing has reached the connection.
synctest.Wait()
if got := w.bytes(); len(got) != 0 {
t.Fatalf("%d bytes written before any write was released; want 0", len(got))
}
w.releaseAll()
wg.Wait()
payloads := readForwardFrames(t, w.bytes(), src, dst)
if len(payloads) != n+1 {
t.Fatalf("got %d frames; want %d", len(payloads), n+1)
}
if string(payloads[0]) != "held" {
t.Fatalf("first frame payload %q; want %q", payloads[0], "held")
}
seen := make([]bool, n)
for _, p := range payloads[1:] {
if len(p) != 4 {
t.Fatalf("payload len %d; want 4", len(p))
}
i := binary.BigEndian.Uint32(p)
if seen[i] {
t.Fatalf("frame %d seen twice", i)
}
seen[i] = true
}
// The first write carried the held frame alone; the rest
// each carried as many frames as fit in a buffer.
frameLen := len(forwardFrame(src, dst, make([]byte, 4)))
perBuf := forwardBufSize / frameLen
if got, want := w.writeCount(), 1+(n+perBuf-1)/perBuf; got != want {
t.Errorf("%d writes for %d frames; want %d", got, n+1, want)
}
})
}
// TestClientForwardPacketBufferFull checks backpressure: a caller that
// finds the buffer full while the other buffer is being written waits,
// rather than growing the buffer or dropping the frame, and proceeds
// once the write completes, with every frame delivered in order.
func TestClientForwardPacketBufferFull(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
w := newGatedWriter()
w.gated.Store(true)
c, _ := newForwardTestClient(t, w)
src, dst := key.NewNode().Public(), key.NewNode().Public()
if err := c.ForwardPacket(src, dst, LoanBytes([]byte{0})); err != nil {
t.Fatal(err)
}
synctest.Wait() // the flusher is in the gated write
// Enough frames to overflow the buffer several times over.
payload := make([]byte, 1000)
frameLen := len(forwardFrame(src, dst, payload))
n := 5 * forwardBufSize / frameLen
var done atomic.Bool
go func() {
for i := range n {
payload[0] = byte(i)
if err := c.ForwardPacket(src, dst, LoanBytes(payload)); err != nil {
t.Error(err)
return
}
}
done.Store(true)
}()
synctest.Wait()
if done.Load() {
t.Fatal("caller didn't wait for room while the write was held")
}
if _, pendingCap, _ := forwardBufState(c); pendingCap != forwardBufSize {
t.Errorf("buffer grew to cap %d; want %d", pendingCap, forwardBufSize)
}
w.gated.Store(false)
w.releaseAll() // the one write blocked on the gate
if !done.Load() {
t.Fatal("caller still waiting after the write was released")
}
payloads := readForwardFrames(t, w.bytes(), src, dst)
if len(payloads) != n+1 {
t.Fatalf("got %d frames; want %d", len(payloads), n+1)
}
for i, p := range payloads[1:] {
if len(p) != len(payload) || p[0] != byte(i) {
t.Fatalf("frame %d out of order or corrupt", i)
}
}
})
}
// TestClientForwardPacketOversizedFrame checks that a frame bigger
// than forwardBufSize still goes through whole and in order, that the
// buffer it grew is kept rather than reallocated each time, and that
// a grown buffer still counts as full at forwardBufSize.
func TestClientForwardPacketOversizedFrame(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
w := newGatedWriter()
c, _ := newForwardTestClient(t, w)
src, dst := key.NewNode().Public(), key.NewNode().Public()
var want []byte
send := func(sizes ...int) {
for _, size := range sizes {
p := bytes.Repeat([]byte{byte(size)}, size)
if err := c.ForwardPacket(src, dst, LoanBytes(p)); err != nil {
t.Fatal(err)
}
want = append(want, forwardFrame(src, dst, p)...)
}
synctest.Wait()
}
send(100, 3*forwardBufSize, 1, MaxPacketSize, 2)
if got := w.bytes(); !bytes.Equal(got, want) {
t.Fatalf("output mismatch: got %d bytes, want %d", len(got), len(want))
}
// Each buffer grows at most once, to fit the biggest frame,
// and is then kept: a further round of the same frames
// changes nothing.
send(3*forwardBufSize, MaxPacketSize, 3*forwardBufSize, MaxPacketSize)
_, cap1, idle1 := forwardBufState(c)
send(3*forwardBufSize, MaxPacketSize, 3*forwardBufSize, MaxPacketSize)
if got := w.bytes(); !bytes.Equal(got, want) {
t.Fatalf("output mismatch after repeated rounds: got %d bytes, want %d", len(got), len(want))
}
_, cap2, idle2 := forwardBufState(c)
if cap2 != cap1 || idle2 != idle1 {
t.Errorf("buffer caps changed from %d+%d to %d+%d; want kept", cap1, idle1, cap2, idle2)
}
if maxCap := 2 * (FrameHeaderLen + KeyLen*2 + MaxPacketSize); cap2 > maxCap || idle2 > maxCap {
t.Errorf("buffer caps %d and %d; want at most %d", cap2, idle2, maxCap)
}
// A grown buffer is full at forwardBufSize, not at its
// capacity: hold a write and fill from a goroutine, which
// must end up waiting with no more than that buffered.
w.gated.Store(true)
if err := c.ForwardPacket(src, dst, LoanBytes([]byte{0})); err != nil {
t.Fatal(err)
}
synctest.Wait() // the flusher is in the gated write
payload := make([]byte, 1000)
var done atomic.Bool
go func() {
for range 4 * forwardBufSize / len(payload) {
if err := c.ForwardPacket(src, dst, LoanBytes(payload)); err != nil {
t.Error(err)
return
}
}
done.Store(true)
}()
synctest.Wait()
if done.Load() {
t.Fatal("caller didn't wait for room")
}
if pendingLen, pendingCap, _ := forwardBufState(c); pendingLen > forwardBufSize || pendingCap <= forwardBufSize {
t.Errorf("pending has %d bytes in a buffer of cap %d; want at most %d bytes in a grown buffer", pendingLen, pendingCap, forwardBufSize)
}
w.gated.Store(false)
w.releaseAll()
if !done.Load() {
t.Fatal("caller still waiting after the write was released")
}
})
}
// TestClientForwardPacketWriteError checks that a failed write closes
// the connection, wakes a caller waiting for room with the error,
// returns it from every later call without starting another flusher,
// so a derphttp.Client owner reconnects, and counts and logs what was
// lost.
func TestClientForwardPacketWriteError(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
w := newGatedWriter()
w.gated.Store(true)
c, nc, logs := newForwardTestClientLogging(t, w)
src, dst := key.NewNode().Public(), key.NewNode().Public()
wantErr := errors.New("boom")
frames0, bytes0 := ForwardDroppedFrames.Load(), ForwardDroppedBytes.Load()
// Hold the first write, fill the other buffer so a caller
// waits, then make the held write fail.
if err := c.ForwardPacket(src, dst, LoanBytes([]byte("x"))); err != nil {
t.Fatal(err)
}
synctest.Wait() // the flusher is in the gated write
payload := make([]byte, 1000)
waiterErr := make(chan error, 1)
go func() {
for {
if err := c.ForwardPacket(src, dst, LoanBytes(payload)); err != nil {
waiterErr <- err
return
}
}
}()
synctest.Wait() // the goroutine is waiting for room
w.setErr(wantErr)
w.gated.Store(false)
w.releaseAll()
if got := nc.closes.Load(); got != 1 {
t.Fatalf("connection closed %d times; want 1", got)
}
select {
case err := <-waiterErr:
if !errors.Is(err, wantErr) {
t.Fatalf("waiting caller got %v; want %v", err, wantErr)
}
default:
t.Fatal("waiting caller was not woken by the failed write")
}
for i := range 2 {
if err := c.ForwardPacket(src, dst, LoanBytes([]byte("y"))); !errors.Is(err, wantErr) {
t.Fatalf("call %d after failure err = %v; want %v", i, err, wantErr)
}
}
synctest.Wait()
if got := nc.closes.Load(); got != 1 {
t.Fatalf("connection closed %d times after later calls; want still 1", got)
}
if got := w.writeCount(); got != 0 {
t.Fatalf("%d successful writes after the failure; want 0", got)
}
// Lost: the "x" frame in the failed write, plus the frames
// the waiter had buffered, which filled a buffer.
frameLen := len(forwardFrame(src, dst, payload))
wantFrames := int64(1 + forwardBufSize/frameLen)
wantBytes := int64(len(forwardFrame(src, dst, []byte("x")))) + (wantFrames-1)*int64(frameLen)
if gotFrames, gotBytes := ForwardDroppedFrames.Load()-frames0, ForwardDroppedBytes.Load()-bytes0; gotFrames != wantFrames || gotBytes != wantBytes {
t.Errorf("dropped counters rose by %d frames, %d bytes; want %d, %d", gotFrames, gotBytes, wantFrames, wantBytes)
}
if got := logs.String(); !strings.Contains(got, "dropping") {
t.Errorf("no drop logged; logs: %q", got)
}
})
}
// TestClientForwardPacketOwnerClosed checks that when the owner closes
// the connection under a write in flight, which derphttp.Client does
// on shutdown and reconnect, the failure is handled but not logged as
// an error, since it is routine and the loss is by design.
func TestClientForwardPacketOwnerClosed(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
w := newGatedWriter()
w.gated.Store(true)
c, nc, logs := newForwardTestClientLogging(t, w)
src, dst := key.NewNode().Public(), key.NewNode().Public()
if err := c.ForwardPacket(src, dst, LoanBytes([]byte("x"))); err != nil {
t.Fatal(err)
}
synctest.Wait() // the flusher is in the gated write
w.setErr(net.ErrClosed)
w.gated.Store(false)
w.releaseAll()
if err := c.ForwardPacket(src, dst, LoanBytes([]byte("y"))); !errors.Is(err, net.ErrClosed) {
t.Fatalf("call after owner close err = %v; want %v", err, net.ErrClosed)
}
if got := logs.String(); got != "" {
t.Errorf("owner close was logged: %q", got)
}
if got := nc.closes.Load(); got != 1 {
t.Errorf("connection closed %d times; want 1", got)
}
})
}
// TestClientForwardPacketSerializesConnWrites checks that the client's
// other senders can't write to the connection while a forward buffer
// is being written, since that write happens with wmu released and
// not every Conn tolerates overlapping writes.
//
// This can't run in a synctest bubble: the other sender blocks on a
// mutex, which isn't a durable block. The sleep only gives it time to
// reach the write before the forward's is released; a shorter one
// could let a regression slip past, but can't fail a correct client.
func TestClientForwardPacketSerializesConnWrites(t *testing.T) {
w := newGatedWriter()
w.gated.Store(true)
c, _ := newForwardTestClient(t, w)
src, dst := key.NewNode().Public(), key.NewNode().Public()
fwdDone := make(chan error, 1)
go func() { fwdDone <- c.ForwardPacket(src, dst, LoanBytes([]byte("fwd"))) }()
for !w.pending.Load() {
runtime.Gosched()
}
w.gated.Store(false)
pingDone := make(chan error, 1)
go func() { pingDone <- c.SendPing([8]byte{1}) }()
time.Sleep(10 * time.Millisecond)
if got := w.bytes(); len(got) != 0 {
t.Fatalf("ping wrote %d bytes while the forward write was in progress", len(got))
}
w.gate <- struct{}{}
for _, ch := range []chan error{fwdDone, pingDone} {
if err := <-ch; err != nil {
t.Fatal(err)
}
}
want := append(forwardFrame(src, dst, []byte("fwd")), byte(FramePing), 0, 0, 0, 8, 1, 0, 0, 0, 0, 0, 0, 0)
if got := w.bytes(); !bytes.Equal(got, want) {
t.Errorf("wire order\n got: % 02x\nwant: % 02x", got, want)
}
}