coder/coderd/tailnet_internal_test.go

package coderd

import (
	"bytes"
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"net/http/httptrace"
	"sync/atomic"
	"testing"
	"time"

	"github.com/google/uuid"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/stretchr/testify/require"
	"github.com/tailscale/wireguard-go/device"
	"golang.org/x/xerrors"
	"tailscale.com/derp"
	"tailscale.com/tailcfg"
	"tailscale.com/types/key"

	"cdr.dev/slog/v3"
	"cdr.dev/slog/v3/sloggers/slogjson"
	"github.com/coder/coder/v2/coderd/workspaceapps"
	"github.com/coder/coder/v2/tailnet"
	"github.com/coder/coder/v2/testutil"
	"github.com/coder/quartz"
)

func TestEmbeddedDERPRegionDialer(t *testing.T) {
	t.Parallel()

	srv := derp.NewServer(key.NewNode(), func(format string, args ...any) { t.Logf(format, args...) })
	t.Cleanup(func() { _ = srv.Close() })
	ctx, cancel := context.WithCancel(testutil.Context(t, testutil.WaitShort))
	dial := embeddedDERPRegionDialer(ctx, testutil.Logger(t), srv, prometheus.NewCounter(prometheus.CounterOpts{Name: "derp_connects_total"}))

	require.Nil(t, dial(ctx, &tailcfg.DERPRegion{EmbeddedRelay: false}))

	cancel()
	conn := dial(ctx, &tailcfg.DERPRegion{EmbeddedRelay: true})
	require.NotNil(t, conn, "an embedded region must never fall back to a network dial")
	_, err := conn.Read(make([]byte, 1))
	require.Error(t, err)
}

func TestPollingDERPClient_FirstRecvDoesNotWaitForTick(t *testing.T) {
	t.Parallel()

	ctx := testutil.Context(t, testutil.WaitShort)
	derpMap := &tailcfg.DERPMap{Regions: map[int]*tailcfg.DERPRegion{1: {RegionID: 1}}}
	// The mock clock never advances, so a poll loop that waits for the
	// first tick would block forever.
	client := newPollingDERPClient(func() *tailcfg.DERPMap { return derpMap }, testutil.Logger(t), quartz.NewMock(t))
	defer client.Close()

	got := make(chan *tailcfg.DERPMap, 1)
	go func() {
		dm, err := client.Recv()
		if err != nil {
			t.Error("recv derp map:", err)
			return
		}
		got <- dm
	}()
	require.Equal(t, derpMap, testutil.RequireReceive(ctx, t, got))
}

func TestUnreachableReason(t *testing.T) {
	t.Parallel()

	now := time.Date(2026, 1, 7, 12, 0, 0, 0, time.UTC)
	node := &tailcfg.Node{}
	for _, tc := range []struct {
		name string
		diag tailnet.PeerDiagnostics
		want string
	}{
		{name: "NoNode", diag: tailnet.PeerDiagnostics{}, want: "no_node"},
		{name: "NoHandshake", diag: tailnet.PeerDiagnostics{ReceivedNode: node}, want: "no_handshake"},
		{
			name: "Lost",
			diag: tailnet.PeerDiagnostics{ReceivedNode: node, Lost: true, LastWireguardHandshake: now.Add(-time.Minute)},
			want: "peer_lost",
		},
		{
			name: "Stale",
			diag: tailnet.PeerDiagnostics{ReceivedNode: node, LastWireguardHandshake: now.Add(-device.RejectAfterTime - time.Second)},
			want: "handshake_stale",
		},
		{
			name: "OK",
			diag: tailnet.PeerDiagnostics{ReceivedNode: node, LastWireguardHandshake: now.Add(-time.Minute)},
			want: "handshake_ok",
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()
			require.Equal(t, tc.want, unreachableReason(tc.diag, now))
		})
	}
}

func fieldMap(fields []slog.Field) map[string]any {
	m := make(map[string]any, len(fields))
	for _, f := range fields {
		m[f.Name] = f.Value
	}
	return m
}

func TestUnreachableFields(t *testing.T) {
	t.Parallel()

	t.Run("NoNode", func(t *testing.T) {
		t.Parallel()
		got := fieldMap(unreachableFields(tailnet.PeerDiagnostics{PreferredDERP: 1}))
		require.Equal(t, false, got["peer_node_received"])
		require.Equal(t, 1, got["server_preferred_derp"])
		require.NotContains(t, got, "peer_last_handshake")
		require.NotContains(t, got, "peer_preferred_derp")
		require.NotContains(t, got, "peer_tx_bytes")
		require.NotContains(t, got, "peer_rx_bytes")
	})

	t.Run("NodeReceived", func(t *testing.T) {
		t.Parallel()
		handshake := time.Date(2026, 1, 7, 12, 0, 0, 0, time.UTC)
		got := fieldMap(unreachableFields(tailnet.PeerDiagnostics{
			PreferredDERP:          1,
			ReceivedNode:           &tailcfg.Node{DERP: "127.3.3.40:2"},
			LastWireguardHandshake: handshake,
			TxBytes:                148,
			RxBytes:                0,
		}))
		require.Equal(t, true, got["peer_node_received"])
		require.Equal(t, false, got["peer_lost"])
		require.Equal(t, handshake, got["peer_last_handshake"])
		require.Equal(t, "2", got["peer_preferred_derp"])
		require.EqualValues(t, 148, got["peer_tx_bytes"])
		require.EqualValues(t, 0, got["peer_rx_bytes"])
	})
}

// newTestServerTailnet returns a ServerTailnet with only the fields that
// recordAgentUnreachable uses. read stands in for the tailnet connection.
func newTestServerTailnet(t *testing.T, read func(uuid.UUID) tailnet.PeerDiagnostics) (*ServerTailnet, *prometheus.Registry) {
	t.Helper()
	counter := prometheus.NewCounterVec(prometheus.CounterOpts{Name: "agent_unreachable_total"}, []string{"reason"})
	registry := prometheus.NewRegistry()
	require.NoError(t, registry.Register(counter))
	return &ServerTailnet{
		clock:               quartz.NewReal(),
		agentUnreachable:    counter,
		readPeerDiagnostics: read,
		peerDiagnosticsSlot: make(chan struct{}, 1),
	}, registry
}

func noPeerDiagnostics(uuid.UUID) tailnet.PeerDiagnostics {
	return tailnet.PeerDiagnostics{}
}

func TestRecordAgentUnreachable_DiagnosticsWait(t *testing.T) {
	t.Parallel()

	ctx := testutil.Context(t, testutil.WaitShort)
	readStarted := make(chan struct{}, 2)
	release := make(chan struct{})
	s, registry := newTestServerTailnet(t, func(uuid.UUID) tailnet.PeerDiagnostics {
		readStarted <- struct{}{}
		<-release
		return tailnet.PeerDiagnostics{ReceivedNode: &tailcfg.Node{}}
	})
	// The mock clock never fires the timeout, so the second failure can only
	// finish by getting the slot.
	clock := quartz.NewMock(t)
	s.clock = clock
	trap := clock.Trap().NewTimer("peerDiagnostics")
	defer trap.Close()

	errCh := make(chan error, 2)
	go func() {
		errCh <- s.recordAgentUnreachable(ctx, uuid.New(), time.Second)
	}()
	trap.MustWait(ctx).MustRelease(ctx)
	testutil.RequireReceive(ctx, t, readStarted)

	go func() {
		errCh <- s.recordAgentUnreachable(ctx, uuid.New(), time.Second)
	}()
	trap.MustWait(ctx).MustRelease(ctx)
	close(release)

	for range 2 {
		err := testutil.RequireReceive(ctx, t, errCh)
		var unreachable *workspaceapps.AgentUnreachableError
		require.ErrorAs(t, err, &unreachable)
		require.Equal(t, "no_handshake", fieldMap(unreachable.Fields)["reason"])
	}

	metrics, err := registry.Gather()
	require.NoError(t, err)
	require.True(t, testutil.PromCounterHasValue(t, metrics, 2, "agent_unreachable_total", "no_handshake"))
}

func TestRecordAgentUnreachable_NoRequestLogger(t *testing.T) {
	t.Parallel()

	s, _ := newTestServerTailnet(t, noPeerDiagnostics)
	var logs bytes.Buffer
	s.logger = slog.Make(slogjson.Sink(&logs))

	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	_ = s.recordAgentUnreachable(ctx, uuid.New(), 100*time.Millisecond)
	require.Empty(t, logs.String())

	agentID := uuid.New()
	_ = s.recordAgentUnreachable(testutil.Context(t, testutil.WaitShort), agentID, time.Second)
	var entry struct {
		Level  string         `json:"level"`
		Msg    string         `json:"msg"`
		Fields map[string]any `json:"fields"`
	}
	require.NoError(t, json.Unmarshal(logs.Bytes(), &entry))
	require.Equal(t, "WARN", entry.Level)
	require.Equal(t, "agent is unreachable", entry.Msg)
	require.Equal(t, agentID.String(), entry.Fields["agent_id"])
	require.Equal(t, "no_node", entry.Fields["reason"])
}

func TestRecordAgentUnreachable_ClientCanceled(t *testing.T) {
	t.Parallel()

	s, registry := newTestServerTailnet(t, noPeerDiagnostics)

	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	// A short wait ended by the client is counted but is not an unreachable
	// agent, so the PTY handler logs it at debug like any other dial error.
	err := s.recordAgentUnreachable(ctx, uuid.New(), 100*time.Millisecond)
	require.Error(t, err)
	var unreachable *workspaceapps.AgentUnreachableError
	require.NotErrorAs(t, err, &unreachable)

	// A long wait counts however the context ended.
	err = s.recordAgentUnreachable(ctx, uuid.New(), clientCanceledWait)
	require.ErrorAs(t, err, &unreachable)
	require.Equal(t, "no_node", fieldMap(unreachable.Fields)["reason"])

	metrics, err := registry.Gather()
	require.NoError(t, err)
	require.True(t, testutil.PromCounterHasValue(t, metrics, 1, "agent_unreachable_total", "client_canceled"))
	require.True(t, testutil.PromCounterHasValue(t, metrics, 1, "agent_unreachable_total", "no_node"))
}

func TestReportProxyDialFailure_PooledConn(t *testing.T) {
	t.Parallel()

	s, registry := newTestServerTailnet(t, noPeerDiagnostics)

	req := httptest.NewRequest(http.MethodGet, "http://127.0.0.1", nil)
	s.director(uuid.New(), func(*http.Request) {})(req)
	ctx, cancel := context.WithCancel(req.Context())
	ds := dialStateFromContext(ctx)
	require.NotNil(t, ds)

	// The dial started for this request is still waiting on the agent, but
	// the transport gave the request a pooled connection and the request
	// ended while it was on that connection. The dial goroutine can reach
	// dialContext after GotConn.
	trace := httptrace.ContextClientTrace(ctx)
	trace.GetConn("127.0.0.1:80")
	trace.GotConn(httptrace.GotConnInfo{Reused: true})
	ds.set(dialPhaseAwaitReachable)
	require.True(t, ds.hadConn())
	cancel()
	err := s.reportProxyDialFailure(ctx, uuid.New(), context.Canceled)
	require.Same(t, context.Canceled, err)

	metrics, err := registry.Gather()
	require.NoError(t, err)
	require.Empty(t, metrics)
}

func TestReportProxyDialFailure_RetryDial(t *testing.T) {
	t.Parallel()

	s, registry := newTestServerTailnet(t, noPeerDiagnostics)

	req := httptest.NewRequest(http.MethodGet, "http://127.0.0.1", nil)
	s.director(uuid.New(), func(*http.Request) {})(req)
	ctx, cancel := context.WithCancel(req.Context())
	ds := dialStateFromContext(ctx)
	require.NotNil(t, ds)

	// The pooled connection failed and the transport retried on a new dial,
	// which is still waiting on the agent when the request ends.
	trace := httptrace.ContextClientTrace(ctx)
	trace.GetConn("127.0.0.1:80")
	trace.GotConn(httptrace.GotConnInfo{Reused: true})
	trace.GetConn("127.0.0.1:80")
	ds.set(dialPhaseAwaitReachable)
	require.False(t, ds.hadConn())
	cancel()
	err := s.reportProxyDialFailure(ctx, uuid.New(), context.Canceled)
	require.ErrorIs(t, err, context.Canceled)

	metrics, err := registry.Gather()
	require.NoError(t, err)
	require.True(t, testutil.PromCounterHasValue(t, metrics, 1, "agent_unreachable_total", "client_canceled"))
}

func TestReportProxyDialFailure_DialTimedOut(t *testing.T) {
	t.Parallel()

	s, registry := newTestServerTailnet(t, noPeerDiagnostics)

	// The dial hit proxyDialTimeout before the request ended, so the error
	// already names the unreachable agent and must not be prefixed again.
	ds := &dialState{}
	ds.set(dialPhaseAwaitReachable)
	ctx := context.WithValue(testutil.Context(t, testutil.WaitShort), dialStateKey{}, ds)
	dialErr := xerrors.Errorf("acquire agent conn: %w", errAgentUnreachable)
	err := s.reportProxyDialFailure(ctx, uuid.New(), dialErr)
	require.Same(t, dialErr, err)

	metrics, err := registry.Gather()
	require.NoError(t, err)
	require.True(t, testutil.PromCounterHasValue(t, metrics, 1, "agent_unreachable_total", "no_node"))
}

func TestReportProxyDialFailure_CoordinatorError(t *testing.T) {
	t.Parallel()

	s, registry := newTestServerTailnet(t, noPeerDiagnostics)

	// The request is still waiting and the dial failed before the first
	// ping, so the agent was never asked.
	ds := &dialState{}
	ds.set(dialPhaseAwaitReachable)
	ctx := context.WithValue(testutil.Context(t, testutil.WaitShort), dialStateKey{}, ds)
	dialErr := xerrors.New("acquire agent conn: ensure agent: send failed")
	err := s.reportProxyDialFailure(ctx, uuid.New(), dialErr)
	require.Same(t, dialErr, err)

	metrics, err := registry.Gather()
	require.NoError(t, err)
	require.Empty(t, metrics)
}

func TestRecordAgentUnreachable_DiagnosticsSlot(t *testing.T) {
	t.Parallel()

	ctx := testutil.Context(t, testutil.WaitShort)
	readStarted := make(chan struct{}, 1)
	release := make(chan struct{})
	var reads atomic.Int32
	s, registry := newTestServerTailnet(t, func(uuid.UUID) tailnet.PeerDiagnostics {
		reads.Add(1)
		readStarted <- struct{}{}
		<-release
		return tailnet.PeerDiagnostics{ReceivedNode: &tailcfg.Node{}}
	})
	clock := quartz.NewMock(t)
	s.clock = clock
	trap := clock.Trap().NewTimer("peerDiagnostics")
	defer trap.Close()
	var unreachable *workspaceapps.AgentUnreachableError

	errCh := make(chan error, 1)
	record := func() {
		go func() {
			errCh <- s.recordAgentUnreachable(ctx, uuid.New(), time.Second)
		}()
		trap.MustWait(ctx).MustRelease(ctx)
	}

	// The first read waits out the timeout while the read stays in flight.
	record()
	testutil.RequireReceive(ctx, t, readStarted)
	clock.Advance(peerDiagnosticsTimeout).MustWait(ctx)
	err := testutil.RequireReceive(ctx, t, errCh)
	require.ErrorAs(t, err, &unreachable)
	require.Equal(t, "diagnostics_timeout", fieldMap(unreachable.Fields)["reason"])

	// A second failure during that read waits for the slot, times out, and
	// does not start another read.
	record()
	clock.Advance(peerDiagnosticsTimeout).MustWait(ctx)
	err = testutil.RequireReceive(ctx, t, errCh)
	require.ErrorAs(t, err, &unreachable)
	require.Equal(t, "diagnostics_timeout", fieldMap(unreachable.Fields)["reason"])
	require.EqualValues(t, 1, reads.Load())

	// Once the read returns, the slot frees and later failures read again.
	// The mock clock does not advance, so this call cannot time out.
	close(release)
	record()
	testutil.RequireReceive(ctx, t, readStarted)
	err = testutil.RequireReceive(ctx, t, errCh)
	require.ErrorAs(t, err, &unreachable)
	got := fieldMap(unreachable.Fields)
	require.Equal(t, "no_handshake", got["reason"])
	require.Equal(t, true, got["peer_node_received"])

	metrics, err := registry.Gather()
	require.NoError(t, err)
	require.True(t, testutil.PromCounterHasValue(t, metrics, 2, "agent_unreachable_total", "diagnostics_timeout"))
	require.True(t, testutil.PromCounterHasValue(t, metrics, 1, "agent_unreachable_total", "no_handshake"))
}