Record origin changes and show them on the prefix page (closes #161)

The routing table records a row each time a prefix and origin AS gets its first peer (announced) or loses its last (withdrawn), with the time and how many peers carry the prefix just after. The prefix handler writes the rows to a new origin_changes table, apart from routes. Every hour the database maintainer keeps each prefix's rows from the last 72 hours or its newest 500, whichever keeps more. The prefix page lists them newest first, at most 5,000.

At most 100,000 rows wait in memory to be written; past that the oldest are dropped and logged. A stop loses the rows past its last write.

Model: deepseek-v4-pro (implementation); opus-5-5 (commit message)

TODO.md

@@ -12,17 +12,17 @@
 
 RouteWatch is pre-1.0, with no git tags. `main` last took `next` on 2026-10-27,
 through https://git.eeqj.de/sneak/routewatch/pulls/166.
-https://git.eeqj.de/sneak/routewatch/pulls/202 carries the twelve commits since
+https://git.eeqj.de/sneak/routewatch/pulls/202 carries the commits since then
-then into `main`. It is labelled `merge-ready` and waits for sneak.
+into `main`. It is labelled `merge-ready` and waits for sneak.
 
 The linter and the tests are phases of the Docker build. The Gitea workflow runs
-`script/cibuild` on every push. That script checks the formatting of the Go code
+`script/cibuild` on every push. That script runs the two phases and checks the
-and the markdown. Then it builds the image, which runs the two phases.
+formatting of the Go code and the markdown. Then it builds the image.
 
 The image sets memory ceilings for a 5 GiB container (README "Memory"). README
 "Running under upaas" says how to run it.
 
-A five-day run of `1c65656` on the live feed, from 27 October to 1 November,
+A five-day run of `1c65656` on the live feed, from 2026-10-27 to 2026-11-01,
 found three problems:
 
 - The prefix handler dropped updates while a peer's session went down
@@ -34,39 +34,77 @@ found three problems:
 
 All three are fixed on `next`. So is
 https://git.eeqj.de/sneak/routewatch/issues/222, which the third fix brought up.
-A 24-hour run of `26d939a` on the live feed is checking whether any handler
+A run of `26d939a` on the live feed showed the prefix handler still dropping
-still drops updates.
+updates while the home page's cards were worked out.
+https://git.eeqj.de/sneak/routewatch/issues/226 fixed that on `next`.
+https://git.eeqj.de/sneak/routewatch/issues/225, also on `next`, takes the peers
+whose sessions go down together out of the table in one walk. A 24-hour run of
+`9bd0e91` on the live feed is checking whether any handler still drops updates.
 
-The README rewrite, https://git.eeqj.de/sneak/routewatch/pulls/210, is paused
+One question waits for sneak: which fields the IP lookup answer gains
-until sneak has reviewed it. It is part of
+(https://git.eeqj.de/sneak/routewatch/issues/203).
-https://git.eeqj.de/sneak/routewatch/issues/157.
 
-Two questions wait for sneak: which history the prefix page keeps
+The order of the project's remaining work is kept on
-(https://git.eeqj.de/sneak/routewatch/issues/161), and which fields the IP
-lookup answer gains (https://git.eeqj.de/sneak/routewatch/issues/203).
-
-The state of what remains is kept on
 https://git.eeqj.de/sneak/routewatch/issues/86.
 
 # Next Step
 
-Post the outcome of the 24-hour run of `26d939a` on
+Post the outcome of the 24-hour run of `9bd0e91` on
-https://git.eeqj.de/sneak/routewatch/issues/216: whether any handler still drops
+https://git.eeqj.de/sneak/routewatch/issues/226: whether any handler still drops
 updates, and the process's peak memory against 5 GiB. File an issue for each
 problem the run finds.
 
 # Completed Steps
 
+- 2026-11-05: the prefix page lists the prefix's origin changes, newest first.
+  The routing table records each change, and the prefix handler writes them to a
+  new `origin_changes` table. The database maintainer prunes each prefix to its
+  newest 500 rows or its rows from the last 72 hours, whichever keeps more. The
+  page shows at most the newest 5,000. Each row shows:
+    - the time
+    - announced or withdrawn
+    - the origin AS
+    - the peer count
+
+    A row's peer count is how many peers carry the prefix just after the change,
+    with any origin AS (closes #161).
+
+- 2026-11-03: the streamer drops an update for a handler only when that
+  handler's queue is full. It used to drop a random share of the updates once
+  the queue was half full. The prefix handler's queue now holds 100,000
+  messages, about 100 MiB. It used to hold 20,000. While the routing table's
+  `Overview` walks every prefix, the prefix handler applies updates more slowly.
+  In a run on the live feed, the feed paused during such a walk and then sent
+  its backlog. The streamer then dropped about 10,000 of the prefix handler's
+  updates (closes #229)
+- 2026-11-02: `README.md` is rewritten from scratch, sentence by sentence,
+  against the rules of https://git.eeqj.de/sneak/routewatch/issues/157. Every
+  statement was checked against the code on `next` (closes #157)
 - 2026-11-02: `TODO.md` is rewritten from scratch, sentence by sentence, against
   the rules of https://git.eeqj.de/sneak/routewatch/issues/157. Every Completed
   Steps entry keeps its date, its facts and its issue. Status and Next Step
   describe the repo as the tracker shows it (part of #157)
+- 2026-11-02: when the sessions of several peers go down, the prefix handler
+  takes the peers out of the routing table together. The table's `PeersDown`
+  takes several peers out of every prefix in one walk. One walk runs at a time.
+  The next one takes every peer whose session went down since the previous one
+  began. Each peer used to have a walk of its own. A peer's messages stayed held
+  back while the walks queued ahead of its own ran. The handler holds back at
+  most 150,000 messages, about 150 MiB. It used to hold back at most 20,000. The
+  status page shows how many messages the handler holds back (closes #225)
+- 2026-11-02: working out the home page's cards no longer stops the prefix
+  handler. The walk over the routing table holds the table's lock for 10,000
+  prefixes at a time, then lets the lock go for as long again, as `PeerDown`
+  does. It used to take the lock back at once. The prefix handler then applied
+  about one update per 10,000 prefixes walked. During a walk, its queue could
+  fill, and the streamer then dropped updates for it (closes #226)
 - 2026-11-02: A skipped WHOIS lookup now makes the fetcher wait longer before
-  the next one, as a failed lookup does. A lookup is skipped when its registry
+  the next one, as a failed lookup does. A lookup is skipped when its RIR is
-  is paused. The longest wait is one minute. The status page does not count a
+  paused. An RIR is one of the five regional Internet registries. The longest
-  skip as a failure. A skip used to leave the wait as it was. Each lookup asks
+  wait is one minute. The status page does not count a skip as a failure. A skip
-  IANA before it finds its registry paused. So while most stale ASes belonged to
+  used to leave the wait as it was. Each lookup asks IANA which RIR holds the AS
-  a paused registry, the fetcher could ask IANA every second (closes #222)
+  before it finds that RIR paused. So while most stale ASes belonged to a paused
+  RIR, the fetcher could ask IANA every second (closes #222)
 - 2026-11-02: A peer's session going down no longer stops the prefix handler.
   The handler runs the routing table's `PeerDown` on a goroutine of its own, one
   peer at a time. Meanwhile it goes on taking messages. `PeerDown` holds the
@@ -77,15 +115,16 @@ problem the run finds.
   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 a registry's WHOIS server for an hour after the
+- 2026-11-02: RouteWatch pauses an RIR's or NIC.br's WHOIS server for an hour
-  server refuses a query. The registries are the five RIRs (the regional
+  after the server refuses a query. NIC.br is Brazil's registry. A refused or
-  Internet registries) and NIC.br. A refused or reset connection is a refusal.
+  reset connection is a refusal. So is an answer with an `%ERROR:201:` or
-  So is an answer with an `%ERROR:201:` or `%ERROR:202:` line. Lookups at the
+  `%ERROR:202:` line. Lookups at the other servers go on. An AS whose RIR is
-  other servers go on. An AS whose registry is paused is looked up after the
+  paused is looked up after the pause. While NIC.br is paused, a Brazilian AS
-  pause. IANA is never paused, because a pause of IANA would stop every lookup.
+  gets the details from LACNIC's answer. IANA is never paused, because a pause
-  A refusal by IANA is an ordinary failure. NIC.br is also paused for an hour
+  of IANA would stop every lookup. A refusal by IANA is an ordinary failure.
-  after it says RouteWatch has passed its query limit. Before, the fetcher asked
+  NIC.br is also paused for an hour after it says RouteWatch has passed its
-  a server again within seconds of a refusal (closes #218)
+  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
@@ -96,15 +135,15 @@ problem the run finds.
   host. Five of them gave `HandleMessage` two seconds to apply a full prefix
   batch. Under the race detector that can take longer. A test now waits up to 15
   seconds for each step it expects. The tests of the handlers' limits now give
-  the handlers small limits. So the whole package runs well inside its 90-second
+  the handlers small limits. So the package no longer runs past its 90-second
   limit (closes #213, #215)
 - 2026-10-27: The server works out the home page's cards every minute until it
   stops. It used to work them out only once, at startup. The end of the context
   given to `Start` stopped the refreshes (closes #211)
-- 2026-10-27: README "Rationale" says why RouteWatch exists. It answers our
+- 2026-10-27: README "Rationale" says why RouteWatch exists. RouteWatch answers
-  other services' questions about IP addresses, from its own routing table and
+  our other services' questions about IP addresses, from its own routing table
-  the WHOIS details it keeps. The section still says what RIS Live does not give
+  and the WHOIS details it keeps. The section still says what RIS Live does not
-  (closes #204)
+  give (closes #204)
 - 2026-10-27: Every JSON answer carries `@meta`. A panic on a JSON route answers
   500 with the JSON error body. `/status.json` and the health check answer 405
   with the JSON error body to any method they do not take. The 500 and the 405
@@ -184,23 +223,24 @@ problem the run finds.
   uses it yet (part of #122)
 - 2026-10-27: Each result of the IP lookup lists the public blocklists the
   address is on, in `blocklists`. The lists are DShield, Feodo Tracker, two
-  FireHOL lists, Spamhaus DROP and the Tor exit list. RouteWatch downloads each
+  FireHOL lists, two Spamhaus DROP lists and the Tor exit list. RouteWatch
-  list on its own schedule and answers from copies in memory. The copies are
+  downloads each list on its own schedule and answers from copies in memory. The
-  also kept in the state directory's `blocklists` directory. IP lookup answers
+  copies are also kept in the state directory's `blocklists` directory. IP
-  carry `Cache-Control`. An address with no route now has a result too, without
+  lookup answers carry `Cache-Control`. An address with no route now has a
-  route fields. So the IP lookup no longer answers 404 (closes #158)
+  result too, without route fields. So the IP lookup no longer answers 404
+  (closes #158)
 - 2026-10-27: The home page has an IP information API card. It shows the `curl`
   command that looks up the visitor's own address at `/ip/{addr}`. It also shows
   that command's live answer (closes #156)
 - 2026-10-27: On the AS page, a click on the heading of the IPv4 or the IPv6
   prefix list folds or unfolds the list. Both lists start open (closes #155)
 - 2026-10-27: The AS page lists its IPv4 and its IPv6 prefixes by mask length,
-  shortest first. Prefixes of one mask length are listed by address. The lists
+  shortest first. Prefixes of one mask length are listed by address. They used
-  used to be in address order alone (closes #153)
+  to be sorted by address first, then by mask length (closes #153)
-- 2026-10-27: On `/` and `/status`, a spinner beside the status turns one frame
+- 2026-10-27: The connection status on `/` and `/status` gets a spinner. It
-  each time the stats show more messages than the time before. After three
+  turns one frame each time the stats show more messages than the time before.
-  refreshes with no new messages, the page shows "stale" in red instead. Next to
+  After three refreshes with no new messages, the page shows "stale" in red
-  it is the time since the last fresh answer (closes #152)
+  instead. Next to it is the time since the last fresh answer (closes #152)
 - 2026-10-27: The WHOIS details of a Brazilian AS keep their accented letters.
   LACNIC's answer for such an AS is NIC.br's answer with each accented letter
   replaced. NIC.br is Brazil's registry. The details now come from NIC.br's own
@@ -226,8 +266,8 @@ problem the run finds.
   and stored. RIPE and APNIC send Latin-1 (closes #165)
 - 2026-10-27: The Age column on the AS and prefix pages reads in words, such as
   "5 minutes ago" or "2 weeks ago". It used to print Go durations such as
-  "360h0m0s ago". An age under a minute reads in seconds. An age over a month
+  "360h0m0s ago". An age under a minute reads "now" or in seconds. An age over a
-  reads in months or years instead of a date (closes #163)
+  month reads in months or years instead of a date (closes #163)
 - 2026-10-26: A write to a stored AS, peering or peer keeps the earlier of the
   two first-seen times. It keeps the later of the two last-seen times. An old
   update from a collector catching up on a backlog no longer moves last seen
@@ -382,8 +422,9 @@ problem the run finds.
 - 2026-09-21: `make test` skips the live feed test, with `-short`. So
   `make check` no longer depends on the network (closes #2)
 - 2026-07-07: The scripts-to-rule-them-all layout is adopted. `script/` holds
-  the entrypoints, and the `Makefile` targets call them. README gains an
+  the entrypoints. The `Makefile` targets `check`, `test`, `lint`, `fmt` and
-  "Entrypoints" section
+  `fmt-check` call them. So do `bootstrap`, `setup`, `docker` and `hooks`.
+  README gains an "Entrypoints" section
 - 2026-02-22: The repo is brought in line with the repo policies. The required
   policy files are added. `.gitignore` is updated. The `Makefile` gains
   `fmt-check`, `check`, `docker` and `hooks` targets. The Go code is formatted
@@ -403,7 +444,3 @@ problem the run finds.
 - 2025-07-27: The first commit: RouteWatch, a BGP stream monitor
 
 # Future Steps
-
-- Merge https://git.eeqj.de/sneak/routewatch/pulls/210 once sneak has reviewed
-  it. The last PR for https://git.eeqj.de/sneak/routewatch/issues/157 to merge
-  closes that issue.

internal/database/database_internal_test.go

@@ -273,7 +273,7 @@ func TestVacuumReturnsFreePages(t *testing.T) {
 	writeTableChanges(t, db, table)
 
 	// The peer going down removes every prefix, and so every row.
-	table.PeerDown(peerA)
+	table.PeersDown([]string{peerA})
 	writeTableChanges(t, db, table)
 
 	var before int

internal/database/interface.go

@@ -1,6 +1,7 @@
 // Package database provides SQLite storage for BGP routing data: ASNs,
-// prefixes, announcements, peerings and BGP peers. Its routes table holds a
+// prefixes, announcements, peerings, BGP peers and origin changes. Its routes
-// copy of the whole routing table that internal/routetable keeps in memory.
+// table holds a copy of the whole routing table that internal/routetable
+// keeps in memory.
 package database
 
 import (
@@ -55,6 +56,14 @@ type Store interface {
 	WriteRouteChanges(changes []routetable.Change, newPeers map[string]uint) error
 	ReadRouteTable(ctx context.Context) (*routetable.Table, error)
 
+	// Origin change operations: the prefix and origin AS pairs' first and
+	// last peers, as the routing table records them.
+	WriteOriginChanges(changes []routetable.OriginChange) error
+	GetOriginChanges(
+		ctx context.Context, prefix netip.Prefix, limit int,
+	) ([]routetable.OriginChange, error)
+	PruneOriginChanges(ctx context.Context) error
+
 	// ASN WHOIS operations
 	GetNextStaleASN(ctx context.Context, staleThreshold time.Duration) (int, error)
 	UpdateASNWHOIS(ctx context.Context, update *ASNWHOISUpdate) error

internal/database/originchanges.go

@@ -0,0 +1,172 @@
+package database
+
+import (
+	"context"
+	"fmt"
+	"net/netip"
+	"time"
+
+	"git.eeqj.de/sneak/routewatch/internal/routetable"
+)
+
+// minOriginChangesPerPrefix is how many of a prefix's newest origin changes
+// the prune always keeps, whatever their age.
+const minOriginChangesPerPrefix = 500
+
+// originChangesRetention is how long the prune keeps an origin change: rows
+// older than this are deleted, unless they are among the prefix's newest
+// minOriginChangesPerPrefix.
+const originChangesRetention = 72 * time.Hour
+
+// The statements of the origin_changes table (schema.sql).
+const (
+	insertOriginChangeSQL = `INSERT INTO origin_changes
+		(prefix, origin_asn, announced, peer_count, time)
+		VALUES (?, ?, ?, ?, ?)`
+	originChangesSQL = `SELECT prefix, origin_asn, announced, peer_count, time
+		FROM origin_changes WHERE prefix = ? ORDER BY id DESC LIMIT ?`
+	prunePrefixesSQL = `SELECT prefix FROM origin_changes
+		GROUP BY prefix HAVING COUNT(*) > ?`
+	pruneOriginChangesSQL = `DELETE FROM origin_changes
+		WHERE prefix = ? AND time < ? AND id < (
+			SELECT id FROM origin_changes WHERE prefix = ?
+			ORDER BY id DESC LIMIT 1 OFFSET ?)`
+)
+
+// WriteOriginChanges writes, in one transaction, the origin changes the
+// routing table handed over. When it fails, it writes nothing; the caller then
+// writes them again.
+func (d *Database) WriteOriginChanges(changes []routetable.OriginChange) error {
+	if len(changes) == 0 {
+		return nil
+	}
+
+	d.lock("WriteOriginChanges")
+	defer d.unlock()
+
+	tx, err := d.beginTx()
+	if err != nil {
+		return fmt.Errorf("failed to begin transaction: %w", err)
+	}
+	defer d.rollback(tx)
+
+	ctx := context.Background()
+
+	stmt, err := tx.PrepareContext(ctx, insertOriginChangeSQL)
+	if err != nil {
+		return fmt.Errorf("failed to prepare origin change insert: %w", err)
+	}
+	defer func() { _ = stmt.Close() }()
+
+	for _, change := range changes {
+		_, err := stmt.ExecContext(ctx, change.Prefix.String(),
+			change.OriginASN, change.Announced, change.PeerCount,
+			change.Time.Unix())
+		if err != nil {
+			return fmt.Errorf("failed to write origin change %s: %w",
+				change.Prefix, err)
+		}
+	}
+
+	err = tx.Commit()
+	if err != nil {
+		return fmt.Errorf("failed to commit transaction: %w", err)
+	}
+
+	return nil
+}
+
+// GetOriginChanges returns at most limit of prefix's origin changes, newest
+// first.
+func (d *Database) GetOriginChanges(
+	ctx context.Context, prefix netip.Prefix, limit int,
+) ([]routetable.OriginChange, error) {
+	rows, err := d.db.QueryContext(ctx, originChangesSQL,
+		prefix.Masked().String(), limit)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read origin changes: %w", err)
+	}
+	defer func() { _ = rows.Close() }()
+
+	var changes []routetable.OriginChange
+
+	for rows.Next() {
+		var (
+			change     routetable.OriginChange
+			prefixText string
+			timeSec    int64
+		)
+
+		err := rows.Scan(&prefixText, &change.OriginASN, &change.Announced,
+			&change.PeerCount, &timeSec)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read an origin change: %w", err)
+		}
+
+		change.Prefix, err = netip.ParsePrefix(prefixText)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read origin change %s: %w",
+				prefixText, err)
+		}
+
+		change.Time = time.Unix(timeSec, 0).UTC()
+		changes = append(changes, change)
+	}
+
+	err = rows.Err()
+	if err != nil {
+		return nil, fmt.Errorf("failed to read origin changes: %w", err)
+	}
+
+	return changes, nil
+}
+
+// PruneOriginChanges deletes, for each prefix with more than
+// minOriginChangesPerPrefix rows, the rows older than originChangesRetention
+// that are not among its newest minOriginChangesPerPrefix. It finds those
+// prefixes on a reading connection and deletes one prefix per write, so no
+// write holds the lock for long.
+func (d *Database) PruneOriginChanges(ctx context.Context) error {
+	rows, err := d.db.QueryContext(ctx, prunePrefixesSQL,
+		minOriginChangesPerPrefix)
+	if err != nil {
+		return fmt.Errorf("failed to find prefixes to prune: %w", err)
+	}
+	defer func() { _ = rows.Close() }()
+
+	var prefixes []string
+
+	for rows.Next() {
+		var prefix string
+
+		err := rows.Scan(&prefix)
+		if err != nil {
+			return fmt.Errorf("failed to read a prefix to prune: %w", err)
+		}
+
+		prefixes = append(prefixes, prefix)
+	}
+
+	err = rows.Err()
+	if err != nil {
+		return fmt.Errorf("failed to find prefixes to prune: %w", err)
+	}
+
+	cutoff := time.Now().Add(-originChangesRetention).Unix()
+
+	for _, prefix := range prefixes {
+		d.lock("PruneOriginChanges")
+
+		_, err := d.writer.ExecContext(ctx, pruneOriginChangesSQL,
+			prefix, cutoff, prefix, minOriginChangesPerPrefix-1)
+
+		d.unlock()
+
+		if err != nil {
+			return fmt.Errorf("failed to prune origin changes of %s: %w",
+				prefix, err)
+		}
+	}
+
+	return nil
+}

internal/database/originchanges_internal_test.go

@@ -0,0 +1,217 @@
+package database
+
+import (
+	"net/netip"
+	"reflect"
+	"testing"
+	"time"
+
+	"git.eeqj.de/sneak/routewatch/internal/routetable"
+)
+
+// TestOriginChangesWriteThenRead writes three origin changes and reads them
+// back newest first, honouring the limit.
+func TestOriginChangesWriteThenRead(t *testing.T) {
+	t.Parallel()
+
+	db := newRoutesTestDatabase(t)
+	prefix := netip.MustParsePrefix(tablePrefixV4)
+
+	changes := []routetable.OriginChange{
+		{Prefix: prefix, OriginASN: asA, Announced: true, PeerCount: 1,
+			Time: tableTime(10)},
+		{Prefix: prefix, OriginASN: asA, Announced: false, PeerCount: 0,
+			Time: tableTime(20)},
+		{Prefix: prefix, OriginASN: asB, Announced: true, PeerCount: 1,
+			Time: tableTime(30)},
+	}
+
+	err := db.WriteOriginChanges(changes)
+	if err != nil {
+		t.Fatalf("WriteOriginChanges: %v", err)
+	}
+
+	got, err := db.GetOriginChanges(t.Context(), prefix, 2)
+	if err != nil {
+		t.Fatalf("GetOriginChanges: %v", err)
+	}
+
+	want := []routetable.OriginChange{
+		{Prefix: prefix, OriginASN: asB, Announced: true, PeerCount: 1,
+			Time: tableTime(30)},
+		{Prefix: prefix, OriginASN: asA, Announced: false, PeerCount: 0,
+			Time: tableTime(20)},
+	}
+
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("GetOriginChanges = %+v, want %+v", got, want)
+	}
+}
+
+// TestOriginChangesPrune checks that the prune keeps, for a prefix with more
+// than 500 rows, its newest 500 rows and every row newer than 72 hours, and
+// deletes the rest. It checks which rows survive by their times, not only how
+// many.
+func TestOriginChangesPrune(t *testing.T) {
+	t.Parallel()
+
+	db := newRoutesTestDatabase(t)
+	oldPrefix := netip.MustParsePrefix(tablePrefixV4)
+	straddling := netip.MustParsePrefix(tablePrefixShortV4)
+	busy := netip.MustParsePrefix(tablePrefixBusyV4)
+	other := netip.MustParsePrefix("10.20.0.0/16")
+
+	// Whole seconds, so the stored times round-trip exactly.
+	now := time.Now().Truncate(time.Second)
+
+	// 600 rows for oldPrefix, all older than 72 hours, oldest first. The
+	// prune keeps the newest 500 and deletes the oldest 100.
+	changes := rows(oldPrefix, now.Add(-100*time.Hour), 600)
+	// 600 rows for straddling: 400 older than 72 hours, then 200 newer. The
+	// prune keeps all 200 newer and the newest 300 of the 400 older.
+	changes = append(changes, rows(straddling, now.Add(-100*time.Hour), 400)...)
+	changes = append(changes, rows(straddling, now.Add(-71*time.Hour), 200)...)
+	// 100 rows older than 72 hours and minOriginChangesPerPrefix+100 newer
+	// rows for busy. The prune keeps all the newer rows and none of the older,
+	// because the newer rows alone are more than minOriginChangesPerPrefix.
+	changes = append(changes, rows(busy, now.Add(-100*time.Hour), 100)...)
+	changes = append(changes, rows(busy, now.Add(-71*time.Hour),
+		minOriginChangesPerPrefix+100)...)
+	// One row for other, which the prune leaves alone.
+	changes = append(changes, rows(other, now.Add(-100*time.Hour), 1)...)
+
+	err := db.WriteOriginChanges(changes)
+	if err != nil {
+		t.Fatalf("WriteOriginChanges: %v", err)
+	}
+
+	err = db.PruneOriginChanges(t.Context())
+	if err != nil {
+		t.Fatalf("PruneOriginChanges: %v", err)
+	}
+
+	// The newest row survives, and the oldest survivor is the 100th oldest,
+	// so a prune that kept the oldest 500 instead would fail.
+	checkSurvivors(t, db, oldPrefix, 500,
+		now.Add(-100*time.Hour).Add(599*time.Minute),
+		now.Add(-100*time.Hour).Add(100*time.Minute))
+
+	// The newest of the newer-than-72-hours rows survives, and the oldest
+	// survivor is the 100th oldest of the older-than-72-hours rows.
+	checkSurvivors(t, db, straddling, 500,
+		now.Add(-71*time.Hour).Add(199*time.Minute),
+		now.Add(-100*time.Hour).Add(100*time.Minute))
+
+	// Every newer-than-72-hours row survives, so the oldest survivor is the
+	// first of them and the newest is minOriginChangesPerPrefix+99 minutes
+	// after it.
+	checkSurvivors(t, db, busy, minOriginChangesPerPrefix+100,
+		now.Add(-71*time.Hour).
+			Add(time.Duration(minOriginChangesPerPrefix+99)*time.Minute),
+		now.Add(-71*time.Hour))
+
+	gotOther, err := db.GetOriginChanges(t.Context(), other, 10)
+	if err != nil {
+		t.Fatalf("GetOriginChanges: %v", err)
+	}
+
+	if len(gotOther) != 1 {
+		t.Errorf("GetOriginChanges(other) = %d rows, want 1", len(gotOther))
+	}
+}
+
+// rows returns n origin changes for prefix, at times from plus one minute per
+// step, oldest first.
+func rows(prefix netip.Prefix, from time.Time, n int) []routetable.OriginChange {
+	changes := make([]routetable.OriginChange, 0, n)
+
+	for i := range n {
+		changes = append(changes, routetable.OriginChange{
+			Prefix: prefix, OriginASN: asA, Announced: true, PeerCount: 1,
+			Time: from.Add(time.Duration(i) * time.Minute),
+		})
+	}
+
+	return changes
+}
+
+// survivorTimes returns the times of prefix's rows after the prune, newest
+// first.
+func survivorTimes(
+	t *testing.T, db *Database, prefix netip.Prefix, limit int,
+) []time.Time {
+	t.Helper()
+
+	rows, err := db.GetOriginChanges(t.Context(), prefix, limit)
+	if err != nil {
+		t.Fatalf("GetOriginChanges: %v", err)
+	}
+
+	times := make([]time.Time, len(rows))
+	for i, row := range rows {
+		times[i] = row.Time
+	}
+
+	return times
+}
+
+// checkSurvivors checks that prefix kept exactly wantCount rows, with the
+// newest survivor at wantNewest and the oldest at wantOldest.
+func checkSurvivors(
+	t *testing.T, db *Database, prefix netip.Prefix,
+	wantCount int, wantNewest, wantOldest time.Time,
+) {
+	t.Helper()
+
+	times := survivorTimes(t, db, prefix, wantCount+1)
+	if len(times) != wantCount {
+		t.Errorf("GetOriginChanges(%s) = %d rows, want %d",
+			prefix, len(times), wantCount)
+
+		return
+	}
+
+	if !times[0].Equal(wantNewest) {
+		t.Errorf("the newest surviving row's time = %v, want %v",
+			times[0], wantNewest)
+	}
+
+	if !times[wantCount-1].Equal(wantOldest) {
+		t.Errorf("the oldest surviving row's time = %v, want %v",
+			times[wantCount-1], wantOldest)
+	}
+}
+
+// TestOriginChangesSurviveRouteDeletion checks that deleting a routes row
+// leaves the prefix's origin_changes rows.
+func TestOriginChangesSurviveRouteDeletion(t *testing.T) {
+	t.Parallel()
+
+	db := newRoutesTestDatabase(t)
+	prefix := netip.MustParsePrefix(tablePrefixV4)
+
+	err := db.WriteOriginChanges([]routetable.OriginChange{{
+		Prefix: prefix, OriginASN: asA, Announced: true, PeerCount: 1,
+		Time: tableTime(10),
+	}})
+	if err != nil {
+		t.Fatalf("WriteOriginChanges: %v", err)
+	}
+
+	// Write a route, then delete it.
+	table := routetable.New(routetable.Params{})
+	announceTo(table, tablePrefixV4, asA, peerA, 10)
+	writeTableChanges(t, db, table)
+
+	table.Withdraw(prefix, peerA)
+	writeTableChanges(t, db, table)
+
+	got, err := db.GetOriginChanges(t.Context(), prefix, 10)
+	if err != nil {
+		t.Fatalf("GetOriginChanges: %v", err)
+	}
+
+	if len(got) != 1 {
+		t.Errorf("GetOriginChanges = %d rows, want 1", len(got))
+	}
+}

internal/database/routes_internal_test.go

@@ -34,6 +34,7 @@ const (
 
 	tablePrefixV4      = "10.1.2.0/24"
 	tablePrefixShortV4 = "10.9.0.0/16"
+	tablePrefixBusyV4  = "10.10.0.0/16"
 	tablePrefixV6      = "2001:db8:1::/48"
 )
 
@@ -107,7 +108,7 @@ func TestReadRouteTableGivesBackWhatWasWritten(t *testing.T) {
 
 	// A peer moves to another origin, one goes down and a new one comes.
 	announceTo(source, tablePrefixV4, asB, peerA, 60)
-	source.PeerDown(peerC)
+	source.PeersDown([]string{peerC})
 	announceTo(source, tablePrefixV6, asB, peerE, 70)
 
 	writeTableChanges(t, db, source)
@@ -133,7 +134,7 @@ func TestReadRouteTableGivesATableThatTakesLaterUpdates(t *testing.T) {
 	// another one goes down, and a new peer comes.
 	for _, table := range []*routetable.Table{source, read} {
 		table.Withdraw(netip.MustParsePrefix(tablePrefixV4), peerA)
-		table.PeerDown(peerB)
+		table.PeersDown([]string{peerB})
 		announceTo(table, tablePrefixShortV4, asB, peerD, 50)
 	}
 

internal/database/schema.sql

@@ -120,3 +120,19 @@ CREATE TABLE IF NOT EXISTS route_peers (
 	number INTEGER PRIMARY KEY,
 	address TEXT NOT NULL UNIQUE
 );
+
+-- The origin changes of the routing table: each time a prefix and origin AS
+-- gained its first peer or lost its last. The prefix handler writes them, and
+-- the database maintainer prunes each prefix to its newest 500 rows or its
+-- rows from the last 72 hours, whichever keeps more. time is whole seconds
+-- since 1970, UTC.
+CREATE TABLE IF NOT EXISTS origin_changes (
+	id INTEGER PRIMARY KEY,
+	prefix TEXT NOT NULL,
+	origin_asn INTEGER NOT NULL,
+	announced BOOLEAN NOT NULL,
+	peer_count INTEGER NOT NULL,
+	time INTEGER NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_origin_changes_prefix ON origin_changes(prefix);

internal/routetable/originchanges.go

@@ -0,0 +1,103 @@
+package routetable
+
+import (
+	"net/netip"
+	"slices"
+	"time"
+)
+
+// MaxWaitingOriginChanges is the most origin changes the table keeps waiting
+// for the prefix handler to write. The oldest change past it is dropped and
+// counted.
+const MaxWaitingOriginChanges = 100000
+
+// OriginChange is one change of a prefix and origin AS pair: the origin AS
+// gained its first peer, or lost its last.
+type OriginChange struct {
+	Prefix    netip.Prefix
+	OriginASN int
+	// Announced says the origin AS gained its first peer. When false, it
+	// lost its last peer.
+	Announced bool
+	// PeerCount is how many peers carry the prefix just after the change,
+	// counting every origin AS.
+	PeerCount int
+	// Time is when RouteWatch applied the change.
+	Time time.Time
+}
+
+// recordOriginChange records that the pair of prefix and asn gained its first
+// peer, when announced, or lost its last, when not. The caller holds the lock.
+// peerCount is how many peers carry the prefix just after the change.
+func (t *Table) recordOriginChange(
+	prefix netip.Prefix, asn int, announced bool, peerCount int,
+) {
+	change := OriginChange{
+		Prefix:    prefix,
+		OriginASN: asn,
+		Announced: announced,
+		PeerCount: peerCount,
+		Time:      t.now(),
+	}
+
+	t.originChanges = append(t.originChanges, change)
+	t.trimOriginChanges()
+}
+
+// TakeOriginChanges returns at most limit of the origin changes that wait to
+// be written, oldest first, and how many were dropped since the last call. It
+// then forgets what it returned.
+func (t *Table) TakeOriginChanges(limit int) ([]OriginChange, int) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	n := min(limit, len(t.originChanges))
+	changes := slices.Clone(t.originChanges[:n])
+	t.originChanges = t.originChanges[n:]
+
+	dropped := t.originChangesDropped
+	t.originChangesDropped = 0
+
+	return changes, dropped
+}
+
+// PutBackOriginChanges records again what one TakeOriginChanges call returned,
+// for a caller that could not write it. The next TakeOriginChanges returns
+// those changes first, before the ones that arrived since. The oldest changes
+// past the cap are dropped and counted.
+func (t *Table) PutBackOriginChanges(changes []OriginChange) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	t.originChanges = append(changes, t.originChanges...)
+	t.trimOriginChanges()
+}
+
+// trimOriginChanges drops the oldest origin changes past the cap and counts
+// them. The caller holds the lock.
+func (t *Table) trimOriginChanges() {
+	if extra := len(t.originChanges) - MaxWaitingOriginChanges; extra > 0 {
+		t.originChangesDropped += extra
+		t.originChanges = t.originChanges[extra:]
+	}
+}
+
+// OriginChangeCount returns how many origin changes wait for TakeOriginChanges.
+func (t *Table) OriginChangeCount() int {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+
+	return len(t.originChanges)
+}
+
+// prefixPeerCount returns how many peers carry prefix, counting every origin
+// AS. The caller holds the lock.
+func (t *Table) prefixPeerCount(prefix netip.Prefix) int {
+	count := 0
+
+	for _, o := range t.origins(prefix) {
+		count += int(o.peers.Count())
+	}
+
+	return count
+}

internal/routetable/originchanges_test.go

@@ -0,0 +1,302 @@
+package routetable_test
+
+import (
+	"math"
+	"net/netip"
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/bits-and-blooms/bitset"
+
+	"git.eeqj.de/sneak/routewatch/internal/routetable"
+)
+
+// originChange returns the origin change the table records for prefix and asn
+// at the clock's time, announced or withdrawn, with peerCount peers.
+func originChange(
+	prefix netip.Prefix, asn int, announced bool, peerCount int, at time.Time,
+) routetable.OriginChange {
+	return routetable.OriginChange{
+		Prefix:    prefix,
+		OriginASN: asn,
+		Announced: announced,
+		PeerCount: peerCount,
+		Time:      at,
+	}
+}
+
+// TestOriginChangesRecordAnnouncedAndWithdrawn checks that a prefix and origin
+// AS pair records an announced row when it gains its first peer, nothing for a
+// second peer, and a withdrawn row when it loses its last peer.
+func TestOriginChangesRecordAnnouncedAndWithdrawn(t *testing.T) {
+	t.Parallel()
+
+	now := at(0)
+	table := routetable.New(routetable.Params{
+		Now: func() time.Time { return now },
+	})
+	prefix := netip.MustParsePrefix(prefixV4)
+
+	table.Announce(prefix, asA, routeOf(peerA, asA, 10))
+	table.Announce(prefix, asA, routeOf(peerB, asA, 20))
+
+	checkOriginChanges(t, table, []routetable.OriginChange{
+		originChange(prefix, asA, true, 1, now),
+	})
+
+	table.Withdraw(prefix, peerA)
+	table.Withdraw(prefix, peerB)
+
+	checkOriginChanges(t, table, []routetable.OriginChange{
+		originChange(prefix, asA, false, 0, now),
+	})
+}
+
+// TestOriginChangesRecordAPeerMovingOrigin checks that a peer moving from one
+// origin AS to another records a withdrawn row for the old one and an
+// announced row for the new one.
+func TestOriginChangesRecordAPeerMovingOrigin(t *testing.T) {
+	t.Parallel()
+
+	now := at(0)
+	table := routetable.New(routetable.Params{
+		Now: func() time.Time { return now },
+	})
+	prefix := netip.MustParsePrefix(prefixV4)
+
+	table.Announce(prefix, asA, routeOf(peerA, asA, 10))
+	table.Announce(prefix, asB, routeOf(peerA, asB, 20))
+
+	checkOriginChanges(t, table, []routetable.OriginChange{
+		originChange(prefix, asA, true, 1, now),
+		originChange(prefix, asA, false, 0, now),
+		originChange(prefix, asB, true, 1, now),
+	})
+}
+
+// TestOriginChangesRecordPeersDown checks that PeersDown records a withdrawn
+// row for each prefix and origin AS pair it empties.
+func TestOriginChangesRecordPeersDown(t *testing.T) {
+	t.Parallel()
+
+	now := at(0)
+	table := routetable.New(routetable.Params{
+		Now: func() time.Time { return now },
+	})
+	v4 := netip.MustParsePrefix(prefixV4)
+	v6 := netip.MustParsePrefix(prefixV6)
+
+	table.Announce(v4, asA, routeOf(peerA, asA, 10))
+	table.Announce(v6, asA, routeOf(peerA, asA, 10))
+	table.Announce(v4, asB, routeOf(peerB, asB, 20))
+
+	table.PeersDown([]string{peerA})
+
+	// peerA carried v4 and v6 with asA; both pairs are emptied. peerB still
+	// carries v4 with asB, so that pair is not.
+	checkOriginChanges(t, table, []routetable.OriginChange{
+		originChange(v4, asA, true, 1, now),
+		originChange(v6, asA, true, 1, now),
+		originChange(v4, asB, true, 2, now),
+		originChange(v4, asA, false, 1, now),
+		originChange(v6, asA, false, 0, now),
+	})
+}
+
+// TestOriginChangesLoadRecordsNothing checks that Load records no origin
+// change.
+func TestOriginChangesLoadRecordsNothing(t *testing.T) {
+	t.Parallel()
+
+	table := routetable.New(routetable.Params{
+		PeerNumbers: map[string]uint{peerA: 0},
+	})
+	prefix := netip.MustParsePrefix(prefixV4)
+
+	peers := new(bitset.BitSet)
+	peers.Set(0)
+
+	table.Load(routetable.Change{
+		Prefix:    prefix,
+		OriginASN: asA,
+		Peers:     peers,
+		Routes:    []routetable.Route{routeOf(peerA, asA, 10)},
+	})
+
+	changes, dropped := table.TakeOriginChanges(math.MaxInt)
+	if dropped != 0 || len(changes) != 0 {
+		t.Errorf("TakeOriginChanges = %+v, %d; want none", changes, dropped)
+	}
+}
+
+// TestOriginChangesTakeReturnsOldestFirst checks that TakeOriginChanges
+// returns the changes oldest first and at most max.
+func TestOriginChangesTakeReturnsOldestFirst(t *testing.T) {
+	t.Parallel()
+
+	table := routetable.New(routetable.Params{})
+	prefix := netip.MustParsePrefix(prefixV4)
+
+	table.Announce(prefix, asA, routeOf(peerA, asA, 10))
+	table.Announce(prefix, asB, routeOf(peerB, asB, 20))
+	table.Announce(prefix, asC, routeOf(peerC, asC, 30))
+
+	first, dropped := table.TakeOriginChanges(2)
+	if dropped != 0 {
+		t.Errorf("dropped = %d, want 0", dropped)
+	}
+
+	if len(first) != 2 || first[0].OriginASN != asA || first[1].OriginASN != asB {
+		t.Errorf("first TakeOriginChanges = %+v, want the first two", first)
+	}
+
+	second, _ := table.TakeOriginChanges(math.MaxInt)
+	if len(second) != 1 || second[0].OriginASN != asC {
+		t.Errorf("second TakeOriginChanges = %+v, want the third", second)
+	}
+}
+
+// TestOriginChangesPutBackPutsRowsInFront checks that PutBackOriginChanges
+// puts rows back in front, so the next TakeOriginChanges returns them first.
+func TestOriginChangesPutBackPutsRowsInFront(t *testing.T) {
+	t.Parallel()
+
+	table := routetable.New(routetable.Params{})
+	prefix := netip.MustParsePrefix(prefixV4)
+
+	table.Announce(prefix, asA, routeOf(peerA, asA, 10))
+	table.Announce(prefix, asB, routeOf(peerB, asB, 20))
+
+	taken, _ := table.TakeOriginChanges(math.MaxInt)
+	table.Announce(prefix, asC, routeOf(peerC, asC, 30))
+	table.PutBackOriginChanges(taken)
+
+	changes, _ := table.TakeOriginChanges(math.MaxInt)
+	if len(changes) != 3 || changes[0].OriginASN != asA ||
+		changes[1].OriginASN != asB || changes[2].OriginASN != asC {
+		t.Errorf("TakeOriginChanges = %+v, want the put-back rows first", changes)
+	}
+}
+
+// TestOriginChangesDropPastTheLimit checks that a change past 100,000 waiting
+// rows is dropped and counted.
+func TestOriginChangesDropPastTheLimit(t *testing.T) {
+	t.Parallel()
+
+	table := routetable.New(routetable.Params{})
+	addr := netip.MustParseAddr("10.0.0.0")
+
+	for range routetable.MaxWaitingOriginChanges + 1 {
+		prefix := netip.PrefixFrom(addr, 32)
+		table.Announce(prefix, asA, routeOf(peerA, asA, 10))
+
+		addr = addr.Next()
+	}
+
+	changes, dropped := table.TakeOriginChanges(routetable.MaxWaitingOriginChanges)
+	if dropped != 1 {
+		t.Errorf("dropped = %d, want 1", dropped)
+	}
+
+	if len(changes) != routetable.MaxWaitingOriginChanges {
+		t.Errorf("len(changes) = %d, want %d", len(changes),
+			routetable.MaxWaitingOriginChanges)
+	}
+}
+
+// TestOriginChangesTakeResetsTheDroppedCount checks that a TakeOriginChanges
+// that reports dropped rows leaves the next one reporting none, so each drop
+// is logged once, not on every later write.
+func TestOriginChangesTakeResetsTheDroppedCount(t *testing.T) {
+	t.Parallel()
+
+	table := routetable.New(routetable.Params{})
+	addr := netip.MustParseAddr("10.0.0.0")
+
+	for range routetable.MaxWaitingOriginChanges + 1 {
+		table.Announce(netip.PrefixFrom(addr, 32), asA, routeOf(peerA, asA, 10))
+		addr = addr.Next()
+	}
+
+	_, dropped := table.TakeOriginChanges(routetable.MaxWaitingOriginChanges)
+	if dropped != 1 {
+		t.Fatalf("dropped = %d, want 1", dropped)
+	}
+
+	_, dropped = table.TakeOriginChanges(0)
+	if dropped != 0 {
+		t.Errorf("the next take's dropped = %d, want 0", dropped)
+	}
+}
+
+// TestOriginChangesPutBackTrimsToTheLimit checks that putting back what a
+// failed write took, after new changes filled the table again, drops the
+// oldest changes past the cap and counts them, so the cap still holds. The
+// clock advances one second per change, so a change's time says its age.
+func TestOriginChangesPutBackTrimsToTheLimit(t *testing.T) {
+	t.Parallel()
+
+	const (
+		taken = 20000
+		added = 20000
+	)
+
+	now := time.Unix(0, 0)
+	table := routetable.New(routetable.Params{
+		Now: func() time.Time {
+			now = now.Add(time.Second)
+
+			return now
+		},
+	})
+	addr := netip.MustParseAddr("10.0.0.0")
+
+	announce := func(n int) {
+		for range n {
+			table.Announce(netip.PrefixFrom(addr, 32), asA, routeOf(peerA, asA, 10))
+			addr = addr.Next()
+		}
+	}
+
+	announce(routetable.MaxWaitingOriginChanges)
+
+	takenChanges, _ := table.TakeOriginChanges(taken)
+
+	announce(added)
+
+	table.PutBackOriginChanges(takenChanges)
+
+	changes, dropped := table.TakeOriginChanges(math.MaxInt)
+	if dropped != added {
+		t.Errorf("dropped = %d, want %d", dropped, added)
+	}
+
+	if len(changes) != routetable.MaxWaitingOriginChanges {
+		t.Errorf("len(changes) = %d, want %d", len(changes),
+			routetable.MaxWaitingOriginChanges)
+	}
+
+	// The put-back rows are the oldest, so they are the ones dropped. The
+	// first remaining change is the one that followed them.
+	if want := time.Unix(int64(added+1), 0); !changes[0].Time.Equal(want) {
+		t.Errorf("first change's time = %v, want %v", changes[0].Time, want)
+	}
+}
+
+// checkOriginChanges checks that TakeOriginChanges, taking every change,
+// returns want.
+func checkOriginChanges(
+	t *testing.T, table *routetable.Table, want []routetable.OriginChange,
+) {
+	t.Helper()
+
+	changes, dropped := table.TakeOriginChanges(math.MaxInt)
+	if dropped != 0 {
+		t.Errorf("dropped = %d, want 0", dropped)
+	}
+
+	if !reflect.DeepEqual(changes, want) {
+		t.Errorf("TakeOriginChanges = %+v, want %+v", changes, want)
+	}
+}

internal/routetable/overview.go

@@ -79,8 +79,9 @@ type OriginPeers struct {
 
 // Overview works out the table's overview. It lists at most maxListed bogon
 // announcements and at most maxListed prefixes with more than one origin AS:
-// the first in prefix order. It walks over every prefix, but holds the read
+// the first in prefix order. It visits every prefix in the table, in batches
-// lock for walkBatch prefixes at a time.
+// of walkBatch. After each batch, it lets the table's lock go for as long as
+// it held the lock for the batch.
 func (t *Table) Overview(maxListed int) Overview {
 	var overview Overview
 
@@ -110,15 +111,18 @@ func (t *Table) Overview(maxListed int) Overview {
 }
 
 // walkPrefixes calls visit with every prefix of the table and its origins. It
-// holds the read lock, but lets it go after every walkBatch prefixes, so the
+// holds the read lock for walkBatch prefixes at a time. After each batch, it
-// updates waiting for it get in. A prefix that is added or removed in such a
+// lets the table's lock go for as long as it held the lock for the batch.
-// gap may or may not be visited, as Go allows for a map that changes during a
+// During a walk over many batches, updates can then take the lock about half
-// range over it. visit must not keep origins once it returns.
+// of the time. The walk may or may not visit a prefix that is added or removed
+// while the lock is let go. Go allows this in a range over a map that changes.
+// visit must not keep origins once it returns.
 func (t *Table) walkPrefixes(visit func(prefix netip.Prefix, origins []*origin)) {
 	t.mu.RLock()
 	defer t.mu.RUnlock()
 
 	visited := 0
+	batchStart := time.Now()
 
 	for _, f := range []*family{&t.ipv4, &t.ipv6} {
 		for _, prefixes := range f.byLength {
@@ -127,8 +131,13 @@ func (t *Table) walkPrefixes(visit func(prefix netip.Prefix, origins []*origin))
 
 				visited++
 				if visited%walkBatch == 0 {
+					held := time.Since(batchStart)
+
 					t.mu.RUnlock()
+					time.Sleep(held)
 					t.mu.RLock()
+
+					batchStart = time.Now()
 				}
 			}
 		}

internal/routetable/overview_internal_test.go

@@ -131,3 +131,61 @@ func TestWalkPrefixesLetsUpdatesIn(t *testing.T) {
 			"before its second batch")
 	}
 }
+
+// TestOverviewLetsUpdatesIn applies another peer's 1,000 announcements while
+// Overview walks a table with more prefixes than five of the walk's batches.
+// After each batch, the walk lets the lock go for as long as it held the lock.
+// So all 1,000 announcements must be applied before Overview returns. A walk
+// that takes the lock back at once lets in about one update per batch.
+func TestOverviewLetsUpdatesIn(t *testing.T) {
+	t.Parallel()
+
+	const (
+		asn       = 1000
+		peer      = "192.0.2.1"
+		otherPeer = "192.0.2.2"
+		// prefixCount is more prefixes than five of the walk's batches.
+		prefixCount   = 5*walkBatch + 1
+		announcements = 1000
+	)
+
+	routeTime := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
+	table := New(Params{})
+	prefixes := make([]netip.Prefix, 0, prefixCount)
+	addr := netip.MustParseAddr("1.0.0.0")
+
+	for range prefixCount {
+		prefix := netip.PrefixFrom(addr, addr.BitLen())
+		table.Announce(prefix, asn, Route{Peer: peer, Time: routeTime})
+		prefixes = append(prefixes, prefix)
+		addr = addr.Next()
+	}
+
+	overviewDone := make(chan struct{})
+
+	go func() {
+		defer close(overviewDone)
+
+		table.Overview(0)
+	}()
+
+	// Wait until Overview holds the read lock. TryLock fails while a reader
+	// holds it.
+	for table.mu.TryLock() {
+		table.mu.Unlock()
+		runtime.Gosched()
+	}
+
+	for _, prefix := range prefixes[:announcements] {
+		table.Announce(prefix, asn, Route{Peer: otherPeer, Time: routeTime})
+	}
+
+	select {
+	case <-overviewDone:
+		t.Fatal("Overview returned before another peer's 1,000 announcements " +
+			"were applied")
+	default:
+	}
+
+	<-overviewDone
+}

internal/routetable/reads_test.go

@@ -227,7 +227,7 @@ func TestCounts(t *testing.T) {
 	})
 
 	// peerC was the last peer of the IPv6 prefix. It keeps its number.
-	table.PeerDown(peerC)
+	table.PeersDown([]string{peerC})
 	checkCounts(t, table, routetable.Counts{
 		Pairs:                2,
 		Peers:                3,

internal/routetable/routetable.go

@@ -2,7 +2,9 @@
 // and origin AS, it keeps the peers whose current route to the prefix has that
 // origin AS. It also keeps the two newest of those routes in full. It applies
 // announcements, withdrawals and peers' sessions going down. It answers the
-// route reads that the pages and the API need.
+// route reads that the pages and the API need. It also records each prefix and
+// origin AS pair's origin changes: when it gains its first peer and when it
+// loses its last.
 package routetable
 
 import (
@@ -28,7 +30,7 @@ const (
 	keptTimesDegree = 32
 
 	// walkBatch is how many prefixes a walk over the whole table visits in
-	// one hold of the lock. Overview and PeerDown each walk the whole table.
+	// one hold of the lock. Overview and PeersDown each walk the whole table.
 	// Callers waiting for the lock get it between batches.
 	walkBatch = 10000
 
@@ -91,6 +93,9 @@ type Params struct {
 	// peer not in it gets the next number above all of them the first time it
 	// announces a route.
 	PeerNumbers map[string]uint
+	// Now returns the time RouteWatch applies a change at. It is time.Now
+	// unless a test gives one.
+	Now func() time.Time
 }
 
 // Table is the routing table. Its methods are safe for concurrent use.
@@ -126,11 +131,24 @@ type Table struct {
 	// originsWithoutKeptRoute counts the origins that peers carry but that
 	// keep no route.
 	originsWithoutKeptRoute int
+
+	// now returns the time RouteWatch applies a change at.
+	now func() time.Time
+	// originChanges holds the origin changes that wait to be written, oldest
+	// first. originChangesDropped counts those dropped since the last
+	// TakeOriginChanges.
+	originChanges        []OriginChange
+	originChangesDropped int
 }
 
 // New returns an empty table that numbers new peers on from
 // params.PeerNumbers.
 func New(params Params) *Table {
+	now := params.Now
+	if now == nil {
+		now = time.Now
+	}
+
 	t := &Table{
 		ipv4:          newFamily(ipv4Bits, minLookupMaskLengthV4),
 		ipv6:          newFamily(ipv6Bits, minLookupMaskLengthV6),
@@ -139,6 +157,7 @@ func New(params Params) *Table {
 		peerAddresses: make(map[uint]string, len(params.PeerNumbers)),
 		newPeers:      make(map[string]uint),
 		changed:       make(map[pair]struct{}),
+		now:           now,
 		keptTimes: btree.NewG(keptTimesDegree, func(a, b keptTime) bool {
 			return a.time.Before(b.time)
 		}),
@@ -205,20 +224,29 @@ func (t *Table) Withdraw(prefix netip.Prefix, peer string) {
 	}
 }
 
-// PeerDown applies a peer's session going down: the peer leaves every peer
+// PeersDown applies the sessions of peers going down: each of the peers leaves
-// set, and every route of the peer is dropped. When the peer carries any
+// every peer set, and every route of each is dropped. When any of them carries
-// prefix, PeerDown visits every prefix in the table, in batches of walkBatch.
+// a prefix, PeersDown visits every prefix in the table once, in batches of
-// After each batch, it lets the table's lock go for as long as it held the
+// walkBatch. After each batch, it lets the table's lock go for as long as it
-// lock for the batch. During a walk over many batches, reads and updates can
+// held the lock for the batch. During a walk over many batches, reads and
-// then take the lock about half of the time. An announcement by the peer
+// updates can then take the lock about half of the time. An announcement by
-// applied during the walk may be dropped too. So the caller applies the peer's
+// one of the peers applied during the walk may be dropped too. So the caller
-// later updates only once PeerDown has returned.
+// applies those peers' later updates only once PeersDown has returned.
-func (t *Table) PeerDown(peer string) {
+func (t *Table) PeersDown(peers []string) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 
+	// numbers holds the number of each of the peers that carries a prefix.
+	numbers := make([]uint, 0, len(peers))
+
+	for _, peer := range peers {
 		number, known := t.peerNumbers[peer]
-	if !known || t.peerPrefixes[number] == 0 {
+		if known && t.peerPrefixes[number] > 0 {
+			numbers = append(numbers, number)
+		}
+	}
+
+	if len(numbers) == 0 {
 		return
 	}
 
@@ -228,9 +256,15 @@ func (t *Table) PeerDown(peer string) {
 	for _, f := range []*family{&t.ipv4, &t.ipv6} {
 		for _, prefixes := range f.byLength {
 			for prefix, origins := range prefixes {
+				for _, number := range numbers {
 					o := originWithPeer(origins, number)
 					if o != nil {
 						t.leave(prefix, o, number)
+
+						// leave may have removed an origin of the
+						// prefix, or the prefix itself.
+						origins = prefixes[prefix]
+					}
 				}
 
 				// Go allows the map to change before the next step of
@@ -253,8 +287,9 @@ func (t *Table) PeerDown(peer string) {
 // Load puts one prefix and origin AS pair that the database holds into the
 // table, as startup does with each row it reads. The change's peers join the
 // pair's peer set, and its routes are kept as an announcement's would be.
-// Unlike an update, Load records no change, so TakeChanges does not hand the
+// Unlike an update, Load records no change and no origin change, so
-// pair back to be written again. Loading a pair twice changes nothing.
+// TakeChanges does not hand the pair back to be written again. Loading a pair
+// twice changes nothing.
 //
 // The caller makes sure that every peer change names, in its peer set or in
 // its routes, has a number in Params.PeerNumbers.
@@ -445,6 +480,7 @@ func (t *Table) removeOrigin(prefix netip.Prefix, o *origin) {
 // r in full when it is one of the two newest routes of o. The peer's earlier
 // route, if o keeps one, is replaced: it is no longer the peer's current route.
 // The pair is recorded as changed only when its peer set or kept routes change.
+// When r is the origin's first peer, an announced origin change is recorded.
 func (t *Table) join(prefix netip.Prefix, o *origin, r route) {
 	if slices.ContainsFunc(o.routes, r.equal) {
 		// The peer announced its kept route again.
@@ -454,6 +490,8 @@ func (t *Table) join(prefix netip.Prefix, o *origin, r route) {
 	wasWithoutKeptRoute := o.withoutKeptRoute()
 
 	joined := !o.peers.Test(r.peer)
+
+	firstPeer := joined && o.peers.None()
 	if joined {
 		o.peers.Set(r.peer)
 
@@ -467,6 +505,10 @@ func (t *Table) join(prefix netip.Prefix, o *origin, r route) {
 	if joined || dropped || kept {
 		t.recordChange(prefix, o, wasWithoutKeptRoute)
 	}
+
+	if firstPeer {
+		t.recordOriginChange(prefix, o.asn, true, t.prefixPeerCount(prefix))
+	}
 }
 
 // leave takes peer out of the peer set of o, an origin of prefix, and drops
@@ -525,7 +567,8 @@ func (t *Table) dropKeptRoute(o *origin, peer uint) bool {
 // recordChange records that the pair of prefix and the AS of o changed, and
 // brings the count of origins without a kept route up to date.
 // wasWithoutKeptRoute says whether o was such an origin before the change. An
-// origin that no peer carries any more is removed.
+// origin that no peer carries any more is removed, and a withdrawn origin
+// change is recorded.
 func (t *Table) recordChange(
 	prefix netip.Prefix, o *origin, wasWithoutKeptRoute bool,
 ) {
@@ -534,6 +577,7 @@ func (t *Table) recordChange(
 	t.countOriginWithoutKeptRoute(o, wasWithoutKeptRoute)
 
 	if o.peers.None() {
+		t.recordOriginChange(prefix, o.asn, false, t.prefixPeerCount(prefix))
 		t.removeOrigin(prefix, o)
 	}
 }

internal/routetable/routetable_internal_test.go

@@ -7,11 +7,11 @@ import (
 	"time"
 )
 
-// TestPeerDownLetsReadsIn reads the table's counts again and again while a peer
+// TestPeersDownLetsReadsIn reads the table's counts again and again while a
-// that carries more than two batches of prefixes goes down. PeerDown lets the
+// peer that carries more than two batches of prefixes goes down. PeersDown lets
-// lock go after every walkBatch prefixes, so a read must get in while the peer
+// the lock go after every walkBatch prefixes, so a read must get in while the
-// still carries some of its prefixes but no longer all.
+// peer still carries some of its prefixes but no longer all.
-func TestPeerDownLetsReadsIn(t *testing.T) {
+func TestPeersDownLetsReadsIn(t *testing.T) {
 	t.Parallel()
 
 	const (
@@ -32,8 +32,8 @@ func TestPeerDownLetsReadsIn(t *testing.T) {
 		addr = addr.Next()
 	}
 
-	// Hold a read lock until PeerDown waits for the lock, so that the first
+	// Hold a read lock until PeersDown waits for the lock, so that the first
-	// read below starts once PeerDown holds it. TryRLock fails once a writer
+	// read below starts once PeersDown holds it. TryRLock fails once a writer
 	// waits.
 	table.mu.RLock()
 
@@ -42,7 +42,7 @@ func TestPeerDownLetsReadsIn(t *testing.T) {
 	go func() {
 		defer close(peerDownDone)
 
-		table.PeerDown(peer)
+		table.PeersDown([]string{peer})
 	}()
 
 	for table.mu.TryRLock() {
@@ -60,7 +60,7 @@ func TestPeerDownLetsReadsIn(t *testing.T) {
 
 		select {
 		case <-peerDownDone:
-			t.Fatal("no read got in while PeerDown ran")
+			t.Fatal("no read got in while PeersDown ran")
 		default:
 		}
 	}

internal/routetable/routetable_test.go

@@ -69,10 +69,10 @@ func withdraw(peer string) step {
 	}
 }
 
-// peerDown is the step of peer's session going down.
+// peersDown is the step of the sessions of peers going down.
-func peerDown(peer string) step {
+func peersDown(peers ...string) step {
 	return func(table *routetable.Table, _ netip.Prefix) {
-		table.PeerDown(peer)
+		table.PeersDown(peers)
 	}
 }
 
@@ -272,21 +272,28 @@ func TestWithdrawalsAndPeerDowns(t *testing.T) {
 		{
 			name: "a peer going down leaves the prefix",
 			steps: []step{announce(peerA, asA, 10), announce(peerB, asA, 20),
-				peerDown(peerB)},
+				peersDown(peerB)},
 			want: []routetable.Origin{{ASN: asA, Peers: []string{peerA},
 				Routes: []routetable.Route{routeOf(peerA, asA, 10)}}},
 		},
 		{
 			name: "a peer going down after it withdrew changes nothing",
 			steps: []step{announce(peerA, asA, 10), announce(peerB, asA, 20),
-				withdraw(peerB), peerDown(peerB)},
+				withdraw(peerB), peersDown(peerB)},
 			want: []routetable.Origin{{ASN: asA, Peers: []string{peerA},
 				Routes: []routetable.Route{routeOf(peerA, asA, 10)}}},
 		},
+		{
+			name: "peers going down together each leave the prefix",
+			steps: []step{announce(peerA, asA, 10), announce(peerB, asB, 20),
+				announce(peerC, asB, 30), peersDown(peerA, peerB)},
+			want: []routetable.Origin{{ASN: asB, Peers: []string{peerC},
+				Routes: []routetable.Route{routeOf(peerC, asB, 30)}}},
+		},
 	})
 }
 
-func TestPeerDownLeavesEveryPrefix(t *testing.T) {
+func TestPeersDownLeavesEveryPrefix(t *testing.T) {
 	t.Parallel()
 
 	table := routetable.New(routetable.Params{})
@@ -303,7 +310,7 @@ func TestPeerDownLeavesEveryPrefix(t *testing.T) {
 	}
 
 	table.Announce(prefixes[0], asB, routeOf(peerB, asB, 20))
-	table.PeerDown(peerA)
+	table.PeersDown([]string{peerA})
 
 	want := []routetable.Origin{{ASN: asB, Peers: []string{peerB},
 		Routes: []routetable.Route{routeOf(peerB, asB, 20)}}}
@@ -333,6 +340,32 @@ func TestPeerDownLeavesEveryPrefix(t *testing.T) {
 	}
 }
 
+// TestPeersDownTakesOutPeersOfDifferentPrefixes takes peerA and peerB out in
+// one PeersDown. peerA carries only prefixV4, and peerB only prefixV6. So
+// peerA's leaving removes prefixV4 before PeersDown looks for peerB there. Both
+// prefixes must then be gone.
+func TestPeersDownTakesOutPeersOfDifferentPrefixes(t *testing.T) {
+	t.Parallel()
+
+	table := routetable.New(routetable.Params{})
+	prefixes := []netip.Prefix{
+		netip.MustParsePrefix(prefixV4),
+		netip.MustParsePrefix(prefixV6),
+	}
+
+	table.Announce(prefixes[0], asA, routeOf(peerA, asA, 10))
+	table.Announce(prefixes[1], asB, routeOf(peerB, asB, 20))
+	table.PeersDown([]string{peerA, peerB})
+
+	for _, prefix := range prefixes {
+		got := table.Origins(prefix)
+		if got != nil {
+			t.Errorf("Origins(%s) = %+v after its only peer went down, want nil",
+				prefix, got)
+		}
+	}
+}
+
 func TestCountsOriginsWithoutKeptRoute(t *testing.T) {
 	t.Parallel()
 
@@ -358,7 +391,7 @@ func TestCountsOriginsWithoutKeptRoute(t *testing.T) {
 		{announce(peerC, asA, 30), 0},
 		{withdraw(peerB), 0},
 		{withdraw(peerC), 1},
-		{peerDown(peerA), 0},
+		{peersDown(peerA), 0},
 	}
 
 	for i, s := range steps {
@@ -464,7 +497,7 @@ func TestTakeChangesHoldsOnlyPairsThatChanged(t *testing.T) {
 	checkChanges(t, table, []change{}, map[string]uint{})
 
 	// peerA going down changes v4, the one prefix it carries.
-	table.PeerDown(peerA)
+	table.PeersDown([]string{peerA})
 	checkChanges(t, table, []change{
 		{Prefix: prefixV4, OriginASN: asA, Peers: []uint{1, 2},
 			Routes: []routetable.Route{
@@ -473,7 +506,7 @@ func TestTakeChangesHoldsOnlyPairsThatChanged(t *testing.T) {
 	}, map[string]uint{})
 
 	// peerA, which carries nothing now, goes down again.
-	table.PeerDown(peerA)
+	table.PeersDown([]string{peerA})
 	checkChanges(t, table, []change{}, map[string]uint{})
 }
 
@@ -617,7 +650,7 @@ func TestConcurrentUse(t *testing.T) {
 		func(round int) { table.Announce(prefix, asA, routeOf(peerA, asA, round)) },
 		func(round int) { table.Announce(prefix, asB, routeOf(peerB, asB, round)) },
 		func(int) { table.Withdraw(prefix, peerA) },
-		func(int) { table.PeerDown(peerB) },
+		func(int) { table.PeersDown([]string{peerB}) },
 		func(int) { table.PutBackChanges(table.TakeChanges(math.MaxInt)) },
 		func(int) { table.ChangeCount() },
 		func(int) { table.Lookup(prefix.Addr()) },

internal/routewatch/app_integration_internal_test.go

@@ -330,6 +330,23 @@ func (m *mockStore) ReadRouteTable(_ context.Context) (*routetable.Table, error)
 	return routetable.New(routetable.Params{}), nil
 }
 
+// WriteOriginChanges records nothing: the mock keeps no origin changes.
+func (m *mockStore) WriteOriginChanges(_ []routetable.OriginChange) error {
+	return nil
+}
+
+// GetOriginChanges returns no origin changes.
+func (m *mockStore) GetOriginChanges(
+	_ context.Context, _ netip.Prefix, _ int,
+) ([]routetable.OriginChange, error) {
+	return nil, nil
+}
+
+// PruneOriginChanges does nothing.
+func (m *mockStore) PruneOriginChanges(_ context.Context) error {
+	return nil
+}
+
 // GetOrCreateASNBatch mock implementation
 func (m *mockStore) GetOrCreateASNBatch(asns map[int]database.SeenTimes) error {
 	m.mu.Lock()

internal/routewatch/batchwrite_internal_test.go

@@ -179,6 +179,7 @@ type stopLog struct {
 	Updates       int    `json:"updates"`
 	TableChanges  int    `json:"table_changes"`
 	HeldMessages  int    `json:"held_messages"`
+	OriginChanges int    `json:"origin_changes"`
 	Paths         int    `json:"paths"`
 	BatchEntries  int    `json:"batch_entries"`
 }
@@ -472,6 +473,42 @@ func (s tableWriteBlockingStore) WriteRouteChanges(
 	return s.mockStore.WriteRouteChanges(changes, newPeers)
 }
 
+// stopPrefixHandlerWithBlockedWrite stops the prefix handler, whose store
+// holds the write that fill starts, after handing it the message extra, and
+// returns what Stop logged. blocking is the store the write waits on; store is
+// a blockingStore wrapped for the routing table's changes or the origin
+// changes.
+func stopPrefixHandlerWithBlockedWrite(
+	t *testing.T,
+	blocking *blockingStore,
+	store database.Store,
+	logs *bytes.Buffer,
+	fill, extra *ristypes.RISMessage,
+) stopLog {
+	t.Helper()
+
+	h := NewPrefixHandler(store, newErrorLogger(logs),
+		routetable.New(routetable.Params{}))
+	stopWhileAWriteIsBlocked(t, blocking, h, h.done,
+		[]*ristypes.RISMessage{fill}, extra)
+
+	return lastLogged(t, logs)
+}
+
+// originChangesBlockingStore is a blockingStore whose writes of the origin
+// changes also wait until the test calls releaseWrites.
+type originChangesBlockingStore struct {
+	*blockingStore
+}
+
+func (s originChangesBlockingStore) WriteOriginChanges(
+	changes []routetable.OriginChange,
+) error {
+	s.waitForRelease()
+
+	return s.mockStore.WriteOriginChanges(changes)
+}
+
 // TestPrefixHandlerStopCountsTheTableChangesOfAWriteInProgress holds the write
 // of the routing table's changes that a full batch starts, then hands the
 // prefix handler one more prefix and stops it with a deadline that passes
@@ -485,19 +522,67 @@ func TestPrefixHandlerStopCountsTheTableChangesOfAWriteInProgress(t *testing.T)
 	var logs bytes.Buffer
 
 	blocking := newBlockingStore()
-	h := NewPrefixHandler(tableWriteBlockingStore{blockingStore: blocking},
-		newErrorLogger(&logs), routetable.New(routetable.Params{}))
 	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
-	stopWhileAWriteIsBlocked(t, blocking, h, h.done,
+	logged := stopPrefixHandlerWithBlockedWrite(t, blocking,
-		[]*ristypes.RISMessage{prefixAnnouncement(fill)},
+		tableWriteBlockingStore{blockingStore: blocking}, &logs,
-		prefixAnnouncement([]string{extraPrefix}))
+		prefixAnnouncement(fill), prefixAnnouncement([]string{extraPrefix}))
 
-	if logged := lastLogged(t, &logs); logged.TableChanges != 2 {
+	if logged.TableChanges != 2 {
 		t.Errorf("Stop logged %d table changes not written, want 2",
 			logged.TableChanges)
 	}
 }
 
+// TestPrefixHandlerStopCountsTheOriginChangesOfAWriteInProgress holds the
+// write of the origin changes that a full batch starts, then hands the prefix
+// handler one more prefix and stops it with a deadline that passes meanwhile.
+// Stop must log as not written both the change the held write took and the one
+// since.
+func TestPrefixHandlerStopCountsTheOriginChangesOfAWriteInProgress(t *testing.T) {
+	t.Parallel()
+
+	const extraPrefix = "198.51.100.0/24"
+
+	var logs bytes.Buffer
+
+	blocking := newBlockingStore()
+	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
+	logged := stopPrefixHandlerWithBlockedWrite(t, blocking,
+		originChangesBlockingStore{blockingStore: blocking}, &logs,
+		prefixAnnouncement(fill), prefixAnnouncement([]string{extraPrefix}))
+
+	if logged.OriginChanges != 2 {
+		t.Errorf("Stop logged %d origin changes not written, want 2",
+			logged.OriginChanges)
+	}
+}
+
+// TestPrefixHandlerStopCountsTheWaitingOriginChanges stops the prefix handler
+// with a deadline that has already passed, so the flush goroutine skips its
+// last write. The origin changes still waiting in the routing table are then
+// not written, and Stop must log them.
+func TestPrefixHandlerStopCountsTheWaitingOriginChanges(t *testing.T) {
+	t.Parallel()
+
+	var logs bytes.Buffer
+
+	store := newMockStore()
+	h := newPrefixHandler(store, newErrorLogger(&logs),
+		routetable.New(routetable.Params{}), longerThanAnyTest,
+		maxTableChangesPerWrite)
+
+	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix, testV6Prefix))
+
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	h.Stop(ctx)
+
+	if logged := lastLogged(t, &logs); logged.OriginChanges != 2 {
+		t.Errorf("Stop logged %d origin changes not written, want 2",
+			logged.OriginChanges)
+	}
+}
+
 // TestASHandlerSkipsItsLastWriteAfterTheStopDeadline also checks what Stop
 // logs: the asnBatchSize entries of the held write and the entry for testASNB,
 // handled since, are not written.

internal/routewatch/dbmaintainer.go

@@ -25,6 +25,10 @@ const (
 	// analyzeInterval is how often to run ANALYZE.
 	analyzeInterval = 1 * time.Hour
 
+	// originChangesPruneInterval is how often to prune the origin_changes
+	// table.
+	originChangesPruneInterval = 1 * time.Hour
+
 	// checkpointTimeout is the max time for WAL checkpoint.
 	checkpointTimeout = 10 * time.Second
 
@@ -33,6 +37,9 @@ const (
 
 	// analyzeTimeout is the max time for ANALYZE.
 	analyzeTimeout = 5 * time.Minute
+
+	// originChangesPruneTimeout is the max time for the origin changes prune.
+	originChangesPruneTimeout = 5 * time.Minute
 )
 
 // checkpointIncompleteMessage is logged, at debug level like a completed
@@ -80,6 +87,7 @@ func (m *DBMaintainer) Start(ctx context.Context) {
 		"checkpoint_interval", checkpointInterval,
 		"vacuum_interval", vacuumInterval,
 		"analyze_interval", analyzeInterval,
+		"origin_changes_prune_interval", originChangesPruneInterval,
 	)
 }
 
@@ -111,6 +119,9 @@ func (m *DBMaintainer) run(ctx context.Context) {
 	analyzeTimer := time.NewTimer(analyzeInterval)
 	defer analyzeTimer.Stop()
 
+	pruneTimer := time.NewTimer(originChangesPruneInterval)
+	defer pruneTimer.Stop()
+
 	for {
 		select {
 		case <-m.stopCh:
@@ -127,6 +138,10 @@ func (m *DBMaintainer) run(ctx context.Context) {
 		case <-analyzeTimer.C:
 			m.runAnalyze(ctx)
 			analyzeTimer.Reset(analyzeInterval)
+
+		case <-pruneTimer.C:
+			m.runPruneOriginChanges(ctx)
+			pruneTimer.Reset(originChangesPruneInterval)
 		}
 	}
 }
@@ -175,6 +190,18 @@ func (m *DBMaintainer) runAnalyze(ctx context.Context) {
 	}
 }
 
+// runPruneOriginChanges prunes the origin_changes table.
+func (m *DBMaintainer) runPruneOriginChanges(ctx context.Context) {
+	duration, err := runWithTimeout(ctx, originChangesPruneTimeout,
+		m.db.PruneOriginChanges)
+	if err != nil {
+		m.logger.Error("Origin changes prune failed", "error", err,
+			"duration", duration)
+	} else {
+		m.logger.Debug("Origin changes pruned", "duration", duration)
+	}
+}
+
 // runWithTimeout runs operation with a context that ends after timeout and
 // returns how long it took and its error.
 func runWithTimeout(

internal/routewatch/prefixhandler.go

@@ -15,12 +15,15 @@ import (
 )
 
 const (
-	// prefixHandlerQueueSize is the queue capacity for prefix tracking
+	// prefixHandlerQueueSize is the most messages the handler's queue holds.
-	// operations, about 4 seconds of feed at peak. The streamer drops updates
+	// While the routing table's Overview or PeersDown walks every prefix, it
-	// rather than blocks when a queue is full (a STATE message is held until
+	// holds the table's lock about half of the time. The handler then applies
-	// there is room), so this bounds memory. Batches still flush on a timer
+	// updates more slowly. When the feed pauses for some seconds and then
-	// (prefixBatchTimeout).
+	// catches up, its backlog can arrive faster than that, and the queue
-	prefixHandlerQueueSize = 20000
+	// holds the backlog. The streamer drops an update that finds the queue
+	// full. A message takes about 1 KiB, so a full queue takes about 100 MiB
+	// (README "Memory").
+	prefixHandlerQueueSize = 100000
 
 	// prefixBatchSize is the number of prefix updates to batch together
 	prefixBatchSize = 25000
@@ -45,11 +48,24 @@ const (
 	// going down changes are written in about a minute.
 	maxTableChangesPerWrite = 20000
 
+	// maxOriginChangesPerWrite is the most origin changes of the routing
+	// table that one flush writes; the rest wait for the next flushes.
+	maxOriginChangesPerWrite = 20000
+
 	// maxHeldBackMessages is the most messages the handler holds back from
-	// peers whose PeerDown has not returned. A message takes about 1 KiB, so
+	// peers whose session went down. A peer's messages are held back from
-	// that many take about 20 MiB (README "Memory"). At this number
+	// the moment its session goes down until peerDown has applied them. If
-	// HandleMessage waits until one of them is applied.
+	// peerDown is running at that moment, the peer waits for the rest of
-	maxHeldBackMessages = 20000
+	// peerDown's current PeersDown call. The peer also waits while peerDown
+	// applies the messages held back from that call's peers, those that
+	// arrive meanwhile included. The next PeersDown call takes the peer out.
+	// Once that call returns, peerDown applies the messages held back from
+	// that call's peers, one peer after another, in the order the call lists
+	// them. A held-back message that says the peer's session went down again
+	// makes the peer's later messages wait for the next call. A message takes
+	// about 1 KiB, so 150,000 take about 150 MiB (README "Memory"). At this
+	// number HandleMessage waits until one of them is applied.
+	maxHeldBackMessages = 150000
 )
 
 // PrefixHandler keeps the routing table in memory current from the feed. It
@@ -58,17 +74,19 @@ const (
 // does an announcement whose AS path is empty or holds AS 0 anywhere. A peer
 // whose BGP session goes down leaves every prefix.
 //
-// The table's PeerDown applies a session going down. It visits every prefix,
+// The table's PeersDown applies sessions going down. It visits every prefix,
-// which takes seconds on a full table. So the handler runs PeerDown on
+// which takes seconds on a full table. So the handler runs PeersDown on a
-// a separate goroutine, one peer at a time, and goes on taking messages
+// separate goroutine and goes on taking messages meanwhile. One PeersDown runs
-// meanwhile. PeerDown lets the table's lock go about half of the time, so the
+// at a time. The next one takes every peer whose session went down since the
-// handler applies other peers' messages while PeerDown runs. The handler holds
+// previous one began. PeersDown lets the table's lock go about half of the
-// back the down peer's later messages until PeerDown returns. Then the handler
+// time, so the handler applies other peers' messages while PeersDown runs. The
-// applies them in order, one at a time, and takes new messages in between. It
+// handler holds back a down peer's later messages until peerDown applies them,
-// holds back at most maxHeldBackMessages messages.
+// as the doc of maxHeldBackMessages describes. It applies them in order, one at
+// a time, and takes new messages in between. It holds back at most
+// maxHeldBackMessages messages.
 //
-// Each flush writes the table's changes to the database. It also stores each
+// Each flush writes the table's changes and origin changes to the database.
-// prefix the updates name with the time it was first seen.
+// It also stores each prefix the updates name with the time it was first seen.
 type PrefixHandler struct {
 	db      database.Store
 	logger  *logger.Logger
@@ -81,11 +99,14 @@ type PrefixHandler struct {
 	// tableChangesPerWrite is maxTableChangesPerWrite, except in tests that
 	// give a smaller one.
 	tableChangesPerWrite int
+	// originChangesPerWrite is maxOriginChangesPerWrite, except in tests that
+	// give a smaller one.
+	originChangesPerWrite int
 
-	// Batching. mu guards batch, writing and writingTableChanges: the
+	// Batching. mu guards batch, writing, writingTableChanges and
-	// flush goroutine takes them under mu and writes them after releasing
+	// writingOriginChanges: the flush goroutine takes them under mu and
-	// mu, so HandleMessage waits for a database write only while batch is at
+	// writes them after releasing mu, so HandleMessage waits for a database
-	// batchMaxSize.
+	// write only while batch is at batchMaxSize.
 	mu    sync.Mutex
 	batch []seenPrefix
 	// writing is how many updates the flush goroutine took and has not
@@ -94,6 +115,9 @@ type PrefixHandler struct {
 	// writingTableChanges is how many changes the flush goroutine took from
 	// the routing table and has not finished writing.
 	writingTableChanges int
+	// writingOriginChanges is how many origin changes the flush goroutine
+	// took from the routing table and has not finished writing.
+	writingOriginChanges int
 	// batchTaken is signalled when the flush goroutine takes batch; its L is
 	// &mu.
 	batchTaken  sync.Cond
@@ -103,20 +127,25 @@ type PrefixHandler struct {
 	stopCh chan context.Context
 	done   chan struct{} // closed when the flush goroutine returns
 
-	// Peers whose session went down. peerDownMu lets one PeerDown run at a
+	// Peers whose session went down. heldBack, under mu, holds each such
-	// time, so the handler's own updates wait for the table's lock behind
+	// peer's later messages, in order, until the peer has left every prefix
-	// at most one. heldBack, under mu, holds each such peer's later
+	// and they are applied. A peer with no such messages yet has an empty
-	// messages, in order, until its PeerDown has returned and they are
+	// entry. heldCount, under mu, is how many messages heldBack holds.
-	// applied. A peer with no such messages yet has an empty entry.
-	// heldCount, under mu, is how many messages heldBack holds.
 	// heldTaken is signalled each time a held-back message is taken out to
-	// be applied; its L is &mu. peerDowns counts the peerDown goroutines
+	// be applied; its L is &mu.
-	// that have not returned.
-	peerDownMu sync.Mutex
 	heldBack  map[string][]*ristypes.RISMessage
 	heldCount int
 	heldTaken sync.Cond
+	// downPeers, under mu, lists the peers whose session went down and that
+	// the next PeersDown takes. peerDownRunning, under mu, says whether
+	// peerDown is running. peerDowns counts the peerDown goroutines that have
+	// not returned.
+	downPeers       []string
+	peerDownRunning bool
 	peerDowns       sync.WaitGroup
+	// tablePeersDown is the table's PeersDown. Only peerDown calls it, so a
+	// test may replace it before it hands the handler a message.
+	tablePeersDown func(peers []string)
 }
 
 // seenPrefix is what the batch keeps of an update: the prefix it names and its
@@ -177,10 +206,12 @@ func newPrefixHandler(
 		batch:                 make([]seenPrefix, 0, prefixBatchSize),
 		batchMaxSize:          prefixBatchMaxSize,
 		tableChangesPerWrite:  tableChangesPerWrite,
+		originChangesPerWrite: maxOriginChangesPerWrite,
 		batchFullCh:           make(chan struct{}, 1),
 		stopCh:                make(chan context.Context, 1),
 		done:                  make(chan struct{}),
 		heldBack:              make(map[string][]*ristypes.RISMessage),
+		tablePeersDown:        table.PeersDown,
 	}
 	h.batchTaken.L = &h.mu
 	h.heldTaken.L = &h.mu
@@ -212,7 +243,7 @@ func (h *PrefixHandler) QueueCapacity() int {
 }
 
 // HandleMessage processes a message to track prefix information. A message
-// from a peer whose PeerDown has not returned is held back until it has.
+// from a peer whose session went down is held back until peerDown applies it.
 func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
@@ -235,17 +266,26 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	h.applyMessage(msg)
 }
 
-// Stop waits for every PeerDown the handler started and for the messages held
+// HeldBackCount returns how many messages the handler holds back from peers
+// whose session went down. The status page shows it.
+func (h *PrefixHandler) HeldBackCount() int {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	return h.heldCount
+}
+
+// Stop waits for every PeersDown the handler started and for the messages held
 // back meanwhile to be applied. It gives up this wait once ctx ends. Then Stop
 // stops the flush goroutine and waits for it until ctx ends. The flush
 // goroutine makes one last flush before it returns. That flush writes at most
-// tableChangesPerWrite of the routing table's changes, and the prefixes of the
+// tableChangesPerWrite of the routing table's changes, originChangesPerWrite of
-// updates the handler holds. Once ctx has ended the flush goroutine skips that
+// its origin changes, and the prefixes of the updates the handler holds. Once
-// last flush, and a write still running is lost when the process exits. A
+// ctx has ended the flush goroutine skips that last flush, and a write still
-// failed write of the table's changes is not retried either. Stop logs how
+// running is lost when the process exits. A failed write of the table's
-// many updates and table changes are not written. It also logs how many
+// changes is not retried either. Stop logs how many updates, table changes and
-// held-back messages are not applied. The caller calls Stop once its last
+// origin changes are not written. It also logs how many held-back messages are
-// HandleMessage has returned.
+// not applied. The caller calls Stop once its last HandleMessage has returned.
 func (h *PrefixHandler) Stop(ctx context.Context) {
 	peerDownsDone := make(chan struct{})
 
@@ -270,28 +310,30 @@ func (h *PrefixHandler) Stop(ctx context.Context) {
 	unwritten := h.writing + len(h.batch)
 	tableChanges := h.writingTableChanges + h.table.ChangeCount()
 	heldMessages := h.heldCount
+	originChanges := h.writingOriginChanges + h.table.OriginChangeCount()
 	h.mu.Unlock()
 
-	if unwritten == 0 && tableChanges == 0 && heldMessages == 0 {
+	if unwritten == 0 && tableChanges == 0 && heldMessages == 0 &&
+		originChanges == 0 {
 		return
 	}
 
 	if ctx.Err() != nil {
 		h.logger.Error("Stop deadline passed before the prefix handler wrote "+
 			"its updates", "updates", unwritten, "table_changes", tableChanges,
-			"held_messages", heldMessages)
+			"held_messages", heldMessages, "origin_changes", originChanges)
 
 		return
 	}
 
 	h.logger.Error("The prefix handler stopped before it wrote everything",
 		"updates", unwritten, "table_changes", tableChanges,
-		"held_messages", heldMessages)
+		"held_messages", heldMessages, "origin_changes", originChanges)
 }
 
 // applyMessage applies msg. The caller holds mu. A peer's session going down
-// starts the peer's PeerDown. An update is applied to the routing table, and
+// has the peer leave every prefix. An update is applied to the routing table,
-// its prefixes join the batch.
+// and its prefixes join the batch.
 func (h *PrefixHandler) applyMessage(msg *ristypes.RISMessage) {
 	if msg.Type == ristypes.MessageTypeState {
 		if msg.State == ristypes.StateDown {
@@ -370,34 +412,64 @@ func (h *PrefixHandler) applyMessage(msg *ristypes.RISMessage) {
 	}
 }
 
-// startPeerDown starts peerDown for peer on a new goroutine. The caller holds
+// startPeerDown adds peer to downPeers. It starts peerDown on a new goroutine
-// mu. The peer's messages are held back until peerDown has applied them.
+// unless peerDown is running. The caller holds mu. The peer's messages are
-// Messages held back for an earlier PeerDown of the peer, and not yet applied,
+// held back until peerDown has applied them. Messages held back for an earlier
-// stay held back.
+// session of the peer going down, and not yet applied, stay held back.
 func (h *PrefixHandler) startPeerDown(peer string) {
 	_, goingDown := h.heldBack[peer]
 	if !goingDown {
 		h.heldBack[peer] = []*ristypes.RISMessage{}
 	}
 
+	h.downPeers = append(h.downPeers, peer)
+
+	if h.peerDownRunning {
+		return
+	}
+
+	h.peerDownRunning = true
 	h.peerDowns.Add(1)
 
-	go h.peerDown(peer)
+	go h.peerDown()
 }
 
-// peerDown runs the table's PeerDown for peer, once no other PeerDown is
+// peerDown takes every peer in downPeers and runs the table's PeersDown for
-// running. Then peerDown applies the messages held back from the peer, in
+// them. Then it applies the messages held back from each of those peers. It
-// order. It takes mu for one message at a time, so HandleMessage takes new
+// repeats this until downPeers is empty.
-// messages in between. The peer's new messages join the end of the held-back
+func (h *PrefixHandler) peerDown() {
-// ones. A held-back message that says the peer's session went down again
-// starts a new peerDown, which applies the rest.
-func (h *PrefixHandler) peerDown(peer string) {
 	defer h.peerDowns.Done()
 
-	h.peerDownMu.Lock()
+	for {
-	h.table.PeerDown(peer)
+		h.mu.Lock()
-	h.peerDownMu.Unlock()
 
+		peers := h.downPeers
+		h.downPeers = nil
+
+		if len(peers) == 0 {
+			h.peerDownRunning = false
+			h.mu.Unlock()
+
+			return
+		}
+
+		h.mu.Unlock()
+
+		h.tablePeersDown(peers)
+
+		for _, peer := range peers {
+			h.applyHeldBack(peer)
+		}
+	}
+}
+
+// applyHeldBack applies the messages held back from peer, in order. It takes
+// mu for one message at a time, so HandleMessage takes new messages in
+// between. The peer's new messages join the end of the held-back ones. A
+// held-back message that says the peer's session went down again adds the
+// peer to downPeers. The rest stay held back until the next PeersDown has
+// returned.
+func (h *PrefixHandler) applyHeldBack(peer string) {
 	for {
 		h.mu.Lock()
 
@@ -443,11 +515,12 @@ func (h *PrefixHandler) takeBatch() []seenPrefix {
 	return batch
 }
 
-// flushBatch writes the routing table's changes, then takes the pending batch
+// flushBatch writes the routing table's changes and origin changes, then
-// and stores the prefixes it names, without holding the mutex during the
+// takes the pending batch and stores the prefixes it names, without holding
-// writes.
+// the mutex during the writes.
 func (h *PrefixHandler) flushBatch() {
 	h.writeTableChanges()
+	h.writeOriginChanges()
 
 	batch := h.takeBatch()
 	if len(batch) == 0 {
@@ -549,6 +622,39 @@ func (h *PrefixHandler) writeTableChanges() {
 	}
 }
 
+// writeOriginChanges writes at most originChangesPerWrite of the routing
+// table's origin changes. When the write fails, they go back to the table, so
+// that a later write writes them again. A dropped change is logged as a
+// warning.
+func (h *PrefixHandler) writeOriginChanges() {
+	h.mu.Lock()
+	changes, dropped := h.table.TakeOriginChanges(h.originChangesPerWrite)
+	h.writingOriginChanges = len(changes)
+	h.mu.Unlock()
+
+	if dropped > 0 {
+		h.logger.Warn("Origin changes dropped", "dropped", dropped)
+	}
+
+	if len(changes) == 0 {
+		return
+	}
+
+	err := h.db.WriteOriginChanges(changes)
+
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	h.writingOriginChanges = 0
+
+	if err != nil {
+		h.logger.Error("Failed to write the origin changes; the next flush "+
+			"retries", "changes", len(changes), "error", err)
+
+		h.table.PutBackOriginChanges(changes)
+	}
+}
+
 // canonicalPrefix returns prefix in the text form net/netip prints, the form
 // the prefix tables store and the IP lookup reads them in. The feed sends IPv6
 // announcements compressed ("2001:db8::/32") but IPv6 withdrawals uncompressed

internal/routewatch/prefixhandler_internal_test.go

@@ -238,7 +238,7 @@ func TestPrefixHandlerStopLogsNothingWhenItWroteEverything(t *testing.T) {
 }
 
 // TestPrefixHandlerStopLogsTheMessagesHeldBack stops a prefix handler with a
-// deadline that has passed, while a peer's PeerDown is held up and one message
+// deadline that has passed, while a peer's PeersDown is held up and one message
 // of the peer is held back. Stop must log that message as not applied.
 func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) {
 	t.Parallel()
@@ -249,8 +249,8 @@ func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) {
 		routetable.New(routetable.Params{}), longerThanAnyTest,
 		maxTableChangesPerWrite)
 
-	h.peerDownMu.Lock()
+	release := holdUpPeersDown(h)
-	defer h.peerDownMu.Unlock()
+	defer close(release)
 
 	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
 	h.HandleMessage(sessionDown())
@@ -268,8 +268,8 @@ func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) {
 }
 
 // TestPrefixHandlerStopWaitsForTheMessagesHeldBack stops a prefix handler
-// while a peer's PeerDown is held up and an announcement of the peer is held
+// while a peer's PeersDown is held up and an announcement of the peer is held
-// back. Stop must wait. Once the PeerDown is let go and Stop returns, the
+// back. Stop must wait. Once the PeersDown is let go and Stop returns, the
 // announcement must be in the database.
 func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) {
 	t.Parallel()
@@ -278,7 +278,7 @@ func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) {
 	h := newPrefixHandler(db, logger.New(), routetable.New(routetable.Params{}),
 		longerThanAnyTest, maxTableChangesPerWrite)
 
-	h.peerDownMu.Lock()
+	release := holdUpPeersDown(h)
 	h.HandleMessage(sessionDown())
 	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
 
@@ -291,16 +291,16 @@ func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) {
 
 	select {
 	case <-stopped:
-		t.Error("Stop returned while a PeerDown was held up")
+		t.Error("Stop returned while a PeersDown was held up")
 	case <-time.After(stillWaitingAfter):
 	}
 
-	h.peerDownMu.Unlock()
+	close(release)
 
 	select {
 	case <-stopped:
 	case <-time.After(stepTimeout):
-		t.Fatalf("Stop did not return within %s of the PeerDown's release",
+		t.Fatalf("Stop did not return within %s of the PeersDown's release",
 			stepTimeout)
 	}
 
@@ -324,16 +324,33 @@ const longerThanAnyTest = time.Hour
 // applies the updates to an empty routing table. It is built without
 // NewPrefixHandler's flush goroutine, so the test flushes itself.
 func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler {
+	table := routetable.New(routetable.Params{})
+
 	return &PrefixHandler{
 		db:                    store,
 		logger:                logger.New(),
-		table:                routetable.New(routetable.Params{}),
+		table:                 table,
 		batchMaxSize:          prefixBatchMaxSize,
 		tableChangesPerWrite:  maxTableChangesPerWrite,
+		originChangesPerWrite: maxOriginChangesPerWrite,
 		heldBack:              make(map[string][]*ristypes.RISMessage),
+		tablePeersDown:        table.PeersDown,
 	}
 }
 
+// holdUpPeersDown makes each PeersDown of h wait until the returned channel is
+// closed, as another peer's long PeersDown would hold it up. The test calls it
+// before it hands h a message.
+func holdUpPeersDown(h *PrefixHandler) chan struct{} {
+	release := make(chan struct{})
+	h.tablePeersDown = func(peers []string) {
+		<-release
+		h.table.PeersDown(peers)
+	}
+
+	return release
+}
+
 // newTestDatabase opens a database in a new temporary directory. It is closed
 // when the test ends.
 func newTestDatabase(t *testing.T) *database.Database {

internal/routewatch/prefixhandler_originchanges_internal_test.go

@@ -0,0 +1,181 @@
+package routewatch
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"log/slog"
+	"net/netip"
+	"slices"
+	"testing"
+
+	"git.eeqj.de/sneak/routewatch/internal/database"
+	"git.eeqj.de/sneak/routewatch/internal/logger"
+	"git.eeqj.de/sneak/routewatch/internal/routetable"
+)
+
+// TestPrefixHandlerWritesOriginChanges hands a running prefix handler an
+// announcement and flushes. The flush must write the origin change to the
+// database.
+func TestPrefixHandlerWritesOriginChanges(t *testing.T) {
+	t.Parallel()
+
+	db := newTestDatabase(t)
+	h := prefixHandlerWithoutTimer(db)
+
+	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
+	h.flushBatch()
+
+	got, err := db.GetOriginChanges(t.Context(),
+		netip.MustParsePrefix(testV4Prefix), 10)
+	if err != nil {
+		t.Fatalf("GetOriginChanges: %v", err)
+	}
+
+	if len(got) != 1 || !got[0].Announced || got[0].OriginASN != testASNB {
+		t.Errorf("GetOriginChanges = %+v, want one announced row for AS%d",
+			got, testASNB)
+	}
+}
+
+// errTestOriginChangeWrite is the error a test store's failing write of the
+// origin changes returns.
+var errTestOriginChangeWrite = errors.New(
+	"test: the write of the origin changes failed")
+
+// originChangesFailsStore is a real database whose first write of the origin
+// changes fails, writing nothing. writes counts the writes asked for.
+type originChangesFailsStore struct {
+	*database.Database
+
+	writes int
+}
+
+func (s *originChangesFailsStore) WriteOriginChanges(
+	changes []routetable.OriginChange,
+) error {
+	s.writes++
+	if s.writes == 1 {
+		return errTestOriginChangeWrite
+	}
+
+	return s.Database.WriteOriginChanges(changes)
+}
+
+// TestPrefixHandlerWritesAFailedOriginChangeWriteAgain runs two flushes into a
+// database whose first write of the origin changes fails. The second write must
+// hold the change of the first.
+func TestPrefixHandlerWritesAFailedOriginChangeWriteAgain(t *testing.T) {
+	t.Parallel()
+
+	db := newTestDatabase(t)
+	store := &originChangesFailsStore{Database: db}
+	h := prefixHandlerWithoutTimer(store)
+
+	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
+	h.flushBatch()
+	h.flushBatch()
+
+	got, err := db.GetOriginChanges(t.Context(),
+		netip.MustParsePrefix(testV4Prefix), 10)
+	if err != nil {
+		t.Fatalf("GetOriginChanges: %v", err)
+	}
+
+	if len(got) != 1 || !got[0].Announced || got[0].OriginASN != testASNB {
+		t.Errorf("GetOriginChanges = %+v, want one announced row for AS%d",
+			got, testASNB)
+	}
+}
+
+// originChangesWriteSizesStore is a real database that records how many
+// changes each write of the origin changes holds.
+type originChangesWriteSizesStore struct {
+	*database.Database
+
+	counts []int
+}
+
+func (s *originChangesWriteSizesStore) WriteOriginChanges(
+	changes []routetable.OriginChange,
+) error {
+	s.counts = append(s.counts, len(changes))
+
+	return s.Database.WriteOriginChanges(changes)
+}
+
+// TestPrefixHandlerWritesManyOriginChangesOverSeveralFlushes has more origin
+// changes than one write takes, then flushes three times. The first write must
+// hold one write's worth, the second the rest, and the third nothing. The
+// handler's limit is small.
+func TestPrefixHandlerWritesManyOriginChangesOverSeveralFlushes(t *testing.T) {
+	t.Parallel()
+
+	const perWrite = 10
+
+	db := newTestDatabase(t)
+	store := &originChangesWriteSizesStore{Database: db}
+	h := prefixHandlerWithoutTimer(store)
+	h.originChangesPerWrite = perWrite
+
+	announceManyPrefixes(h.table, perWrite*3/2)
+
+	h.flushBatch()
+	h.flushBatch()
+	h.flushBatch()
+
+	want := []int{perWrite, perWrite*3/2 - perWrite}
+	if !slices.Equal(store.counts, want) {
+		t.Errorf("WriteOriginChanges held %v changes, want %v", store.counts, want)
+	}
+}
+
+// TestPrefixHandlerLogsTheDroppedOriginChanges fills the routing table with
+// one origin change more than it keeps, then flushes. The warning the flush
+// logs must carry the dropped count.
+func TestPrefixHandlerLogsTheDroppedOriginChanges(t *testing.T) {
+	t.Parallel()
+
+	var logs bytes.Buffer
+
+	db := newTestDatabase(t)
+	h := prefixHandlerWithoutTimer(db)
+	h.logger = &logger.Logger{
+		Logger: slog.New(slog.NewJSONHandler(&logs,
+			&slog.HandlerOptions{Level: slog.LevelWarn})),
+	}
+
+	announceManyPrefixes(h.table, routetable.MaxWaitingOriginChanges+1)
+
+	h.flushBatch()
+
+	var warning struct {
+		Msg     string `json:"msg"`
+		Dropped *int   `json:"dropped"`
+	}
+
+	found := false
+
+	for line := range bytes.SplitSeq(bytes.TrimSpace(logs.Bytes()), []byte("\n")) {
+		warning.Dropped = nil
+
+		err := json.Unmarshal(line, &warning)
+		if err != nil {
+			t.Fatalf("want a JSON log line, got %q: %v", line, err)
+		}
+
+		if warning.Msg != "Origin changes dropped" {
+			continue
+		}
+
+		found = true
+
+		if warning.Dropped == nil || *warning.Dropped != 1 {
+			t.Errorf("the warning's dropped = %v, want 1", warning.Dropped)
+		}
+	}
+
+	if !found {
+		t.Errorf("no warning logged the dropped origin changes")
+	}
+}

internal/routewatch/prefixhandler_table_internal_test.go

@@ -103,23 +103,24 @@ func TestPrefixHandlerAppliesUpdatesToTheRoutingTable(t *testing.T) {
 	})
 }
 
-// TestPrefixHandlerTakesMessagesWhileAPeerGoesDown holds up the PeerDown of a
+// TestPrefixHandlerTakesMessagesWhileAPeerGoesDown holds up the PeersDown of a
-// peer whose session went down, as another peer's long PeerDown would. The
+// peer whose session went down, as another peer's long PeersDown would. The
 // handler must go on taking messages. It applies another peer's announcement
 // at once. It holds back these messages of the down peer, in this order:
 //   - an announcement;
 //   - its session going down again;
 //   - a second announcement.
 //
-// Once the PeerDowns have returned, the down peer must carry only the prefix
+// HeldBackCount must count them. Once the PeersDown calls have returned, the
-// of the second announcement.
+// down peer must carry only the prefix of the second announcement.
 func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
 	t.Parallel()
 
-	h := prefixHandlerWithoutTimer(newMockStore())
+	const heldMessages = 3
-	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
 
-	h.peerDownMu.Lock()
+	h := prefixHandlerWithoutTimer(newMockStore())
+	release := holdUpPeersDown(h)
+	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
 
 	handled := returnsInTime(func() {
 		for _, msg := range []*ristypes.RISMessage{
@@ -144,10 +145,15 @@ func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
 		setV4Prefix:   {},
 	})
 
-	h.peerDownMu.Unlock()
+	if h.HeldBackCount() != heldMessages {
+		t.Errorf("HeldBackCount() = %d, want %d", h.HeldBackCount(),
+			heldMessages)
+	}
+
+	close(release)
 
 	if !returnsInTime(h.peerDowns.Wait) {
-		t.Fatalf("the PeerDowns did not return within %s", stepTimeout)
+		t.Fatalf("the PeersDown calls did not return within %s", stepTimeout)
 	}
 
 	checkTableOrigins(t, h.table, map[string]map[int][]string{
@@ -158,21 +164,85 @@ func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
 	})
 }
 
-// TestPrefixHandlerAppliesMessagesWhilePeerDownRuns gives a peer 50,000
+// TestPrefixHandlerTakesPeersDownTogether has four peers announce testV4Prefix.
+// It holds up the PeersDown of the first peer, whose session goes down.
+// Meanwhile the sessions of the three others go down, and each of them
+// announces otherV4Prefix. Once the first PeersDown is let go, the next one
+// must take the three others together. No peer may then carry testV4Prefix,
+// and the three others must carry otherV4Prefix.
+func TestPrefixHandlerTakesPeersDownTogether(t *testing.T) {
+	t.Parallel()
+
+	others := []string{secondPeerIP, thirdPeerIP, fourthPeerIP}
+	h := prefixHandlerWithoutTimer(newMockStore())
+
+	// calls receives the peers of each PeersDown, with room for one call
+	// for each peer.
+	calls := make(chan []string, len(others)+1)
+	release := make(chan struct{})
+	h.tablePeersDown = func(peers []string) {
+		calls <- slices.Clone(peers)
+
+		<-release
+		h.table.PeersDown(peers)
+	}
+
+	for _, peer := range append([]string{testPeerIP}, others...) {
+		h.HandleMessage(peerAnnouncement(peer, testV4Prefix))
+	}
+
+	h.HandleMessage(sessionDown())
+
+	select {
+	case <-calls:
+	case <-time.After(stepTimeout):
+		t.Fatalf("PeersDown did not begin within %s", stepTimeout)
+	}
+
+	for _, peer := range others {
+		h.HandleMessage(sessionDownOf(peer))
+		h.HandleMessage(peerAnnouncement(peer, otherV4Prefix))
+	}
+
+	close(release)
+
+	if !returnsInTime(h.peerDowns.Wait) {
+		t.Fatalf("the PeersDown calls did not return within %s", stepTimeout)
+	}
+
+	close(calls)
+
+	later := make([][]string, 0, len(calls))
+	for peers := range calls {
+		later = append(later, peers)
+	}
+
+	if !reflect.DeepEqual(later, [][]string{others}) {
+		t.Errorf("the PeersDown calls after the first took %v, want %v "+
+			"together", later, others)
+	}
+
+	checkTableOrigins(t, h.table, map[string]map[int][]string{
+		testV4Prefix:  {},
+		otherV4Prefix: {testASNB: others},
+	})
+}
+
+// TestPrefixHandlerAppliesMessagesWhilePeersDownRuns gives a peer 50,000
 // prefixes and hands the prefix handler that peer's session going down. The
-// table's PeerDown then visits the prefixes in five batches. Once PeerDown has
+// table's PeersDown then visits the prefixes in five batches. Once PeersDown
-// begun, the handler is handed 1,000 announcements of another peer. The
+// has begun, the handler is handed 1,000 announcements of another peer. The
-// handler must apply them all before PeerDown ends, while the down peer still
+// handler must apply them all before PeersDown ends, while the down peer still
 // carries some of its prefixes.
-func TestPrefixHandlerAppliesMessagesWhilePeerDownRuns(t *testing.T) {
+func TestPrefixHandlerAppliesMessagesWhilePeersDownRuns(t *testing.T) {
 	t.Parallel()
 
 	const (
-		// downPeerPrefixes is five times the 10,000 prefixes PeerDown visits
+		// downPeerPrefixes is five times the 10,000 prefixes PeersDown visits
 		// in one hold of the table's lock.
 		downPeerPrefixes = 50000
 		// otherPeerMessages is far more messages than the handler applies
-		// while PeerDown runs, if PeerDown takes the lock back at once after
+		// while PeersDown runs, if PeersDown takes the lock back at once after
 		// each batch. The handler then applies only a few for each batch.
 		otherPeerMessages = 1000
 		// firstOtherAddr is the address of the other peer's first prefix.
@@ -184,7 +254,7 @@ func TestPrefixHandlerAppliesMessagesWhilePeerDownRuns(t *testing.T) {
 
 	h.HandleMessage(sessionDown())
 
-	// Wait until PeerDown has begun: the down peer has left some of its
+	// Wait until PeersDown has begun: the down peer has left some of its
 	// prefixes. The down peer's prefixes are IPv4 and the other peer's are
 	// IPv6, so the counts tell them apart.
 	for h.table.Counts().IPv4Routes == downPeerPrefixes {
@@ -207,15 +277,76 @@ func TestPrefixHandlerAppliesMessagesWhilePeerDownRuns(t *testing.T) {
 	}
 
 	if counts.IPv4Routes == 0 {
-		t.Errorf("PeerDown ended before the handler applied the other peer's "+
+		t.Errorf("PeersDown ended before the handler applied the other peer's "+
 			"%d announcements", otherPeerMessages)
 	}
 
 	if !returnsInTime(h.peerDowns.Wait) {
-		t.Fatalf("the PeerDown did not return within %s", stepTimeout)
+		t.Fatalf("the PeersDown did not return within %s", stepTimeout)
 	}
 }
 
+// TestPrefixHandlerTakesMessagesWhileAReturningPeerSendsItsTable gives a peer
+// 50,000 prefixes and hands the prefix handler that peer's session going down.
+// Once PeersDown has begun, the peer comes back and sends more announcements
+// than the handler's queue holds, as a peer that sends its whole table again
+// does. The handler must hold them back and go on taking messages. Another
+// peer's announcement, handed next, must be applied before PeersDown ends.
+func TestPrefixHandlerTakesMessagesWhileAReturningPeerSendsItsTable(t *testing.T) {
+	t.Parallel()
+
+	const (
+		// downPeerPrefixes is five times the 10,000 prefixes PeersDown visits
+		// in one hold of the table's lock.
+		downPeerPrefixes  = 50000
+		returningMessages = prefixHandlerQueueSize + 1
+	)
+
+	table := routetable.New(routetable.Params{})
+	announceManyPrefixes(table, downPeerPrefixes)
+
+	h := newPrefixHandler(newMockStore(), logger.New(), table,
+		longerThanAnyTest, maxTableChangesPerWrite)
+	defer stopInTime(t, h)
+
+	h.HandleMessage(sessionDown())
+
+	// Wait until PeersDown has begun: the down peer has left some of its
+	// prefixes, which are all IPv4.
+	for h.table.Counts().IPv4Routes == downPeerPrefixes {
+		runtime.Gosched()
+	}
+
+	returning := peerAnnouncement(testPeerIP, testV6Prefix)
+
+	if !returnsInTime(func() {
+		for range returningMessages {
+			h.HandleMessage(returning)
+		}
+
+		h.HandleMessage(peerAnnouncement(secondPeerIP, testV6Prefix))
+	}) {
+		t.Fatalf("the handler did not take the messages within %s", stepTimeout)
+	}
+
+	if h.table.Counts().IPv4Routes == 0 {
+		t.Error("PeersDown ended before the handler took the other peer's " +
+			"announcement")
+	}
+
+	checkTableOrigins(t, h.table, map[string]map[int][]string{
+		testV6Prefix: {testASNB: {secondPeerIP}},
+	})
+
+	if !returnsInTime(h.peerDowns.Wait) {
+		t.Fatalf("the PeersDown did not return within %s", stepTimeout)
+	}
+
+	checkTableOrigins(t, h.table, map[string]map[int][]string{
+		testV6Prefix: {testASNB: {testPeerIP, secondPeerIP}},
+	})
+}
+
 // TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes holds back many
 // messages of a peer whose session went down. Each names a prefix that does
 // not parse, so applying it logs an error. Each log write takes a
@@ -237,14 +368,14 @@ func TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes(t *testing.T) {
 
 	defer stopInTime(t, h)
 
-	h.peerDownMu.Lock()
+	release := holdUpPeersDown(h)
 	h.HandleMessage(sessionDown())
 
 	for range heldMessages {
 		h.HandleMessage(peerAnnouncement(testPeerIP, badPrefix))
 	}
 
-	h.peerDownMu.Unlock()
+	close(release)
 
 	select {
 	case <-logs.firstWrite:
@@ -268,10 +399,10 @@ func TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes(t *testing.T) {
 	}
 }
 
-// TestPrefixHandlerWaitsAtTheMaximumHeldBack holds up the PeerDown of a peer
+// TestPrefixHandlerWaitsAtTheMaximumHeldBack holds up the PeersDown of a peer
 // whose session went down. It hands the handler maxHeldBackMessages messages
 // of that peer. HandleMessage must then wait, even for another peer's
-// message, until the PeerDown returns and a held-back message is applied.
+// message, until the PeersDown returns and a held-back message is applied.
 func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) {
 	t.Parallel()
 
@@ -280,7 +411,7 @@ func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) {
 		maxTableChangesPerWrite)
 	defer stopInTime(t, h)
 
-	h.peerDownMu.Lock()
+	release := holdUpPeersDown(h)
 	h.HandleMessage(sessionDown())
 
 	held := peerAnnouncement(testPeerIP, testV6Prefix)
@@ -302,12 +433,12 @@ func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) {
 	case <-time.After(stillWaitingAfter):
 	}
 
-	h.peerDownMu.Unlock()
+	close(release)
 
 	select {
 	case <-returned:
 	case <-time.After(stepTimeout):
-		t.Fatal("HandleMessage did not return after the PeerDown returned")
+		t.Fatal("HandleMessage did not return after the PeersDown returned")
 	}
 }
 
@@ -424,7 +555,7 @@ func TestPrefixHandlerKeepsThePathOfAPathEndingInASet(t *testing.T) {
 }
 
 // checkTableCases hands the messages of each case to a new prefix handler. It
-// checks the handler's routing table once the handler's PeerDowns have
+// checks the handler's routing table once the handler's PeersDown calls have
 // returned, with no flush.
 func checkTableCases(t *testing.T, tests []tableCase) {
 	t.Helper()

internal/server/handlers.go

@@ -88,6 +88,10 @@ const (
 	ipVersionV6 = 6
 )
 
+// maxOriginChangesOnPage is the most origin changes the prefix page shows. A
+// prefix with more shows the newest ones and says so.
+const maxOriginChangesOnPage = 5000
+
 // HealthCheckResponse represents the health check response.
 type HealthCheckResponse struct {
 	Status    string            `json:"status"`
@@ -486,6 +490,7 @@ type HandlerStatsInfo struct {
 	QueueHighWaterMark int     `json:"queue_high_water_mark"`
 	ProcessedCount     uint64  `json:"processed_count"`
 	DroppedCount       uint64  `json:"dropped_count"`
+	HeldBackCount      int     `json:"held_back_count"`
 	AvgProcessTimeMs   float64 `json:"avg_process_time_ms"`
 	MinProcessTimeMs   float64 `json:"min_process_time_ms"`
 	MaxProcessTimeMs   float64 `json:"max_process_time_ms"`
@@ -570,6 +575,7 @@ func handlerStatsInfo(handlerStats []streamer.HandlerStats) []HandlerStatsInfo {
 			QueueHighWaterMark: hs.QueueHighWaterMark,
 			ProcessedCount:     hs.ProcessedCount,
 			DroppedCount:       hs.DroppedCount,
+			HeldBackCount:      hs.HeldBackCount,
 			AvgProcessTimeMs: float64(hs.AvgProcessTime.Microseconds()) /
 				microsecondsPerMillisecond,
 			MinProcessTimeMs: float64(hs.MinProcessTime.Microseconds()) /
@@ -1532,27 +1538,24 @@ func (s *Server) handlePrefixDetail() http.HandlerFunc {
 
 		routes := prefixRoutes(parsed, origins)
 
-		// Prepare template data
+		changes, moreChanges, err := s.prefixOriginChanges(r.Context(), parsed)
-		data := struct {
+		if deadlinePassed(r.Context()) {
-			Prefix      string
+			http.Error(w, requestTimeoutMessage, http.StatusGatewayTimeout)
-			MaskLength  int
+
-			IPVersion   int
+			return
-			Routes      []EnhancedRoute
-			Origins     []*ASNInfo
-			PeerCount   int
-			OriginCount int
-			CSPNonce    string
-		}{
-			Prefix:      parsed.String(),
-			MaskLength:  parsed.Bits(),
-			IPVersion:   ipVersionOf(parsed),
-			Routes:      routesWithASPathHandles(routes),
-			Origins:     originInfos,
-			PeerCount:   len(routes),
-			OriginCount: len(originInfos),
-			CSPNonce:    cspNonce(r.Context()),
 		}
 
+		if err != nil {
+			s.logger.Error("Failed to get the origin changes", "error", err)
+			http.Error(w, internalErrorMessage, http.StatusInternalServerError)
+
+			return
+		}
+
+		// Prepare template data
+		data := newPrefixDetailData(parsed, routes, originInfos, changes,
+			moreChanges, cspNonce(r.Context()))
+
 		// Check if context is still valid before writing response
 		select {
 		case <-r.Context().Done():
@@ -1604,6 +1607,61 @@ func (s *Server) originASNs(
 	return infos, nil
 }
 
+// prefixOriginChanges returns prefix's origin changes for the page, newest
+// first, at most maxOriginChangesOnPage, and whether there are more.
+func (s *Server) prefixOriginChanges(
+	ctx context.Context, prefix netip.Prefix,
+) ([]routetable.OriginChange, bool, error) {
+	// One more is read to know whether there are more.
+	changes, err := s.db.GetOriginChanges(ctx, prefix, maxOriginChangesOnPage+1)
+	if err != nil {
+		return nil, false, err
+	}
+
+	more := len(changes) > maxOriginChangesOnPage
+	if more {
+		changes = changes[:maxOriginChangesOnPage]
+	}
+
+	return changes, more, nil
+}
+
+// prefixDetailData is what the prefix detail page is rendered with.
+type prefixDetailData struct {
+	Prefix            string
+	MaskLength        int
+	IPVersion         int
+	Routes            []EnhancedRoute
+	Origins           []*ASNInfo
+	PeerCount         int
+	OriginCount       int
+	OriginChanges     []routetable.OriginChange
+	MoreOriginChanges bool
+	// MaxOriginChanges is the most origin changes the page shows.
+	MaxOriginChanges int
+	CSPNonce         string
+}
+
+// newPrefixDetailData builds the prefix detail page's data.
+func newPrefixDetailData(
+	parsed netip.Prefix, routes []LiveRoute, originInfos []*ASNInfo,
+	changes []routetable.OriginChange, moreChanges bool, nonce string,
+) prefixDetailData {
+	return prefixDetailData{
+		Prefix:            parsed.String(),
+		MaskLength:        parsed.Bits(),
+		IPVersion:         ipVersionOf(parsed),
+		Routes:            routesWithASPathHandles(routes),
+		Origins:           originInfos,
+		PeerCount:         len(routes),
+		OriginCount:       len(originInfos),
+		OriginChanges:     changes,
+		MoreOriginChanges: moreChanges,
+		MaxOriginChanges:  maxOriginChangesOnPage,
+		CSPNonce:          nonce,
+	}
+}
+
 // prefixRoutes returns one route for each peer that carries prefix, by origin
 // AS in the order of origins. Each origin's kept routes come first, newest
 // first, with their AS path, next hop and time. Its other peers follow without

internal/server/handlers_internal_test.go

@@ -295,6 +295,43 @@ func TestStatsAnswerCarriesChurnPerMinute(t *testing.T) {
 	}
 }
 
+// heldBackMessages is how many messages heldBackHandler holds back.
+const heldBackMessages = 7
+
+// heldBackHandler is a stream handler that wants no message and holds back
+// heldBackMessages messages.
+type heldBackHandler struct{}
+
+func (heldBackHandler) WantsMessage(string) bool           { return false }
+func (heldBackHandler) HandleMessage(*ristypes.RISMessage) {}
+func (heldBackHandler) QueueCapacity() int                 { return 1 }
+func (heldBackHandler) HeldBackCount() int                 { return heldBackMessages }
+
+// TestStatsAnswerGivesHowManyMessagesAHandlerHoldsBack registers a handler
+// that holds back messages. Its figures in /api/v1/stats, which the status
+// page shows, must give how many.
+func TestStatsAnswerGivesHowManyMessagesAHandlerHoldsBack(t *testing.T) {
+	t.Parallel()
+
+	s := newServerWithTestRoutes(t)
+	s.streamer.RegisterHandler(heldBackHandler{})
+
+	var body struct {
+		Data StatsResponse `json:"data"`
+	}
+
+	err := json.Unmarshal(get(t, s, statsPath).Body.Bytes(), &body)
+	if err != nil {
+		t.Fatalf("decoding the answer: %v", err)
+	}
+
+	stats := body.Data.HandlerStats
+	if len(stats) != 1 || stats[0].HeldBackCount != heldBackMessages {
+		t.Errorf("handler figures %+v, want one handler holding back %d "+
+			"messages", stats, heldBackMessages)
+	}
+}
+
 // fixedIntervalASNFetcher is a WHOIS fetcher that does nothing and reports
 // interval as its wait between fetches.
 type fixedIntervalASNFetcher struct {

internal/server/middleware_internal_test.go

@@ -261,7 +261,7 @@ func checkAnswerCarriesMeta(
 ) {
 	t.Helper()
 
-	// The values README "API v1" gives the two fields that never change.
+	// The values README "JSON API" gives the two fields that never change.
 	const (
 		wantTimeZone   = "UTC"
 		wantAPIVersion = 1

internal/server/prefix_originchanges_internal_test.go

@@ -0,0 +1,230 @@
+package server
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"net/netip"
+	"strings"
+	"testing"
+
+	"git.eeqj.de/sneak/routewatch/internal/config"
+	"git.eeqj.de/sneak/routewatch/internal/database"
+	"git.eeqj.de/sneak/routewatch/internal/logger"
+	"git.eeqj.de/sneak/routewatch/internal/metrics"
+	"git.eeqj.de/sneak/routewatch/internal/ristypes"
+	"git.eeqj.de/sneak/routewatch/internal/routetable"
+	"git.eeqj.de/sneak/routewatch/internal/streamer"
+)
+
+// testUncarriedASN is an AS the test's routing table does not carry for the prefix.
+const testUncarriedASN = 64501
+
+// newPrefixDetailServer returns a server whose routing table holds testPrefixV4
+// with origin AS 64500, and whose database holds the given origin changes.
+func newPrefixDetailServer(
+	t *testing.T, changes []routetable.OriginChange,
+) *Server {
+	t.Helper()
+
+	db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New())
+	if err != nil {
+		t.Fatalf("database.New: %v", err)
+	}
+
+	t.Cleanup(func() { _ = db.Close() })
+
+	table := emptyTable()
+	announce(table, testPrefixV4, 64500, testPeer(1), testRouteTime(0))
+
+	err = db.WriteOriginChanges(changes)
+	if err != nil {
+		t.Fatalf("WriteOriginChanges: %v", err)
+	}
+
+	return New(db, table, streamer.New(logger.New(), metrics.New()),
+		logger.New(), &config.Config{})
+}
+
+// TestPrefixDetailShowsOriginChanges checks that the prefix page shows the
+// origin changes newest first, with the announced and withdrawn rows and the
+// AS link. The rows use an AS the routing table does not carry for the
+// prefix, so the link comes from the origin changes table.
+func TestPrefixDetailShowsOriginChanges(t *testing.T) {
+	t.Parallel()
+
+	prefix := netip.MustParsePrefix(testPrefixV4)
+
+	s := newPrefixDetailServer(t, []routetable.OriginChange{
+		{Prefix: prefix, OriginASN: testUncarriedASN, Announced: true,
+			PeerCount: 1, Time: testRouteTime(1)},
+		{Prefix: prefix, OriginASN: testUncarriedASN, Announced: false,
+			PeerCount: 0, Time: testRouteTime(2)},
+	})
+
+	rec := httptest.NewRecorder()
+	s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
+		t.Context(), http.MethodGet, "/prefix/"+testPrefixV4, nil))
+
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d, want %d; body %s", rec.Code, http.StatusOK, rec.Body)
+	}
+
+	page := rec.Body.String()
+
+	withdrawn := strings.Index(page, "Withdrawn")
+	announced := strings.Index(page, "Announced")
+
+	if withdrawn < 0 || announced < 0 {
+		t.Fatalf("page does not show the announced and withdrawn rows")
+	}
+
+	// The withdrawn row is newer, so it comes first.
+	if withdrawn > announced {
+		t.Errorf("the withdrawn row is not shown before the announced row")
+	}
+
+	if !strings.Contains(page, `href="/as/64501"`) {
+		t.Errorf("page does not link to AS64501")
+	}
+}
+
+// TestPrefixDetailShowsOriginZeroAsNoKnownOrigin checks that the origin
+// changes table shows AS 0 as no known origin, without a link to AS0.
+func TestPrefixDetailShowsOriginZeroAsNoKnownOrigin(t *testing.T) {
+	t.Parallel()
+
+	prefix := netip.MustParsePrefix(testPrefixV4)
+
+	s := newPrefixDetailServer(t, []routetable.OriginChange{
+		{Prefix: prefix, OriginASN: ristypes.NoOrigin, Announced: true,
+			PeerCount: 1, Time: testRouteTime(1)},
+	})
+
+	rec := httptest.NewRecorder()
+	s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
+		t.Context(), http.MethodGet, "/prefix/"+testPrefixV4, nil))
+
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d, want %d; body %s", rec.Code, http.StatusOK, rec.Body)
+	}
+
+	page := rec.Body.String()
+
+	if !strings.Contains(page, "No known origin") {
+		t.Errorf("page does not say the origin change has no known origin")
+	}
+
+	if strings.Contains(page, `href="/as/0"`) {
+		t.Errorf("page links to AS0")
+	}
+}
+
+// TestPrefixDetailSaysWhenThereAreMoreChangesThanItShows checks that the
+// prefix page says so when a prefix has more origin changes than the page's
+// limit.
+func TestPrefixDetailSaysWhenThereAreMoreChangesThanItShows(t *testing.T) {
+	t.Parallel()
+
+	prefix := netip.MustParsePrefix(testPrefixV4)
+
+	changes := make([]routetable.OriginChange, 0, maxOriginChangesOnPage+1)
+	for i := range maxOriginChangesOnPage + 1 {
+		changes = append(changes, routetable.OriginChange{
+			Prefix: prefix, OriginASN: 64500, Announced: true, PeerCount: 1,
+			Time: testRouteTime(i),
+		})
+	}
+
+	s := newPrefixDetailServer(t, changes)
+
+	rec := httptest.NewRecorder()
+	s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
+		t.Context(), http.MethodGet, "/prefix/"+testPrefixV4, nil))
+
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d, want %d; body %s", rec.Code, http.StatusOK, rec.Body)
+	}
+
+	page := rec.Body.String()
+
+	note := fmt.Sprintf("more than %d changes", maxOriginChangesOnPage)
+	if !strings.Contains(page, note) {
+		t.Errorf("page does not say there are more than %d changes",
+			maxOriginChangesOnPage)
+	}
+
+	if got := strings.Count(page, "<td>Announced</td>"); got != maxOriginChangesOnPage {
+		t.Errorf("page shows %d change rows, want %d", got,
+			maxOriginChangesOnPage)
+	}
+}
+
+// TestPrefixDetailShowsNoNoteAtTheLimit checks that a prefix with exactly the
+// page's limit of origin changes shows all of them and no note.
+func TestPrefixDetailShowsNoNoteAtTheLimit(t *testing.T) {
+	t.Parallel()
+
+	prefix := netip.MustParsePrefix(testPrefixV4)
+
+	changes := make([]routetable.OriginChange, 0, maxOriginChangesOnPage)
+	for i := range maxOriginChangesOnPage {
+		changes = append(changes, routetable.OriginChange{
+			Prefix: prefix, OriginASN: 64500, Announced: true, PeerCount: 1,
+			Time: testRouteTime(i),
+		})
+	}
+
+	s := newPrefixDetailServer(t, changes)
+
+	rec := httptest.NewRecorder()
+	s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
+		t.Context(), http.MethodGet, "/prefix/"+testPrefixV4, nil))
+
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d, want %d; body %s", rec.Code, http.StatusOK, rec.Body)
+	}
+
+	page := rec.Body.String()
+
+	if strings.Contains(page, "more than") {
+		t.Errorf("page says there are more changes, but a prefix has %d",
+			maxOriginChangesOnPage)
+	}
+
+	if got := strings.Count(page, "<td>Announced</td>"); got != maxOriginChangesOnPage {
+		t.Errorf("page shows %d change rows, want %d", got,
+			maxOriginChangesOnPage)
+	}
+}
+
+// TestPrefixDetailShowsOriginChangeTimeAndPeerCount checks that a row shows
+// its time and its peer count.
+func TestPrefixDetailShowsOriginChangeTimeAndPeerCount(t *testing.T) {
+	t.Parallel()
+
+	prefix := netip.MustParsePrefix(testPrefixV4)
+
+	s := newPrefixDetailServer(t, []routetable.OriginChange{
+		{Prefix: prefix, OriginASN: 64500, Announced: true, PeerCount: 7,
+			Time: testRouteTime(5)},
+	})
+
+	rec := httptest.NewRecorder()
+	s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
+		t.Context(), http.MethodGet, "/prefix/"+testPrefixV4, nil))
+
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d, want %d; body %s", rec.Code, http.StatusOK, rec.Body)
+	}
+
+	page := rec.Body.String()
+
+	if !strings.Contains(page, "2026-01-02 03:09:05") {
+		t.Errorf("page does not show the origin change's time")
+	}
+
+	if !strings.Contains(page, "<td>7</td>") {
+		t.Errorf("page does not show the origin change's peer count")
+	}
+}

internal/server/server.go

@@ -7,8 +7,8 @@
 //   - the AS name search
 //
 // Every route it shows comes from the routing table in memory. AS names,
-// WHOIS details and the times a prefix and an AS were first seen come from
+// WHOIS details, the times a prefix and an AS were first seen, and a prefix's
-// the database.
+// origin changes come from the database.
 package server
 
 import (

internal/streamer/streamer.go

@@ -12,7 +12,6 @@ import (
 	"fmt"
 	"io"
 	"math"
-	"math/rand"
 	"net/http"
 	"sync"
 	"sync/atomic"
@@ -58,10 +57,6 @@ const (
 	// Well above an extended BGP message (RFC 8654) in hex plus its prefixes
 	maxLineSize           = bytesPerMB
 	maxConcurrentHandlers = 800 // Maximum number of concurrent message handlers
-
-	// Backpressure constants
-	backpressureThreshold = 0.5 // Start dropping at 50% queue utilization
-	backpressureSlope     = 2.0 // Slope for linear drop probability increase
 )
 
 // messageTypeNotification is the type of a BGP NOTIFICATION message, which ends
@@ -93,9 +88,8 @@ type MessageHandler interface {
 
 	// QueueCapacity returns the desired queue capacity for this handler.
 	// Handlers that process quickly can have larger queues to buffer bursts.
-	// When the queue fills up, messages will be dropped according to the
+	// While the queue is full, the streamer drops the handler's messages,
-	// backpressure algorithm, except STATE messages, which are held until
+	// except STATE messages, which are held until there is room.
-	// there is room.
 	QueueCapacity() int
 }
 
@@ -126,8 +120,8 @@ type handlerInfo struct {
 
 // Streamer manages a connection to the RIPE RIS Live streaming API for receiving
 // real-time BGP UPDATE messages. It handles automatic reconnection with exponential
-// backoff, dispatches messages to registered handlers via per-handler queues, and
+// backoff and dispatches messages to registered handlers via per-handler queues.
-// implements backpressure to prevent queue overflow during high traffic periods.
+// Its read loop never waits for room on a queue (see queueMessage).
 type Streamer struct {
 	logger *logger.Logger
 	client *http.Client
@@ -144,7 +138,6 @@ type Streamer struct {
 	workers      sync.WaitGroup // the handler workers; Stop waits for them
 	metrics      *metrics.Tracker
 	totalDropped atomic.Uint64   // Total dropped messages across all handlers
-	random       *rand.Rand      // Random number generator for backpressure drops
 	bgpPeers     map[string]bool // Track active BGP peers by peer IP
 	bgpPeersMu   sync.RWMutex    // Protects bgpPeers map
 }
@@ -168,8 +161,6 @@ func New(logger *logger.Logger, metrics *metrics.Tracker) *Streamer {
 		},
 		handlers: make([]*handlerInfo, 0),
 		metrics:  metrics,
-		//nolint:gosec // Non-cryptographic randomness is fine for backpressure
-		random:   rand.New(rand.NewSource(time.Now().UnixNano())),
 		bgpPeers: make(map[string]bool),
 	}
 }
@@ -351,6 +342,10 @@ type HandlerStats struct {
 	AvgProcessTime     time.Duration
 	MinProcessTime     time.Duration
 	MaxProcessTime     time.Duration
+	// HeldBackCount is how many messages the handler holds back to apply
+	// later: what its HeldBackCount method returns, or 0 for a handler
+	// without one.
+	HeldBackCount int
 }
 
 // GetHandlerStats returns a snapshot of performance statistics for all registered
@@ -392,6 +387,11 @@ func (s *Streamer) GetHandlerStats() []HandlerStats {
 
 		info.metrics.mu.Unlock()
 
+		holder, holdsBack := info.handler.(interface{ HeldBackCount() int })
+		if holdsBack {
+			hs.HeldBackCount = holder.HeldBackCount()
+		}
+
 		stats = append(stats, hs)
 	}
 
@@ -399,7 +399,7 @@ func (s *Streamer) GetHandlerStats() []HandlerStats {
 }
 
 // GetDroppedMessages returns the total number of messages dropped across all
-// handlers due to queue overflow or backpressure. This counter is monotonically
+// handlers because their queue was full. This counter is monotonically
 // increasing and is safe to call concurrently.
 func (s *Streamer) GetDroppedMessages() uint64 {
 	return s.totalDropped.Load()
@@ -882,13 +882,13 @@ func (s *Streamer) dispatchMessage(
 // msg instead. It never waits for room, so a full queue does not hold up the
 // read loop.
 //
-// A message is dropped while the queue is filling up (shouldDropForBackpressure)
+// A message that finds the queue full is dropped, except a STATE message. A
-// or full, except a STATE message: one whose state is down is the only notice
+// STATE message whose state is down is the only notice that all of a peer's
-// that all of a peer's routes are gone. A STATE message skips the random drop,
+// routes are gone. A STATE message that finds the queue full is held in
-// and one that finds the queue full is held in info.heldState. Held messages
+// info.heldState instead. Held messages go into the queue, oldest first, when a
-// go into the queue, oldest first, before anything else for the handler, as
+// later message for the handler finds room, or when Stop is called. While any
-// soon as there is room, or when Stop is called; while any is still held, the
+// message is held, the handler's other messages are dropped, so none of them
-// handler's other messages are dropped, so none of them overtakes a held down.
+// overtakes a held STATE message.
 //
 // The held list stays small: a STATE message is a few hundred bytes and the
 // feed sends about ten a second, so a queue full for a minute holds about 600.
@@ -907,7 +907,7 @@ func (s *Streamer) queueMessage(
 		return true
 	}
 
-	if len(info.heldState) > 0 || s.shouldDropForBackpressure(info) {
+	if len(info.heldState) > 0 {
 		return false
 	}
 
@@ -923,26 +923,3 @@ func trySend(queue chan *ristypes.RISMessage, msg *ristypes.RISMessage) bool {
 		return false
 	}
 }
-
-// shouldDropForBackpressure determines if a message should be dropped based on
-// queue utilization
-func (s *Streamer) shouldDropForBackpressure(info *handlerInfo) bool {
-	// Calculate queue utilization
-	queueLen := len(info.queue)
-	queueCap := cap(info.queue)
-	utilization := float64(queueLen) / float64(queueCap)
-
-	// No drops below threshold
-	if utilization < backpressureThreshold {
-		return false
-	}
-
-	// Calculate drop probability (0.0 at threshold, 1.0 at 100% full)
-	dropProbability := (utilization - backpressureThreshold) * backpressureSlope
-	if dropProbability > 1.0 {
-		dropProbability = 1.0
-	}
-
-	// Random drop based on probability
-	return s.random.Float64() < dropProbability
-}

internal/streamer/streamer_internal_test.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"io"
 	"log/slog"
 	"net/http"
@@ -258,6 +259,65 @@ func TestHeldStateMessagesKeepTheirPlaceInTheQueue(t *testing.T) {
 	}
 }
 
+// burstQueueCapacity is the size of a burstHandler's queue.
+const burstQueueCapacity = 1000
+
+// burstHandler wants UPDATE messages and has a queue of burstQueueCapacity.
+// Tests read what it is handed from its queue, so HandleMessage does nothing.
+type burstHandler struct{}
+
+func (burstHandler) WantsMessage(messageType string) bool {
+	return messageType == ristypes.MessageTypeUpdate
+}
+
+func (burstHandler) QueueCapacity() int { return burstQueueCapacity }
+
+func (burstHandler) HandleMessage(*ristypes.RISMessage) {}
+
+// TestUpdatesDroppedOnlyWhenTheQueueIsFull has the server send one UPDATE more
+// than the handler's queue holds, each with its own timestamp. Nothing takes
+// messages off the queue meanwhile. The streamer must queue the updates in
+// feed order until the queue is full, and drop only the last one.
+func TestUpdatesDroppedOnlyWhenTheQueueIsFull(t *testing.T) {
+	t.Parallel()
+
+	const (
+		firstTimestamp = 1700000000
+		lineFormat     = `{"type":"ris_message","data":{"type":"UPDATE",` +
+			`"peer":"192.0.2.1","peer_asn":"64496","timestamp":%d}}` + "\n"
+	)
+
+	var lines strings.Builder
+
+	for i := range burstQueueCapacity + 1 {
+		fmt.Fprintf(&lines, lineFormat, firstTimestamp+i)
+	}
+
+	s := New(logger.New(), metrics.New())
+	s.RegisterHandler(burstHandler{})
+
+	streamLines(t, s, lines.String())
+
+	dropped := s.GetHandlerStats()[0].DroppedCount
+	if dropped != 1 {
+		t.Errorf("%d updates dropped, want only the one that found the queue "+
+			"full", dropped)
+	}
+
+	queue := s.handlers[0].queue
+	if len(queue) != burstQueueCapacity {
+		t.Fatalf("%d updates queued, want %d", len(queue), burstQueueCapacity)
+	}
+
+	for i := range burstQueueCapacity {
+		msg := <-queue
+		if msg.Timestamp != float64(firstTimestamp+i) {
+			t.Fatalf("update %d on the queue has timestamp %.0f, want %d",
+				i, msg.Timestamp, firstTimestamp+i)
+		}
+	}
+}
+
 // takeAll takes the messages off queue, as a handler's worker would, and sends
 // them on the returned channel once queue is closed.
 func takeAll(queue chan *ristypes.RISMessage) chan []*ristypes.RISMessage {

internal/templates/prefix_detail.html

@@ -299,6 +299,41 @@
         </div>
         {{end}}
 
+        {{if .OriginChanges}}
+        <div class="origins-section">
+            <h3>Origin Changes</h3>
+            {{if .MoreOriginChanges}}
+            <p>There are more than {{.MaxOriginChanges}} changes. Showing the newest {{.MaxOriginChanges}}.</p>
+            {{end}}
+            <table class="route-table">
+                <thead>
+                    <tr>
+                        <th>Time (UTC)</th>
+                        <th>Change</th>
+                        <th>Origin AS</th>
+                        <th>Peers</th>
+                    </tr>
+                </thead>
+                <tbody>
+                    {{range .OriginChanges}}
+                    <tr>
+                        <td>{{.Time.Format "2006-01-02 15:04:05"}}</td>
+                        <td>{{if .Announced}}Announced{{else}}Withdrawn{{end}}</td>
+                        <td>
+                            {{if .OriginASN}}
+                            <a href="/as/{{.OriginASN}}" class="as-link">AS{{.OriginASN}}</a>
+                            {{else}}
+                            No known origin
+                            {{end}}
+                        </td>
+                        <td>{{.PeerCount}}</td>
+                    </tr>
+                    {{end}}
+                </tbody>
+            </table>
+        </div>
+        {{end}}
+
         {{if .Routes}}
         <div class="routes-section">
             <div class="routes-header">

internal/templates/status.html

@@ -541,6 +541,10 @@
                         <span class="metric-label">Dropped</span>
                         <span class="metric-value ${handler.dropped_count > 0 ? 'disconnected' : ''}">${formatNumber(handler.dropped_count)}</span>
                     </div>
+                    <div class="metric">
+                        <span class="metric-label">Held Back</span>
+                        <span class="metric-value">${formatNumber(handler.held_back_count)}</span>
+                    </div>
                     <div class="metric">
                         <span class="metric-label">Avg Time</span>
                         <span class="metric-value">${formatProcessingTime(handler.avg_process_time_ms)}</span>