smallwebwaf/internal/routewatch/routewatch_test.go

package routewatch_test

import (
	"bytes"
	"context"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"net/netip"
	"slices"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"testing/synctest"
	"time"

	"sneak.berlin/go/smallwebwaf/internal/routewatch"
)

const (
	clientAddr   = "192.0.2.1"
	routedAnswer = `{"status":"ok","data":{"results":[` +
		`{"ip":"192.0.2.1","netblock":"192.0.2.0/24","asn":64496}]}}`
	noRouteAnswer = `{"status":"ok","data":{"results":[{"ip":"192.0.2.1"}]}}`
	// slowAnswer is longer than routewatch allows a reverse DNS lookup.
	slowAnswer = 2500 * time.Millisecond
	// waitLimit is longer than the Client allows a query.
	waitLimit = 10 * time.Second
)

// clock is a time the tests move by hand.
type clock struct {
	mu  sync.Mutex
	now time.Time
}

func (c *clock) Now() time.Time {
	c.mu.Lock()
	defer c.mu.Unlock()

	return c.now
}

func (c *clock) advance(by time.Duration) {
	c.mu.Lock()
	defer c.mu.Unlock()

	c.now = c.now.Add(by)
}

// fixture is a Client asking a stand-in for routewatch. queries counts what the
// stand-in was asked, paths holds the last path it was asked for, and log holds
// the process log.
type fixture struct {
	t       *testing.T
	server  *httptest.Server
	client  *routewatch.Client
	clk     *clock
	queries *atomic.Int32
	path    *atomic.Value
	log     *syncBuffer
}

// syncBuffer is a buffer the background queries can log to while a test reads it.
type syncBuffer struct {
	mu  sync.Mutex
	buf bytes.Buffer
}

func (b *syncBuffer) Write(p []byte) (int, error) {
	b.mu.Lock()
	defer b.mu.Unlock()

	return b.buf.Write(p)
}

func (b *syncBuffer) String() string {
	b.mu.Lock()
	defer b.mu.Unlock()

	return b.buf.String()
}

func newFixture(t *testing.T, standIn http.HandlerFunc) *fixture {
	t.Helper()

	f := &fixture{
		t:       t,
		clk:     &clock{now: time.Date(2026, 11, 9, 0, 0, 0, 0, time.UTC)},
		queries: &atomic.Int32{},
		path:    &atomic.Value{},
		log:     &syncBuffer{},
	}

	f.server = httptest.NewServer(http.HandlerFunc(
		func(w http.ResponseWriter, r *http.Request) {
			f.queries.Add(1)
			f.path.Store(r.URL.Path)
			standIn(w, r)
		}))
	t.Cleanup(f.server.Close)

	f.client = routewatch.New(routewatch.Params{
		URL: f.server.URL + "/", Now: f.clk.Now,
		ProcessLog: slog.New(slog.NewTextHandler(f.log, nil)),
	})

	return f
}

func answering(body string) http.HandlerFunc {
	return func(w http.ResponseWriter, _ *http.Request) {
		_, _ = w.Write([]byte(body))
	}
}

// silentURL is where a routewatch that never answers is: a query to it waits,
// without the network, until it is abandoned. TestMain registers it with Go's
// default transport, through which the Client asks.
const silentURL = "silent://routewatch/"

func TestMain(m *testing.M) {
	transport, _ := http.DefaultTransport.(*http.Transport)
	transport.RegisterProtocol("silent", silentRoutewatch{})

	m.Run()
}

// silentRoutewatch is the routewatch at silentURL.
type silentRoutewatch struct{}

// RoundTrip waits until req is abandoned.
func (silentRoutewatch) RoundTrip(req *http.Request) (*http.Response, error) {
	<-req.Context().Done()

	return nil, req.Context().Err()
}

func (f *fixture) route() (routewatch.Route, routewatch.Missing) {
	addr := netip.MustParseAddr(clientAddr)
	client := netip.PrefixFrom(addr, addr.BitLen())

	return f.client.Route(context.Background(), client, addr)
}

// waitFor asks until the answer is why, or, while why is "", until there is a
// route, and returns the route.
func (f *fixture) waitFor(why routewatch.Missing) routewatch.Route {
	f.t.Helper()

	deadline := time.Now().Add(waitLimit)

	for {
		route, missing := f.route()
		if missing == why {
			return route
		}

		if time.Now().After(deadline) {
			f.t.Fatalf("still %q after %s, want %q", missing, waitLimit, why)
		}

		time.Sleep(time.Millisecond)
	}
}

// wantFailureLoggedWithoutTheClient waits for the process log to name a failed
// query, and checks that it does not name the client.
func (f *fixture) wantFailureLoggedWithoutTheClient() {
	f.t.Helper()

	// The failure is noted before it is logged.
	deadline := time.Now().Add(waitLimit)
	for !strings.Contains(f.log.String(), "asking routewatch failed") {
		if time.Now().After(deadline) {
			f.t.Fatalf("process log %q names no failure", f.log.String())
		}

		time.Sleep(time.Millisecond)
	}

	if strings.Contains(f.log.String(), clientAddr) {
		f.t.Errorf("process log %q names the client", f.log.String())
	}
}

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

	f := newFixture(t, answering(routedAnswer))

	if _, missing := f.route(); missing != routewatch.NoAnswerYet {
		t.Errorf("before the answer: %q, want %q", missing, routewatch.NoAnswerYet)
	}

	route := f.waitFor("")

	want := routewatch.Route{Netblock: netip.MustParsePrefix("192.0.2.0/24"), ASN: 64496}
	if route != want {
		t.Errorf("route %+v, want %+v", route, want)
	}

	if got := f.path.Load(); got != "/api/v1/ip/"+clientAddr {
		t.Errorf("asked for path %v", got)
	}
}

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

	f := newFixture(t, func(w http.ResponseWriter, _ *http.Request) {
		time.Sleep(slowAnswer)

		_, _ = w.Write([]byte(routedAnswer))
	})

	f.route()

	route := f.waitFor("")

	want := routewatch.Route{Netblock: netip.MustParsePrefix("192.0.2.0/24"), ASN: 64496}
	if route != want {
		t.Errorf("route %+v, want %+v", route, want)
	}
}

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

	f := newFixture(t, answering(noRouteAnswer))

	f.route()
	f.waitFor(routewatch.NoRoute)

	if got := f.queries.Load(); got != 1 {
		t.Errorf("%d queries, want 1", got)
	}
}

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

	tests := map[string]http.HandlerFunc{
		"500": func(w http.ResponseWriter, _ *http.Request) {
			w.WriteHeader(http.StatusInternalServerError)
		},
		"not JSON":   answering("no"),
		"no results": answering(`{"status":"ok","data":{"results":[]}}`),
		"bad netblock": answering(
			`{"status":"ok","data":{"results":[{"netblock":"nowhere","asn":1}]}}`),
	}
	for name, standIn := range tests {
		t.Run(name, func(t *testing.T) {
			t.Parallel()

			f := newFixture(t, standIn)

			f.route()
			f.waitFor(routewatch.RequestFailed)
			f.wantFailureLoggedWithoutTheClient()
		})
	}
}

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

	f := newFixture(t, answering(routedAnswer))

	// Another test can take a closed server's port at once, so routewatch's
	// address here is a multicast address, to which TCP cannot connect.
	f.client = routewatch.New(routewatch.Params{
		URL: "http://224.0.0.1/", Now: f.clk.Now,
		ProcessLog: slog.New(slog.NewTextHandler(f.log, nil)),
	})

	f.route()
	f.waitFor(routewatch.RequestFailed)
	f.wantFailureLoggedWithoutTheClient()
}

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

	f := newFixture(t, func(w http.ResponseWriter, _ *http.Request) {
		w.WriteHeader(http.StatusInternalServerError)
	})

	f.route()
	f.waitFor(routewatch.RequestFailed)
	f.route()

	if got := f.queries.Load(); got != 1 {
		t.Errorf("%d queries while the failure is kept, want 1", got)
	}

	f.clk.advance(time.Minute)
	f.route()

	deadline := time.Now().Add(waitLimit)
	for f.queries.Load() != 2 {
		if time.Now().After(deadline) {
			t.Fatalf("%d queries after the failure expired, want 2", f.queries.Load())
		}

		time.Sleep(time.Millisecond)
	}
}

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

	// In a synctest bubble, time passes only while every goroutine in it waits,
	// so Route's wait is measured exactly, however busy the host is. routewatch
	// is the one at silentURL: one reached over the network would keep the
	// bubble's clock from moving.
	synctest.Test(t, func(t *testing.T) {
		// Without newFixture's stand-in, which is reached over the network.
		f := &fixture{t: t, log: &syncBuffer{}}
		f.client = routewatch.New(routewatch.Params{
			URL: silentURL, Now: time.Now,
			ProcessLog: slog.New(slog.NewTextHandler(f.log, nil)),
		})

		began := time.Now()
		_, missing := f.route()

		if waited := time.Since(began); missing != routewatch.NoAnswerYet || waited != 0 {
			t.Errorf("Route gave %q after %s while routewatch hangs, want %q at once",
				missing, waited, routewatch.NoAnswerYet)
		}

		f.waitFor(routewatch.RequestFailed)
		f.wantFailureLoggedWithoutTheClient()
	})
}

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

	release := make(chan struct{})
	f := newFixture(t, func(w http.ResponseWriter, _ *http.Request) {
		<-release

		_, _ = w.Write([]byte(routedAnswer))
	})

	for range 50 {
		f.route()
	}

	close(release)
	f.waitFor("")

	for range 50 {
		f.route()
	}

	if got := f.queries.Load(); got != 1 {
		t.Errorf("%d queries for repeated requests, want 1", got)
	}

	f.clk.advance(time.Hour)
	f.waitFor(routewatch.NoAnswerYet)
	f.waitFor("")

	if got := f.queries.Load(); got != 2 {
		t.Errorf("%d queries after the answer expired, want 2", got)
	}
}

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

	f := newFixture(t, answering(routedAnswer))

	f.route()
	f.waitFor("")

	want := []routewatch.Answer{{
		Client:   netip.MustParsePrefix(clientAddr + "/32"),
		Netblock: netip.MustParsePrefix("192.0.2.0/24"), ASN: 64496,
		Answered: f.clk.Now(),
	}}
	if got := f.client.Snapshot(); !slices.Equal(got, want) {
		t.Errorf("snapshot %+v, want %+v", got, want)
	}

	f.clk.advance(time.Hour)

	if got := f.client.Snapshot(); len(got) != 0 {
		t.Errorf("snapshot %+v an hour after the answer, want none", got)
	}
}

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

	f := newFixture(t, answering(routedAnswer))
	now := f.clk.Now()
	kept := routewatch.Answer{
		Client:   netip.MustParsePrefix(clientAddr + "/32"),
		Netblock: netip.MustParsePrefix("198.51.100.0/24"), ASN: 64497,
		Answered: now.Add(-time.Hour + time.Second),
	}
	stale := routewatch.Answer{
		Client:   netip.MustParsePrefix("192.0.2.2/32"),
		Netblock: netip.MustParsePrefix("192.0.2.0/24"), ASN: 64496,
		Answered: now.Add(-time.Hour),
	}

	f.client.Load([]routewatch.Answer{kept, stale})

	if got := f.client.Snapshot(); !slices.Equal(got, []routewatch.Answer{kept}) {
		t.Errorf("kept %+v, want only the answer that came less than an hour ago", got)
	}

	route, missing := f.route()

	want := routewatch.Route{Netblock: kept.Netblock, ASN: kept.ASN}
	if route != want || missing != "" {
		t.Errorf("route %+v (%q), want %+v", route, missing, want)
	}

	if got := f.queries.Load(); got != 0 {
		t.Errorf("%d queries for a client with an answer loaded, want none", got)
	}
}

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

	const maxAnswers = 100000

	f := newFixture(t, answering(routedAnswer))
	now := f.clk.Now()

	// routewatch.json lists the answers by client. Here each came a
	// millisecond before the one listed before it, so the last listed is
	// the one that came longest ago, and the one dropped.
	answers := make([]routewatch.Answer, maxAnswers+1)
	addr := netip.MustParseAddr("10.0.0.0")

	for i := range answers {
		answers[i] = routewatch.Answer{
			Client:   netip.PrefixFrom(addr, addr.BitLen()),
			Answered: now.Add(-time.Duration(i) * time.Millisecond),
		}
		addr = addr.Next()
	}

	f.client.Load(answers)

	got := f.client.Snapshot()
	if len(got) != maxAnswers {
		t.Fatalf("%d answers kept, want %d", len(got), maxAnswers)
	}

	if got[0] != answers[0] || got[maxAnswers-1] != answers[maxAnswers-1] {
		t.Errorf("answers kept from %s to %s, want from %s to %s", got[0].Client,
			got[maxAnswers-1].Client, answers[0].Client, answers[maxAnswers-1].Client)
	}
}