Fix the origin-changes review findings

Hold the 100,000-row cap after a failed write's rows are put back, dropping and counting the oldest rows past it. Stop counts the unwritten origin changes. The prefix handler's per-write origin change limit is a field, so a test checks that one write leaves the rest. The server and README now say where origin changes come from. README, TODO and schema say 72 hours or 500 rows, whichever keeps more. The prefix page and TODO list the origin change columns and say what the peer count counts. The page takes its 5,000-row note from the constant.

Model: deepseek-v4-pro

internal/database/schema.sql

@@ -123,8 +123,9 @@ CREATE TABLE IF NOT EXISTS route_peers (
 
 -- 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 and 72
--- hours. time is whole seconds since 1970, UTC.
+-- 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,

internal/routetable/originchanges.go

@@ -7,7 +7,8 @@ import (
 )
 
 // maxWaitingOriginChanges is the most origin changes the table keeps waiting
-// for the prefix handler to write. A change past it is dropped and counted.
+// 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
@@ -39,13 +40,8 @@ func (t *Table) recordOriginChange(
 		Time:      t.now(),
 	}
 
-	if len(t.originChanges) == maxWaitingOriginChanges {
-		// The oldest change is dropped to make room.
-		t.originChanges = t.originChanges[1:]
-		t.originChangesDropped++
-	}
-
 	t.originChanges = append(t.originChanges, change)
+	t.trimOriginChanges()
 }
 
 // TakeOriginChanges returns at most limit of the origin changes that wait to
@@ -67,12 +63,31 @@ func (t *Table) TakeOriginChanges(limit int) ([]OriginChange, int) {
 
 // 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.
+// 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

internal/routetable/originchanges_test.go

@@ -206,6 +206,60 @@ func TestOriginChangesDropPastTheLimit(t *testing.T) {
 	}
 }
 
+// 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 (
+		limit = 100000
+		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(limit)
+
+	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) != limit {
+		t.Errorf("len(changes) = %d, want %d", len(changes), limit)
+	}
+
+	// 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(

internal/routewatch/prefixhandler.go

@@ -99,6 +99,9 @@ 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
 	// flush goroutine takes them under mu and writes them after releasing
@@ -194,17 +197,18 @@ func newPrefixHandler(
 	batchTimeout time.Duration, tableChangesPerWrite int,
 ) *PrefixHandler {
 	h := &PrefixHandler{
-		db:                   db,
-		logger:               logger,
-		table:                table,
-		batch:                make([]seenPrefix, 0, prefixBatchSize),
-		batchMaxSize:         prefixBatchMaxSize,
-		tableChangesPerWrite: tableChangesPerWrite,
-		batchFullCh:          make(chan struct{}, 1),
-		stopCh:               make(chan context.Context, 1),
-		done:                 make(chan struct{}),
-		heldBack:             make(map[string][]*ristypes.RISMessage),
-		tablePeersDown:       table.PeersDown,
+		db:                    db,
+		logger:                logger,
+		table:                 table,
+		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
@@ -272,13 +276,13 @@ func (h *PrefixHandler) HeldBackCount() int {
 // 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
-// updates the handler holds. Once ctx has ended the flush goroutine skips that
-// last flush, and a write still running is lost when the process exits. A
-// failed write of the table's changes is not retried either. Stop logs how
-// many updates and table changes are not written. It also logs how many
-// held-back messages are not applied. The caller calls Stop once its last
-// HandleMessage has returned.
+// tableChangesPerWrite of the routing table's changes, originChangesPerWrite of
+// its origin changes, and the prefixes of the updates the handler holds. Once
+// ctx has ended the flush goroutine skips that last flush, and a write still
+// running is lost when the process exits. A failed write of the table's
+// changes is not retried either. Stop logs how many updates, table changes and
+// origin changes are not written. It also logs how many held-back messages are
+// not applied. The caller calls Stop once its last HandleMessage has returned.
 func (h *PrefixHandler) Stop(ctx context.Context) {
 	peerDownsDone := make(chan struct{})
 
@@ -305,21 +309,24 @@ func (h *PrefixHandler) Stop(ctx context.Context) {
 	heldMessages := h.heldCount
 	h.mu.Unlock()
 
-	if unwritten == 0 && tableChanges == 0 && heldMessages == 0 {
+	originChanges := h.table.OriginChangeCount()
+
+	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
@@ -618,7 +625,7 @@ func (h *PrefixHandler) writeTableChanges() {
 // that a later write writes them again. A dropped change is logged as a
 // warning.
 func (h *PrefixHandler) writeOriginChanges() {
-	changes, dropped := h.table.TakeOriginChanges(maxOriginChangesPerWrite)
+	changes, dropped := h.table.TakeOriginChanges(h.originChangesPerWrite)
 
 	if dropped > 0 {
 		h.logger.Warn("Origin changes dropped", "dropped", dropped)

internal/routewatch/prefixhandler_internal_test.go

@@ -327,13 +327,14 @@ func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler {
 	table := routetable.New(routetable.Params{})
 
 	return &PrefixHandler{
-		db:                   store,
-		logger:               logger.New(),
-		table:                table,
-		batchMaxSize:         prefixBatchMaxSize,
-		tableChangesPerWrite: maxTableChangesPerWrite,
-		heldBack:             make(map[string][]*ristypes.RISMessage),
-		tablePeersDown:       table.PeersDown,
+		db:                    store,
+		logger:                logger.New(),
+		table:                 table,
+		batchMaxSize:          prefixBatchMaxSize,
+		tableChangesPerWrite:  maxTableChangesPerWrite,
+		originChangesPerWrite: maxOriginChangesPerWrite,
+		heldBack:              make(map[string][]*ristypes.RISMessage),
+		tablePeersDown:        table.PeersDown,
 	}
 }
 

internal/routewatch/prefixhandler_originchanges_internal_test.go

@@ -3,6 +3,7 @@ package routewatch
 import (
 	"errors"
 	"net/netip"
+	"slices"
 	"testing"
 
 	"git.eeqj.de/sneak/routewatch/internal/database"
@@ -82,3 +83,45 @@ func TestPrefixHandlerWritesAFailedOriginChangeWriteAgain(t *testing.T) {
 			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)
+	}
+}

internal/server/handlers.go

@@ -1637,7 +1637,9 @@ type prefixDetailData struct {
 	OriginCount       int
 	OriginChanges     []routetable.OriginChange
 	MoreOriginChanges bool
-	CSPNonce          string
+	// MaxOriginChanges is the most origin changes the page shows.
+	MaxOriginChanges int
+	CSPNonce         string
 }
 
 // newPrefixDetailData builds the prefix detail page's data.
@@ -1655,6 +1657,7 @@ func newPrefixDetailData(
 		OriginCount:       len(originInfos),
 		OriginChanges:     changes,
 		MoreOriginChanges: moreChanges,
+		MaxOriginChanges:  maxOriginChangesOnPage,
 		CSPNonce:          nonce,
 	}
 }

internal/server/prefix_originchanges_internal_test.go

@@ -142,7 +142,7 @@ func TestPrefixDetailSaysWhenMoreThan5000Changes(t *testing.T) {
 
 	page := rec.Body.String()
 
-	if !strings.Contains(page, "more than 5,000 changes") {
-		t.Errorf("page does not say there are more than 5,000 changes")
+	if !strings.Contains(page, "more than 5000 changes") {
+		t.Errorf("page does not say there are more than 5000 changes")
 	}
 }

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
-// the database.
+// WHOIS details, the times a prefix and an AS were first seen, and a prefix's
+// origin changes come from the database.
 package server
 
 import (

internal/templates/prefix_detail.html

@@ -297,13 +297,13 @@
                 {{end}}
             </div>
         </div>
-{{end}}
+        {{end}}
 
         {{if .OriginChanges}}
         <div class="origins-section">
             <h3>Origin Changes</h3>
             {{if .MoreOriginChanges}}
-            <p>There are more than 5,000 changes. Showing the newest 5,000.</p>
+            <p>There are more than {{.MaxOriginChanges}} changes. Showing the newest {{.MaxOriginChanges}}.</p>
             {{end}}
             <table class="route-table">
                 <thead>