Keep a peer-down until its route delete succeeds (closes #101)

When the delete of a peer's live routes after its session went down
failed, the down was dropped and the routes stayed. The prefix handler
now keeps such a peer, once, with the updates it sent after the down,
for the next flush, which retries the delete before it writes them; each
failure is logged. At most the last 25,000 of those updates are kept,
outside the batch, so they do not start a flush. The delete runs in
chunks of at most 10,000 routes per table, taken in prefix order
through the peer's key, each chunk in its own transaction, so the
database lock is released between chunks and a failure undoes at most
one chunk.

Model: opus-5-5

internal/routewatch/prefixhandler.go

@@ -28,9 +28,9 @@ const (
 	// prefixBatchMaxSize is the most prefix updates the batch holds while the
 	// flush goroutine writes the previous one: ten batches, about 60 MiB
 	// (README "Memory"). At this size HandleMessage waits until the flush
-	// goroutine takes the batch. On top of these, the batch holds at most
+	// goroutine takes the batch. On top of these, the handler holds at most
 	// prefixBatchSize updates, about 6 MiB, that a flush kept for the peers
-	// whose route delete failed (keepForNextFlush).
+	// whose route delete failed (held).
 	prefixBatchMaxSize = 250000
 
 	// prefixBatchTimeout is the maximum time to wait before flushing a batch
@@ -50,13 +50,21 @@ type PrefixHandler struct {
 	logger  *logger.Logger
 	metrics *metrics.Tracker
 
-	// Batching. mu guards batch and peersDown only: the flush goroutine takes
-	// both under mu and writes them after releasing mu, so HandleMessage
-	// waits for a database write only while batch is at prefixBatchMaxSize.
+	// Batching. mu guards batch, held and peersDown only: the flush goroutine
+	// takes them under mu and writes them after releasing mu, so
+	// HandleMessage waits for a database write only while batch is at
+	// prefixBatchMaxSize.
 	mu    sync.Mutex
 	batch []prefixUpdate
-	// peersDown holds the peers whose BGP session went down since the batch
-	// was taken, and those whose route delete failed in the last flush.
+	// held holds the updates that the peers whose route delete failed in the
+	// last flush sent after their session went down, for the next flush to
+	// write after it retries the delete. They are not in batch, so they do
+	// not count toward a full batch: a delete that keeps failing is retried
+	// at the pace of the flush timer.
+	held []prefixUpdate
+	// peersDown holds, each once, the peers whose BGP session went down
+	// since the batch was taken, and those whose route delete failed in the
+	// last flush.
 	peersDown []string
 	// batchTaken is signalled when the flush goroutine takes batch; its L is
 	// &mu.
@@ -202,17 +210,21 @@ func (h *PrefixHandler) Stop() {
 
 // peerSessionDown notes that the peer's BGP session went down, so that the next
 // flush deletes all of the peer's live routes before it writes the batch. The
-// peer's updates still in the batch came before the down, which replaces them,
-// so they are dropped; updates the peer sends later stay in the batch and are
-// written after the delete.
+// peer's updates still in the batch or held came before the down, which
+// replaces them, so they are dropped; updates the peer sends later stay in the
+// batch and are written after the delete. A peer already in peersDown is not
+// added again: one delete removes all of its routes.
 func (h *PrefixHandler) peerSessionDown(peer string) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 
-	h.batch = slices.DeleteFunc(h.batch, func(update prefixUpdate) bool {
-		return update.peer == peer
-	})
-	h.peersDown = append(h.peersDown, peer)
+	isPeers := func(update prefixUpdate) bool { return update.peer == peer }
+	h.batch = slices.DeleteFunc(h.batch, isPeers)
+	h.held = slices.DeleteFunc(h.held, isPeers)
+
+	if !slices.Contains(h.peersDown, peer) {
+		h.peersDown = append(h.peersDown, peer)
+	}
 }
 
 // flushLoop runs in a goroutine, the only one that writes the batch: it
@@ -240,18 +252,23 @@ func (h *PrefixHandler) flushLoop() {
 	}
 }
 
-// takeBatch returns the pending updates and the peers whose session went down
-// since the last flush, and leaves an empty batch and no peers in their place
-// for HandleMessage to fill.
+// takeBatch returns the pending updates, the held ones first, and the peers in
+// peersDown, and leaves an empty batch, no held updates and no peers in their
+// place for HandleMessage to fill.
 func (h *PrefixHandler) takeBatch() ([]prefixUpdate, []string) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 
-	if len(h.batch) == 0 && len(h.peersDown) == 0 {
+	if len(h.held) == 0 && len(h.batch) == 0 && len(h.peersDown) == 0 {
 		return nil, nil
 	}
 
 	batch, peersDown := h.batch, h.peersDown
+	if len(h.held) > 0 {
+		batch = slices.Concat(h.held, h.batch)
+	}
+
+	h.held = nil
 	h.batch = make([]prefixUpdate, 0, prefixBatchSize)
 	h.peersDown = nil
 	h.batchTaken.Broadcast()
@@ -379,15 +396,15 @@ func (h *PrefixHandler) deleteLiveRoutesOfPeers(peers []string) []string {
 	return failed
 }
 
-// keepForNextFlush puts the peers whose route delete failed back in front of
-// peersDown, and their updates from batch back in front of the batch, so that
-// the next flush retries each delete before it writes the peer's updates;
-// written now, they would be deleted by the retry. The updates of a peer whose
-// session went down again since the batch was taken came before that down, so
-// they are dropped, as peerSessionDown drops them. It returns the updates of the
-// other peers, for this flush to write.
+// keepForNextFlush puts the peers whose route delete failed back in peersDown,
+// and their updates from batch in held, so that the next flush retries each
+// delete before it writes the peer's updates; written now, they would be
+// deleted by the retry. The updates of a peer whose session went down again
+// since the batch was taken came before that down, so they are dropped, as
+// peerSessionDown drops them, and the peer is already in peersDown. It returns
+// the updates of the other peers, for this flush to write.
 //
-// While a delete keeps failing, the kept updates would grow with every flush,
+// While a delete keeps failing, the held updates would grow with every flush,
 // so only the last prefixBatchSize of them are kept and the older ones are
 // dropped. This never leaves in place a route the peer withdrew: a prefix with
 // a kept update keeps its last one, and the retry deletes all of the peer's
@@ -418,11 +435,18 @@ func (h *PrefixHandler) keepForNextFlush(
 			"Dropped the oldest updates of peers whose route delete failed",
 			"peers", failedPeers, "dropped", len(kept)-prefixBatchSize)
 
-		kept = kept[len(kept)-prefixBatchSize:]
+		// A copy, so that the dropped updates are freed.
+		kept = slices.Clone(kept[len(kept)-prefixBatchSize:])
 	}
 
-	h.batch = slices.Concat(kept, h.batch)
-	h.peersDown = slices.Concat(failedPeers, h.peersDown)
+	// takeBatch left held empty, and only this flush fills it.
+	h.held = kept
+
+	for _, peer := range failedPeers {
+		if !slices.Contains(h.peersDown, peer) {
+			h.peersDown = append(h.peersDown, peer)
+		}
+	}
 
 	return toWrite
 }

internal/routewatch/prefixhandler_internal_test.go

@@ -599,7 +599,9 @@ func (s *firstPeerDeleteFailsStore) DeleteLiveRoutesOfPeer(
 // on the next flush, and the announcement the peer sent after the down written
 // only after that retry, so it is kept. In the second case the peer's session
 // goes down again, and the peer announces another prefix, while the delete
-// fails: the announcement sent between the two downs must not be written.
+// fails: the announcement sent between the two downs must not be written, and
+// the next flush must delete the peer's routes once, not once for the retry and
+// once for the down.
 func TestPrefixHandlerRetriesAFailedDeleteOfAPeersRoutes(t *testing.T) {
 	t.Parallel()
 
@@ -633,8 +635,8 @@ func TestPrefixHandlerRetriesAFailedDeleteOfAPeersRoutes(t *testing.T) {
 				sessionDown(),
 				peerAnnouncement(testPeerIP, afterSecondDown),
 			},
-			// The retry and the delete for the second down.
-			wantDeletes: 3,
+			// One retry, which is also the delete for the second down.
+			wantDeletes: 2,
 			wantPeers: map[string][]string{
 				beforeDown:      nil,
 				afterDown:       nil,
@@ -694,9 +696,10 @@ func (peerDeleteAlwaysFailsStore) DeleteLiveRoutesOfPeer(string) (int, error) {
 // through the prefix handler into a database whose every delete of a peer's
 // routes fails, a peer's session going down and then three flushes. Before each
 // flush the peer announces prefixBatchSize new prefixes and another peer
-// announces one. After each flush the batch must hold only the peer's last
+// announces one. After each flush the handler must hold only the peer's last
 // prefixBatchSize updates, not every update since the down, and the other
-// peer's route must be written.
+// peer's route must be written. After the last flush, one more update must not
+// wake the flush goroutine: held updates do not count toward a full batch.
 func TestPrefixHandlerKeepsAtMostABatchOfUpdatesWhileADeleteKeepsFailing(
 	t *testing.T,
 ) {
@@ -730,18 +733,27 @@ func TestPrefixHandlerKeepsAtMostABatchOfUpdatesWhileADeleteKeepsFailing(
 		h.HandleMessage(peerAnnouncement(otherPeerIP, otherPeersPrefix))
 		h.flushBatch()
 
-		kept := make([]string, 0, len(h.batch))
-		for _, update := range h.batch {
+		kept := make([]string, 0, len(h.held))
+		for _, update := range h.held {
 			kept = append(kept, update.prefix)
 		}
 
 		if !slices.Equal(kept, prefixes) {
-			t.Fatalf("after flush %d the batch holds %d updates, not just "+
+			t.Fatalf("after flush %d the handler holds %d updates, not just "+
 				"the peer's last %d", flush+1, len(kept), len(prefixes))
 		}
 
 		assertPeers(t, db, map[string][]string{otherPeersPrefix: {otherPeerIP}})
 	}
+
+	// A new channel, so that a wake-up in it comes from the update below.
+	h.batchFullCh = make(chan struct{}, 1)
+	h.HandleMessage(peerAnnouncement(otherPeerIP, testV4Prefix))
+
+	if len(h.batchFullCh) > 0 {
+		t.Error("one update after a flush that held a batch of updates " +
+			"woke the flush goroutine")
+	}
 }
 
 // newTestDatabase opens a database in a new temporary directory. It is closed

internal/server/handlers_internal_test.go

@@ -183,14 +183,10 @@ func TestStatsHandlersAnswerFromMemory(t *testing.T) {
 	}
 }
 
-// TestPrefixDetailHandlersParsePrefix checks that both prefix pages find a
-// stored route whatever valid text form the URL writes its prefix in, and
-// answer 400 for a prefix that does not parse. The tables store the form
-// net/netip prints. Before the fix the URL text was looked up as typed, so
-// "2001:DB8:0:0::/32" was not found although "2001:db8::/32" was stored, and
-// "foo/24" answered 500 (https://git.eeqj.de/sneak/routewatch/issues/71).
-func TestPrefixDetailHandlersParsePrefix(t *testing.T) {
-	t.Parallel()
+// newServerWithTestRoutes returns a server on a new database that holds a
+// route for testPrefixV4 and one for 2001:db8::/32.
+func newServerWithTestRoutes(t *testing.T) *Server {
+	t.Helper()
 
 	db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New())
 	if err != nil {
@@ -217,7 +213,22 @@ func TestPrefixDetailHandlersParsePrefix(t *testing.T) {
 		t.Fatalf("UpsertLiveRouteBatch: %v", err)
 	}
 
-	s := New(db, streamer.New(logger.New(), metrics.New()), logger.New(), &config.Config{})
+	return New(db, streamer.New(logger.New(), metrics.New()), logger.New(),
+		&config.Config{})
+}
+
+// TestPrefixDetailHandlersParsePrefix checks that both prefix pages find a
+// stored route whatever valid text form the URL writes its prefix in, and
+// answer 400 for a prefix that does not parse. The tables store the form
+// net/netip prints. Before the fix the URL text was looked up as typed, so
+// "2001:DB8:0:0::/32" was not found although "2001:db8::/32" was stored, and
+// "foo/24" answered 500 (https://git.eeqj.de/sneak/routewatch/issues/71). A
+// prefix with its colons percent-encoded, as JavaScript's encodeURIComponent
+// writes it, answered 400 (https://git.eeqj.de/sneak/routewatch/issues/103).
+func TestPrefixDetailHandlersParsePrefix(t *testing.T) {
+	t.Parallel()
+
+	s := newServerWithTestRoutes(t)
 
 	tests := []struct {
 		path string
@@ -229,6 +240,8 @@ func TestPrefixDetailHandlersParsePrefix(t *testing.T) {
 		{"/prefix/2001:DB8:0:0::/32", http.StatusOK},
 		{"/api/v1/prefix/192.0.2.1/24", http.StatusOK},
 		{"/prefix/192.0.2.1/24", http.StatusOK},
+		{"/api/v1/prefix/2001%3Adb8%3A%3A/32", http.StatusOK},
+		{"/prefix/2001%3Adb8%3A%3A/32", http.StatusOK},
 	}
 
 	for _, tt := range tests {
@@ -243,6 +256,61 @@ func TestPrefixDetailHandlersParsePrefix(t *testing.T) {
 	}
 }
 
+// TestIPLookupsReadPercentEncodedAddresses checks that the IP lookups find the
+// addresses a URL asks for when it percent-encodes them, as the home page's
+// lookup form does with encodeURIComponent. Before the fix the handler read
+// "2001%3Adb8%3A%3A1" undecoded as a host name, which did not resolve, and
+// answered 400, so the home page could look up no IPv6 address
+// (https://git.eeqj.de/sneak/routewatch/issues/103).
+func TestIPLookupsReadPercentEncodedAddresses(t *testing.T) {
+	t.Parallel()
+
+	s := newServerWithTestRoutes(t)
+
+	tests := []struct {
+		path    string
+		wantIPs string // the addresses of the results, comma-separated
+	}{
+		{"/ip/2001%3Adb8%3A%3A1", "2001:db8::1"},
+		{"/api/v1/ip/192.0.2.1%2C192.0.2.2", "192.0.2.1,192.0.2.2"},
+	}
+
+	for _, tt := range tests {
+		rec := httptest.NewRecorder()
+		s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
+			t.Context(), http.MethodGet, tt.path, nil))
+
+		if rec.Code != http.StatusOK {
+			t.Errorf("%s: status %d, want %d; body %s",
+				tt.path, rec.Code, http.StatusOK, rec.Body)
+
+			continue
+		}
+
+		var body struct {
+			Data IPLookupResponse `json:"data"`
+		}
+
+		err := json.Unmarshal(rec.Body.Bytes(), &body)
+		if err != nil {
+			t.Errorf("%s: decoding the answer %q: %v", tt.path, rec.Body, err)
+
+			continue
+		}
+
+		gotIPs := make([]string, 0, len(body.Data.Results))
+
+		for _, result := range body.Data.Results {
+			gotIPs = append(gotIPs, result.IP)
+		}
+
+		got := strings.Join(gotIPs, ",")
+		if got != tt.wantIPs {
+			t.Errorf("%s: results for %q, want %q", tt.path, got, tt.wantIPs)
+		}
+	}
+}
+
 // errASDetailsLookupFailed is what failingASDetailsDB answers every AS details
 // lookup with.
 var errASDetailsLookupFailed = errors.New("AS details lookup failed")

internal/server/middleware.go

@@ -10,6 +10,8 @@ import (
 	"net/http"
 	"sync"
 	"time"
+
+	"github.com/go-chi/chi/v5"
 )
 
 // responseWriter wraps http.ResponseWriter to capture the response
@@ -281,3 +283,16 @@ func RequestLoggerMiddleware(logger *slog.Logger) func(http.Handler) http.Handle
 		})
 	}
 }
+
+// DecodedPathMiddleware makes the router match routes against the decoded
+// path, so that the handlers read their URL parameters decoded. Without it,
+// chi matches against the escaped path whenever the request has one and hands
+// the handlers their parameters still escaped: /ip/2001%3Adb8%3A%3A1 would
+// look up "2001%3Adb8%3A%3A1".
+func DecodedPathMiddleware(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		chi.RouteContext(r.Context()).RoutePath = r.URL.Path
+
+		next.ServeHTTP(w, r)
+	})
+}

internal/server/routes.go

@@ -12,6 +12,7 @@ func (s *Server) setupRoutes() {
 	r := chi.NewRouter()
 
 	// Middleware
+	r.Use(DecodedPathMiddleware)
 	r.Use(middleware.RequestID)
 	r.Use(middleware.RealIP)
 	r.Use(RequestLoggerMiddleware(s.logger.Logger)) // Structured request logging