Keep taking messages while a peer's session goes down (closes #216)

The table's PeerDown visits every prefix under the write lock, which
takes over a second on a full table. The prefix handler called it
inline, so its queue filled and the streamer dropped updates.

PeerDown now holds the lock for 10,000 prefixes at a time, then lets
the lock go for as long again. The handler runs PeerDown on a separate
goroutine, one peer at a time. The handler holds back the down peer's
later messages until that PeerDown returns. Then the handler applies
them in order, one at a time, and takes new messages in between. It
holds back at most 20,000 messages. Stop waits for all of this before
its last flush.

Model: opus-5-5

TODO.md

@@ -44,14 +44,6 @@ tested.
   messages in between. It holds back at most 20,000 messages. At that number,
   `HandleMessage` waits until one of them is applied. When `Stop` runs past its
   deadline, it logs how many held-back messages were not applied (closes #216)
-- 2026-11-02: RouteWatch pauses an RIR's or NIC.br's WHOIS server for an hour
-  after the server refuses a query. A refused or reset connection is a refusal.
-  So is an answer with an `%ERROR:201:` or `%ERROR:202:` line. Lookups at the
-  other servers go on. An AS whose RIR is paused is looked up after the pause.
-  IANA is never paused, because a pause of IANA would stop every lookup. A
-  refusal by IANA is an ordinary failure. NIC.br is also paused for an hour
-  after it says RouteWatch has passed NIC.br's query limit. Before, the fetcher
-  asked a server again within seconds of a refusal (closes #218)
 - 2026-11-02: the AS handler never waits for a database write. Its batch holds
   one entry per AS, with the earliest and the latest time of its updates. So the
   batch no longer grows with the feed's rate. It used to hold one entry for each

internal/routewatch/asnfetcher.go

@@ -287,13 +287,6 @@ func (f *ASNFetcher) fetchNextStale(ctx context.Context) {
 // so the AS stays stale. An AS IANA refers to no RIR is stored with no
 // WHOIS data, so it is not looked up again until it is stale; that counts
 // as neither a success nor a failure.
-//
-// The WHOIS client pauses an RIR's or NIC.br's server for whois.PauseTime
-// after the server refuses a query. Lookups at the other servers go on. IANA
-// is never paused, so a refusal by IANA is an ordinary failure. A lookup
-// whose RIR is paused fails with whois.ErrPaused. Such a lookup leaves the
-// AS stale, and counts as neither a success nor a failure. The lookups that
-// QueueImmediate asks for also run through fetchAndUpdate.
 func (f *ASNFetcher) fetchAndUpdate(ctx context.Context, asn int) bool {
 	ctx, cancel := context.WithTimeout(ctx, time.Minute)
 	defer cancel()
@@ -314,13 +307,6 @@ func (f *ASNFetcher) fetchAndUpdate(ctx context.Context, asn int) bool {
 		return false
 	}
 
-	if errors.Is(err, whois.ErrPaused) {
-		f.logger.Info("Skipped WHOIS lookup, the RIR's server is paused",
-			"asn", asn, "error", err)
-
-		return false
-	}
-
 	if err != nil {
 		f.logger.Error("WHOIS lookup failed", "asn", asn, "error", err)
 		f.recordFailure()

internal/routewatch/asnfetcher_internal_test.go

@@ -13,7 +13,6 @@ import (
 
 	"git.eeqj.de/sneak/routewatch/internal/database"
 	"git.eeqj.de/sneak/routewatch/internal/logger"
-	"git.eeqj.de/sneak/routewatch/internal/whois"
 )
 
 // The WHOIS servers a lookup asks: IANA first, then the RIR IANA refers it
@@ -132,109 +131,6 @@ func TestFetchAndUpdateKeepsTheRowWhenTheLookupFails(t *testing.T) {
 	}
 }
 
-// TestFetchAndUpdateWaitsOutAPause looks up an AS at RIPE while RIPE refuses
-// the query. The test sets the WHOIS client's clock. The next lookup comes a
-// second before whois.PauseTime has passed. It sends RIPE nothing and leaves
-// the AS never looked up. It counts as neither a success nor a failure. A
-// lookup at whois.PauseTime asks RIPE again.
-func TestFetchAndUpdateWaitsOutAPause(t *testing.T) {
-	t.Parallel()
-
-	f, db := newFetcherWithWHOISAnswers(t, map[string]string{
-		ianaWHOISServer: ianaRefersToRIPE,
-		ripeWHOISServer: ripeAccessDenied,
-	})
-
-	ripeAsks := 0
-	dial := f.whoisClient.DialContext
-	f.whoisClient.DialContext = func(
-		ctx context.Context, network, address string,
-	) (net.Conn, error) {
-		if address == ripeWHOISServer {
-			ripeAsks++
-		}
-
-		return dial(ctx, network, address)
-	}
-
-	start := time.Date(2026, 11, 2, 0, 0, 0, 0, time.UTC)
-	now := start
-	f.whoisClient.Now = func() time.Time { return now }
-
-	f.fetchAndUpdate(t.Context(), testASNA)
-
-	now = start.Add(whois.PauseTime - time.Second)
-
-	f.fetchAndUpdate(t.Context(), testASNA)
-
-	if ripeAsks != 1 {
-		t.Errorf("RIPE was asked %d times during its pause, want 1", ripeAsks)
-	}
-
-	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", stats.NeverFetched)
-	}
-
-	got := f.GetStats()
-	if got.ConsecutiveFails != 1 || got.ErrorsLastHour != 1 ||
-		got.SuccessesLastHour != 0 {
-		t.Errorf("%d failures in a row, %d errors, %d successes; want 1, 1, 0",
-			got.ConsecutiveFails, got.ErrorsLastHour, got.SuccessesLastHour)
-	}
-
-	now = start.Add(whois.PauseTime)
-
-	f.fetchAndUpdate(t.Context(), testASNA)
-
-	if ripeAsks != 2 {
-		t.Errorf("RIPE was asked %d times after its pause, want 2", ripeAsks)
-	}
-}
-
-// TestFetchAndUpdateKeepsAskingIANA looks up an AS while IANA's server
-// refuses the connection. The test sets the WHOIS client's clock. The next
-// lookup comes a second before whois.PauseTime has passed. IANA is never
-// paused, so that lookup asks IANA again. Each refusal counts as a failure.
-func TestFetchAndUpdateKeepsAskingIANA(t *testing.T) {
-	t.Parallel()
-
-	// With no answers, every server refuses the connection.
-	f, _ := newFetcherWithWHOISAnswers(t, map[string]string{})
-
-	ianaAsks := 0
-	dial := f.whoisClient.DialContext
-	f.whoisClient.DialContext = func(
-		ctx context.Context, network, address string,
-	) (net.Conn, error) {
-		if address == ianaWHOISServer {
-			ianaAsks++
-		}
-
-		return dial(ctx, network, address)
-	}
-
-	start := time.Date(2026, 11, 2, 0, 0, 0, 0, time.UTC)
-	now := start
-	f.whoisClient.Now = func() time.Time { return now }
-
-	f.fetchAndUpdate(t.Context(), testASNA)
-
-	now = start.Add(whois.PauseTime - time.Second)
-
-	f.fetchAndUpdate(t.Context(), testASNA)
-
-	got := f.GetStats()
-	if ianaAsks != 2 || got.ConsecutiveFails != 2 {
-		t.Errorf("IANA was asked %d times, with %d failures in a row; "+
-			"want 2 and 2", ianaAsks, got.ConsecutiveFails)
-	}
-}
-
 // 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.

internal/whois/whois.go

@@ -12,7 +12,6 @@ import (
 	"regexp"
 	"strings"
 	"sync"
-	"syscall"
 	"time"
 	"unicode/utf8"
 
@@ -71,12 +70,12 @@ const nicbrHeader = "% Copyright (c) Nic.br"
 // queries to stop the data being taken in bulk.
 const nicbrLimitReached = "Query rate limit exceeded"
 
-// PauseTime is how long the client sends an RIR's or NIC.br's server no
+// NICbrSkipTime is how long the client stops asking NIC.br after NIC.br
-// query after that server refuses a query. queryRegistry says what a
+// answers that the client is over its query limit. NIC.br publishes
-// refusal is. RIPE refuses an address for good when the address keeps
+// neither the limit nor how long it lasts. An hour outlasts a limit
-// passing RIPE's limit (see noPersonalFlag). An hour outlasts a limit
+// counted per minute or per hour. The ASes looked up meanwhile get
-// counted per minute or per hour.
+// NIC.br's fields when they are next refreshed.
-const PauseTime = time.Hour
+const NICbrSkipTime = time.Hour
 
 // RIR identifiers.
 const (
@@ -126,10 +125,6 @@ var (
 	errAnswerTooLarge = errors.New("WHOIS answer reached the size cap")
 )
 
-// ErrPaused is returned when a lookup needs a server that refused a query
-// less than PauseTime ago. The client sends that server nothing meanwhile.
-var ErrPaused = errors.New("WHOIS server paused after refusing a query")
-
 // ASNInfo contains parsed WHOIS information for an ASN.
 type ASNInfo struct {
 	ASN         int
@@ -161,12 +156,12 @@ type Client struct {
 
 	logger *slog.Logger
 
-	// mu guards pausedUntil.
+	// mu guards nicbrSkipUntil.
 	mu sync.Mutex
 
-	// pausedUntil maps a server, given as host:port, to the time the client
+	// nicbrSkipUntil is when the client next asks NIC.br, after NIC.br
-	// may ask it again.
+	// answered that the client is over its query limit.
-	pausedUntil map[string]time.Time
+	nicbrSkipUntil time.Time
 }
 
 // NewClient creates a new WHOIS client that logs to logger.
@@ -177,29 +172,23 @@ func NewClient(logger *slog.Logger) *Client {
 		DialContext: dialer.DialContext,
 		Now:         time.Now,
 		logger:      logger,
-		pausedUntil: make(map[string]time.Time),
 	}
 }
 
 // LookupASN asks IANA which RIR holds the given ASN, asks that RIR, and
 // returns what the RIR's answer says. When LACNIC's answer is NIC.br's, it
 // returns what NIC.br's own server says instead, unless that query fails or
-// NIC.br is paused, as askNICbr says.
+// is skipped, as askNICbr says.
 //
-// It returns ErrNoReferral when IANA refers the AS to no RIR. Its error
+// It returns ErrNoReferral when IANA refers the AS to no RIR, and another
-// wraps ErrPaused when the RIR is paused. It returns another error in each
+// error when the query to IANA or to the RIR fails, when the RIR refuses
-// of these cases:
+// it, when IANA's answer has neither a referral nor the AS's block, when
-//
+// IANA refers the AS to a server none of the five RIRs run, or when the
-//   - the query to IANA or to the RIR fails
+// answer carries none of ASNInfo's fields.
-//   - the RIR refuses the query
-//   - IANA's answer has neither a referral nor the AS's block
-//   - IANA refers the AS to a server none of the five RIRs run
-//   - the answer carries none of ASNInfo's fields
 func (c *Client) LookupASN(ctx context.Context, asn int) (*ASNInfo, error) {
 	// Query IANA first to find the authoritative RIR
 	query := fmt.Sprintf("AS%d", asn)
 
-	// IANA is never paused, because a pause of IANA would stop every lookup.
 	ianaResp, err := c.query(ctx, whoisServerIANA, query)
 	if err != nil {
 		return nil, fmt.Errorf("IANA query failed: %w", err)
@@ -222,7 +211,7 @@ func (c *Client) LookupASN(ctx context.Context, asn int) (*ASNInfo, error) {
 	}
 
 	// Query the authoritative RIR
-	rirResp, err := c.queryRegistry(ctx, whoisServer, rirQuery(rir, query))
+	rirResp, err := c.queryRIR(ctx, whoisServer, rirQuery(rir, query))
 	if err != nil {
 		return nil, fmt.Errorf("RIR query failed: %w", err)
 	}
@@ -252,19 +241,22 @@ func (c *Client) LookupASN(ctx context.Context, asn int) (*ASNInfo, error) {
 // askNICbr returns lacnicInfo, what LACNIC's answer says, and logs a
 // warning when the query fails, when NIC.br sends the reduced answer for a
 // client over its limit, or when NIC.br's answer carries none of ASNInfo's
-// fields. The reduced answer pauses NIC.br, as a refusal does. While NIC.br
+// fields. After the reduced answer, it returns lacnicInfo without asking
-// is paused, askNICbr returns lacnicInfo without a warning. The ASes looked
+// NIC.br until NICbrSkipTime has passed.
-// up meanwhile get NIC.br's fields when they are next refreshed.
 func (c *Client) askNICbr(
 	ctx context.Context, asQuery string, lacnicInfo *ASNInfo,
 ) *ASNInfo {
 	asn := lacnicInfo.ASN
 
-	answer, err := c.queryRegistry(ctx, whoisServerNICbr, asQuery)
+	c.mu.Lock()
-	if errors.Is(err, ErrPaused) {
+	skipping := c.Now().Before(c.nicbrSkipUntil)
+	c.mu.Unlock()
+
+	if skipping {
 		return lacnicInfo
 	}
 
+	answer, err := c.query(ctx, whoisServerNICbr, asQuery)
 	if err != nil {
 		c.logger.Warn("NIC.br query failed, keeping LACNIC's fields",
 			"asn", asn, "error", err)
@@ -273,11 +265,13 @@ func (c *Client) askNICbr(
 	}
 
 	if strings.Contains(answer, nicbrLimitReached) {
-		c.pause(whoisServerNICbr)
+		c.mu.Lock()
+		c.nicbrSkipUntil = c.Now().Add(NICbrSkipTime)
+		c.mu.Unlock()
 
 		c.logger.Warn("NIC.br answered that the client is over its query "+
-			"limit, keeping LACNIC's fields and pausing NIC.br",
+			"limit, keeping LACNIC's fields and skipping NIC.br",
-			"asn", asn, "pause_for", PauseTime)
+			"asn", asn, "skip_for", NICbrSkipTime)
 
 		return lacnicInfo
 	}
@@ -315,65 +309,25 @@ func rirQuery(rir, asQuery string) string {
 	}
 }
 
-// queryRegistry sends query to the server of an RIR or of NIC.br, and
+// queryRIR sends query to an RIR's server and returns the answer. An answer
-// returns the answer. The client pauses the server for PauseTime when the
+// holding an error line with which the server refuses the query is an
-// server refuses the query. Each of these is a refusal:
+// error, and the error holds that line.
-//
+func (c *Client) queryRIR(
-//   - a refused connection
-//   - a reset connection
-//   - an %ERROR:201: or %ERROR:202: line in the answer
-//
-// A refusal is returned as an error. While the server is paused,
-// queryRegistry sends it nothing and returns an error that wraps ErrPaused.
-func (c *Client) queryRegistry(
 	ctx context.Context, server, query string,
 ) (string, error) {
-	c.mu.Lock()
-	pausedUntil := c.pausedUntil[server]
-	c.mu.Unlock()
-
-	if c.Now().Before(pausedUntil) {
-		return "", fmt.Errorf("%w: %s until %s", ErrPaused, server,
-			pausedUntil.UTC().Format(time.RFC3339))
-	}
-
 	answer, err := c.query(ctx, server, query)
-	if err == nil {
-		err = checkRefusal(answer)
-	}
-
-	if errors.Is(err, errRefused) || errors.Is(err, syscall.ECONNREFUSED) ||
-		errors.Is(err, syscall.ECONNRESET) {
-		c.pause(server)
-	}
-
 	if err != nil {
 		return "", err
 	}
 
-	return answer, nil
-}
-
-// checkRefusal returns an error holding the first line of answer that
-// starts with accessDeniedError or accessLimitError. It returns nil when
-// answer has no such line.
-func checkRefusal(answer string) error {
 	for line := range strings.SplitSeq(answer, "\n") {
 		if strings.HasPrefix(line, accessDeniedError) ||
 			strings.HasPrefix(line, accessLimitError) {
-			return fmt.Errorf("%w: %s", errRefused, line)
+			return "", fmt.Errorf("%w: %s", errRefused, line)
 		}
 	}
 
-	return nil
+	return answer, nil
-}
-
-// pause stops queryRegistry from querying server until PauseTime has passed.
-func (c *Client) pause(server string) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	c.pausedUntil[server] = c.Now().Add(PauseTime)
 }
 
 // query performs a raw WHOIS query to the specified server and returns the

internal/whois/whois_test.go

@@ -718,10 +718,9 @@ func TestLookupASNKeepsLACNICsAnswerWhenNICbrFails(t *testing.T) {
 
 // TestLookupASNSkipsNICbrAfterItsLimit looks up an AS that NIC.br
 // registers three times with one client, while NIC.br answers that the
-// client is over its query limit. The test sets the client's clock. The
+// client is over its query limit. The first lookup asks NIC.br. The
-// first lookup asks NIC.br. The next comes a second before PauseTime has
+// second, a second before NICbrSkipTime has passed, does not. The third,
-// passed, and does not ask NIC.br. The last comes at PauseTime and asks
+// once NICbrSkipTime has passed, asks again. Each keeps LACNIC's fields.
-// NIC.br again. Each keeps LACNIC's fields.
 func TestLookupASNSkipsNICbrAfterItsLimit(t *testing.T) {
 	t.Parallel()
 
@@ -741,8 +740,8 @@ func TestLookupASNSkipsNICbrAfterItsLimit(t *testing.T) {
 		wantAsks int
 	}{
 		{after: 0, wantAsks: 1},
-		{after: whois.PauseTime - time.Second, wantAsks: 1},
+		{after: whois.NICbrSkipTime - time.Second, wantAsks: 1},
-		{after: whois.PauseTime, wantAsks: 2},
+		{after: whois.NICbrSkipTime, wantAsks: 2},
 	} {
 		now = start.Add(step.after)
 
@@ -786,142 +785,6 @@ func TestLookupASNReportsARefusal(t *testing.T) {
 	}
 }
 
-// resetServer starts a TCP server on the loopback address and returns the
-// server's address. The server reads one query on each connection, then
-// resets the connection.
-func resetServer(t *testing.T) string {
-	t.Helper()
-
-	var listenConfig net.ListenConfig
-
-	listener, err := listenConfig.Listen(t.Context(), "tcp", "127.0.0.1:0")
-	if err != nil {
-		t.Fatalf("listen: %v", err)
-	}
-
-	t.Cleanup(func() { _ = listener.Close() })
-
-	go func() {
-		for {
-			conn, err := listener.Accept()
-			if err != nil {
-				return
-			}
-
-			_, _ = bufio.NewReader(conn).ReadString('\n')
-
-			// Closing a connection whose linger time is zero resets it.
-			if tcpConn, ok := conn.(*net.TCPConn); ok {
-				_ = tcpConn.SetLinger(0)
-			}
-
-			_ = conn.Close()
-		}
-	}()
-
-	return listener.Addr().String()
-}
-
-// newRIPEClient returns a client whose fake servers send answers. The
-// client adds one to *ripeAsks each time it connects to RIPE's server. When
-// reset is true, RIPE's server resets the connection instead of answering.
-func newRIPEClient(
-	t *testing.T, answers map[string]string, reset bool, ripeAsks *int,
-) *whois.Client {
-	t.Helper()
-
-	dial := fakeServers(t, answers)
-	resetAddress := resetServer(t)
-
-	client := whois.NewClient(slog.New(slog.DiscardHandler))
-	client.DialContext = func(
-		ctx context.Context, network, address string,
-	) (net.Conn, error) {
-		if address == ripeServer {
-			*ripeAsks++
-		}
-
-		if address == ripeServer && reset {
-			var dialer net.Dialer
-
-			return dialer.DialContext(ctx, network, resetAddress)
-		}
-
-		return dial(ctx, network, address)
-	}
-
-	return client
-}
-
-// TestLookupASNPausesAServerThatRefuses looks up an AS at RIPE while RIPE
-// refuses the query, in each of the ways a server refuses. The test sets
-// the client's clock. A lookup a second before PauseTime has passed fails
-// with ErrPaused and asks RIPE nothing. A lookup of an AS at APNIC succeeds
-// during the pause. A lookup at PauseTime asks RIPE again.
-func TestLookupASNPausesAServerThatRefuses(t *testing.T) {
-	t.Parallel()
-
-	for name, test := range map[string]struct {
-		// ripeAnswer is RIPE's answer. With none, RIPE's server refuses
-		// the connection, or resets it when reset is true.
-		ripeAnswer string
-		reset      bool
-	}{
-		"access denied":      {ripeAnswer: ripeRefusal},
-		"limit reached":      {ripeAnswer: ripeLimitReached + "\n"},
-		"refused connection": {},
-		"reset connection":   {reset: true},
-	} {
-		t.Run(name, func(t *testing.T) {
-			t.Parallel()
-
-			answers := map[string]string{
-				ianaQuery:      ianaReferral,
-				apnicIANAQuery: "refer:        whois.apnic.net\n",
-				apnicQuery:     apnicRecord,
-			}
-			if test.ripeAnswer != "" {
-				answers[ripeQuery] = test.ripeAnswer
-			}
-
-			ripeAsks := 0
-			client := newRIPEClient(t, answers, test.reset, &ripeAsks)
-
-			start := time.Date(2026, 11, 2, 0, 0, 0, 0, time.UTC)
-			now := start
-			client.Now = func() time.Time { return now }
-
-			_, err := client.LookupASN(t.Context(), testASN)
-			if err == nil || errors.Is(err, whois.ErrPaused) {
-				t.Fatalf("LookupASN while RIPE refuses: error %v, want a "+
-					"refusal", err)
-			}
-
-			now = start.Add(whois.PauseTime - time.Second)
-
-			_, err = client.LookupASN(t.Context(), testASN)
-			if !errors.Is(err, whois.ErrPaused) || ripeAsks != 1 {
-				t.Errorf("LookupASN during the pause: error %v after %d "+
-					"queries to RIPE, want ErrPaused after 1", err, ripeAsks)
-			}
-
-			_, err = client.LookupASN(t.Context(), apnicASN)
-			if err != nil {
-				t.Errorf("LookupASN of an AS at APNIC during RIPE's "+
-					"pause: %v", err)
-			}
-
-			now = start.Add(whois.PauseTime)
-
-			_, _ = client.LookupASN(t.Context(), testASN)
-			if ripeAsks != 2 {
-				t.Errorf("RIPE was asked %d times after the pause, want 2",
-					ripeAsks)
-			}
-		})
-	}
-}
-
 // TestLookupASNFailsWhenTheRIRGivesNoAnswer looks up an AS IANA refers to
 // RIPE when RIPE's server refuses the connection, and when it answers only
 // with an error. Neither is a lookup that found the AS.