routewatch/internal/routewatch/asnfetcher_internal_test.go

package routewatch

import (
	"bufio"
	"context"
	"errors"
	"io"
	"net"
	"sync"
	"syscall"
	"testing"
	"time"

	"git.eeqj.de/sneak/routewatch/internal/database"
	"git.eeqj.de/sneak/routewatch/internal/logger"
)

// The WHOIS servers a lookup asks: IANA first, then the RIR IANA refers it
// to.
const (
	ianaWHOISServer = "whois.iana.org:43"
	ripeWHOISServer = "whois.ripe.net:43"
)

// ianaRefersToRIPE is IANA's answer for an AS in a block it gave to RIPE.
const ianaRefersToRIPE = `refer:        whois.ripe.net

as-block:     64496-64511
organisation: Assigned by RIPE NCC
changed:      2002-09-01
source:       IANA
`

// ripeAccessDenied is RIPE's answer to a host it no longer serves.
const ripeAccessDenied = `% This is the RIPE Database query service.
%ERROR:201: access denied for 192.0.2.1
`

// ianaPrivateUseBlock is IANA's answer for a private-use AS: its block and
// no referral.
const ianaPrivateUseBlock = `as-block:     64512-65534
organisation: Reserved for Private Use
source:       IANA
`

// newFetcherWithWHOISAnswers returns an ASN fetcher on a new database that
// holds testASNA, never looked up. The fetcher's WHOIS client connects to a
// fake server for each address in answers, which reads the query and sends
// answers[address]; connecting to any other address is refused.
func newFetcherWithWHOISAnswers(
	t *testing.T, answers map[string]string,
) (*ASNFetcher, *database.Database) {
	t.Helper()

	db := newTestDatabase(t)

	_, err := db.GetOrCreateASN(testASNA, time.Now())
	if err != nil {
		t.Fatalf("GetOrCreateASN: %v", err)
	}

	var servers sync.WaitGroup

	t.Cleanup(servers.Wait)

	f := NewASNFetcher(db, logger.New().Logger)
	f.whoisClient.DialContext = func(
		_ context.Context, _, address string,
	) (net.Conn, error) {
		answer, ok := answers[address]
		if !ok {
			return nil, syscall.ECONNREFUSED
		}

		client, server := net.Pipe()

		servers.Go(func() {
			defer func() { _ = server.Close() }()

			_, _ = bufio.NewReader(server).ReadString('\n')
			_, _ = io.WriteString(server, answer)
		})

		return client, nil
	}

	return f, db
}

// TestFetchAndUpdateKeepsTheRowWhenTheLookupFails looks up an AS IANA refers
// to RIPE when RIPE's server refuses the connection, and when it answers only
// with an error. The AS must stay never looked up, and the fetcher must back
// off.
func TestFetchAndUpdateKeepsTheRowWhenTheLookupFails(t *testing.T) {
	t.Parallel()

	for name, answers := range map[string]map[string]string{
		"connection refused": {ianaWHOISServer: ianaRefersToRIPE},
		"only an error": {
			ianaWHOISServer: ianaRefersToRIPE,
			ripeWHOISServer: ripeAccessDenied,
		},
	} {
		t.Run(name, func(t *testing.T) {
			t.Parallel()

			f, db := newFetcherWithWHOISAnswers(t, answers)

			if f.fetchAndUpdate(t.Context(), testASNA) {
				t.Error("fetchAndUpdate reported success")
			}

			stats, err := db.GetWHOISStats(t.Context(), whoisStaleThreshold)
			if err != nil {
				t.Fatalf("GetWHOISStats: %v", err)
			}

			if stats.NeverFetched != 1 {
				t.Errorf("never looked up = %d, want 1: the failed lookup "+
					"was stored", stats.NeverFetched)
			}

			want := baseInterval * backoffNumerator / backoffDenominator

			got := f.GetStats()
			if got.CurrentInterval != want || got.ConsecutiveFails != 1 {
				t.Errorf("interval %v after %d failures, want %v after 1",
					got.CurrentInterval, got.ConsecutiveFails, want)
			}
		})
	}
}

// TestRecordStepsTheInterval has the fetcher fail from its base interval of
// 15 seconds, then succeed. Each failure makes the interval 1.5 times as
// long, up to one minute, and each success halves it, down to one second.
func TestRecordStepsTheInterval(t *testing.T) {
	t.Parallel()

	f := NewASNFetcher(newMockStore(), logger.New().Logger)

	for i, want := range []time.Duration{
		22500 * time.Millisecond,
		33750 * time.Millisecond,
		50625 * time.Millisecond,
		time.Minute,
		time.Minute,
	} {
		f.recordFailure()

		if got := f.getInterval(); got != want {
			t.Errorf("interval after %d failures = %v, want %v",
				i+1, got, want)
		}
	}

	for i, want := range []time.Duration{
		30 * time.Second,
		15 * time.Second,
		7500 * time.Millisecond,
		3750 * time.Millisecond,
		1875 * time.Millisecond,
		time.Second,
		time.Second,
	} {
		f.recordSuccess()

		if got := f.getInterval(); got != want {
			t.Errorf("interval after %d successes = %v, want %v",
				i+1, got, want)
		}
	}
}

// TestFetchAndUpdateMarksAnASWithoutReferral looks up a private-use AS, which
// IANA refers to no RIR. The AS must not be looked up again until it is
// stale, and the lookup counts as neither a success nor a failure.
func TestFetchAndUpdateMarksAnASWithoutReferral(t *testing.T) {
	t.Parallel()

	f, db := newFetcherWithWHOISAnswers(t,
		map[string]string{ianaWHOISServer: ianaPrivateUseBlock})

	f.fetchAndUpdate(t.Context(), testASNA)

	asn, err := db.GetNextStaleASN(t.Context(), whoisStaleThreshold)
	if !errors.Is(err, database.ErrNoStaleASN) {
		t.Errorf("GetNextStaleASN = %d, %v; want ErrNoStaleASN", asn, err)
	}

	got := f.GetStats()
	if got.CurrentInterval != baseInterval || got.SuccessesLastHour != 0 ||
		got.ErrorsLastHour != 0 {
		t.Errorf("interval %v, %d successes, %d errors; want %v, 0, 0",
			got.CurrentInterval, got.SuccessesLastHour, got.ErrorsLastHour,
			baseInterval)
	}
}

// TestASNFetcherStopReturnsByItsDeadline starts a fetcher, has it look up an
// AS whose WHOIS server does not answer until the test ends, and stops it with
// a short deadline while that lookup is in progress. Stop must return all the
// same instead of waiting for the lookup.
func TestASNFetcherStopReturnsByItsDeadline(t *testing.T) {
	t.Parallel()

	const stopDeadline = 100 * time.Millisecond

	lookupStarted := make(chan struct{})
	release := make(chan struct{})

	f := NewASNFetcher(newMockStore(), logger.New().Logger)
	f.whoisClient.DialContext = func(
		context.Context, string, string,
	) (net.Conn, error) {
		close(lookupStarted)
		<-release

		return nil, syscall.ECONNREFUSED
	}

	f.Start(t.Context())
	f.QueueImmediate(testASNA)

	select {
	case <-lookupStarted:
	case <-time.After(handleMessageTimeout):
		t.Fatal("the fetcher did not start the lookup")
	}

	ctx, cancel := context.WithTimeout(t.Context(), stopDeadline)
	defer cancel()

	if !returnsInTime(func() { f.Stop(ctx) }) {
		t.Fatalf("Stop with a deadline of %s did not return while a lookup "+
			"was in progress", stopDeadline)
	}

	close(release)

	select {
	case <-f.done:
	case <-time.After(handleMessageTimeout):
		t.Fatal("the fetcher goroutine did not return after the lookup ended")
	}
}

// TestRecordKeepsOnlyTheStatsWindow records a success and a failure every
// second, by a clock the test sets, far longer than statsWindow, and never
// asks for the stats. The fetcher must keep only the times within the last
// statsWindow.
func TestRecordKeepsOnlyTheStatsWindow(t *testing.T) {
	t.Parallel()

	const (
		records = 100_000
		step    = time.Second
		want    = int(statsWindow / step)
	)

	clock := time.Now()

	f := NewASNFetcher(newMockStore(), logger.New().Logger)
	f.now = func() time.Time { return clock }

	for range records {
		clock = clock.Add(step)

		f.recordSuccess()
		f.recordFailure()
	}

	if len(f.successTimes) != want || len(f.errorTimes) != want {
		t.Errorf("%d success and %d error times kept, want %d of each, "+
			"those of the last %s", len(f.successTimes), len(f.errorTimes),
			want, statsWindow)
	}
}