From 05b39eee7aa85b3ebcf3201d402d3f394b0d4898 Mon Sep 17 00:00:00 2001
From: clawbot <35+clawbot@noreply.example.org>
Date: Tue, 27 Oct 2026 14:56:11 +0100
Subject: [PATCH] Start no AS, peer or peering write after the stop deadline
 (closes #189)

A write can run past the stop deadline while a full batch's wake-up or a
tick waits beside Stop's context. The select picked one at random, so about
half the time the AS, peer or peering handler started another write after
the deadline. Each loop now takes a waiting stop first, as the prefix
handler's does. The prefix, AS and peer handlers' flush loops were
near-copies, so they are now one function, flushLoop. A new test for each
handler fails about half its runs on the old code.

The AS and peer handler tests give the flush timer an interval longer than
the test. A timer flush during the fill can no longer hang
TestPeerHandlerWaitsAtTheMaximumBatchSize.

Model: opus-5-5
---
 TODO.md                                       |   5 +
 internal/routewatch/app.go                    |   4 +-
 internal/routewatch/ashandler.go              |  48 +++-----
 internal/routewatch/batchwrite.go             |  64 ++++++++++
 .../routewatch/batchwrite_internal_test.go    | 111 ++++++++++++++++--
 internal/routewatch/peerhandler.go            |  48 +++-----
 internal/routewatch/peeringhandler.go         |  31 ++++-
 internal/routewatch/prefixhandler.go          |  53 +--------
 .../routewatch/prefixhandler_internal_test.go |   4 +-
 9 files changed, 232 insertions(+), 136 deletions(-)
 create mode 100644 internal/routewatch/batchwrite.go

diff --git a/TODO.md b/TODO.md
index 7e6de2c..59637e2 100644
--- a/TODO.md
+++ b/TODO.md
@@ -33,6 +33,11 @@ tested.
 
 # Completed Steps
 
+- 2026-10-27: once the stop deadline has passed, the AS, peer and peering
+  handlers start no write either. A full batch or a tick of the timer that
+  waited while a write ran past the deadline could still start one. The flush
+  goroutines of the prefix, AS and peer handlers now run the same code (closes
+  #189)
 - 2026-10-27: once the stop deadline has passed, the prefix handler starts no
   flush. A batch that filled while a write ran past the deadline could still be
   written after it (closes #188)
diff --git a/internal/routewatch/app.go b/internal/routewatch/app.go
index a8ee4bd..f0eafef 100644
--- a/internal/routewatch/app.go
+++ b/internal/routewatch/app.go
@@ -2,7 +2,9 @@
 // running instance of the application and contains pointers to its core
 // dependencies, and is responsible for initialization. It also contains the
 // ASN WHOIS fetcher for background updates and the database maintainer for
-// background maintenance tasks.
+// background maintenance tasks. The package also holds the handlers that
+// write the feed's updates to the database in batches, and the flush loop
+// that the AS, peer and prefix handlers share.
 package routewatch
 
 import (
diff --git a/internal/routewatch/ashandler.go b/internal/routewatch/ashandler.go
index d82b9f8..cee6699 100644
--- a/internal/routewatch/ashandler.go
+++ b/internal/routewatch/ashandler.go
@@ -52,7 +52,6 @@ type ASHandler struct {
 	// &mu.
 	batchTaken  sync.Cond
 	batchFullCh chan struct{} // wakes the flush goroutine when batch is full
-	lastFlush   time.Time     // used only by the flush goroutine
 	// stopCh hands Stop's context to the flush goroutine. It holds one, so
 	// Stop does not wait for a write in progress to hand it over.
 	stopCh chan context.Context
@@ -70,19 +69,27 @@ type asnOp struct {
 // Stop when finished to ensure all pending operations are flushed and the
 // background goroutine is terminated.
 func NewASHandler(db database.Store, logger *logger.Logger) *ASHandler {
+	return newASHandler(db, logger, asnBatchTimeout)
+}
+
+// newASHandler is NewASHandler with the flush timer's interval, batchTimeout,
+// in place of asnBatchTimeout. A test that the timer must not disturb gives an
+// interval longer than the test runs.
+func newASHandler(
+	db database.Store, logger *logger.Logger, batchTimeout time.Duration,
+) *ASHandler {
 	h := &ASHandler{
 		db:          db,
 		logger:      logger,
 		batch:       make([]asnOp, 0, asnBatchSize),
 		batchFullCh: make(chan struct{}, 1),
-		lastFlush:   time.Now(),
 		stopCh:      make(chan context.Context, 1),
 		done:        make(chan struct{}),
 	}
 	h.batchTaken.L = &h.mu
 
 	// Start the flush timer goroutine
-	go h.flushLoop()
+	go flushLoop(batchTimeout, h.batchFullCh, h.stopCh, h.done, h.flushBatch)
 
 	return h
 }
@@ -175,33 +182,6 @@ func (h *ASHandler) Stop(ctx context.Context) {
 	}
 }
 
-// flushLoop runs in a goroutine, the only one that writes the batch: it
-// flushes when the batch is full, periodically, and once more on Stop.
-func (h *ASHandler) flushLoop() {
-	defer close(h.done)
-
-	ticker := time.NewTicker(asnBatchTimeout)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-ticker.C:
-			if time.Since(h.lastFlush) >= asnBatchTimeout {
-				h.flushBatch()
-			}
-		case <-h.batchFullCh:
-			h.flushBatch()
-		case ctx := <-h.stopCh:
-			// Final flush, unless the stop deadline has passed
-			if ctx.Err() == nil {
-				h.flushBatch()
-			}
-
-			return
-		}
-	}
-}
-
 // takeBatch returns the pending operations and leaves an empty batch in their
 // place for HandleMessage to fill. They count in writing until flushBatch has
 // written them.
@@ -222,11 +202,11 @@ func (h *ASHandler) takeBatch() []asnOp {
 }
 
 // flushBatch takes the pending batch and writes it to the database, without
-// holding the mutex during the write.
-func (h *ASHandler) flushBatch() {
+// holding the mutex during the write. It reports whether there was a batch.
+func (h *ASHandler) flushBatch() bool {
 	batch := h.takeBatch()
 	if len(batch) == 0 {
-		return
+		return false
 	}
 
 	// The earliest and the latest time the batch saw each AS. The batch holds
@@ -250,5 +230,5 @@ func (h *ASHandler) flushBatch() {
 	h.writing = 0
 	h.mu.Unlock()
 
-	h.lastFlush = time.Now()
+	return true
 }
diff --git a/internal/routewatch/batchwrite.go b/internal/routewatch/batchwrite.go
new file mode 100644
index 0000000..53ef900
--- /dev/null
+++ b/internal/routewatch/batchwrite.go
@@ -0,0 +1,64 @@
+package routewatch
+
+import (
+	"context"
+	"time"
+)
+
+// flushLoop is the flush goroutine of the prefix, AS and peer handlers. It is
+// the only goroutine that writes a handler's batch. flushBatch takes the batch,
+// writes it, and reports whether there was one.
+//
+// flushLoop calls flushBatch when batchFullCh wakes it. It also calls it every
+// batchTimeout, unless it wrote a batch less than batchTimeout ago. When stopCh
+// hands it Stop's context, it calls flushBatch once more, unless the stop
+// deadline has passed. Then it closes done and returns.
+func flushLoop(
+	batchTimeout time.Duration,
+	batchFullCh <-chan struct{},
+	stopCh <-chan context.Context,
+	done chan<- struct{},
+	flushBatch func() bool,
+) {
+	defer close(done)
+
+	ticker := time.NewTicker(batchTimeout)
+	defer ticker.Stop()
+
+	lastFlush := time.Now()
+
+	for {
+		// Stop goes first. A write can run past the stop deadline while a full
+		// batch's wake-up or a tick waits beside Stop's context. The select
+		// below picks one of them at random, but past the deadline no flush
+		// may start.
+		select {
+		case ctx := <-stopCh:
+			// Final flush, unless the stop deadline has passed
+			if ctx.Err() == nil {
+				flushBatch()
+			}
+
+			return
+		default:
+		}
+
+		select {
+		case <-ticker.C:
+			if time.Since(lastFlush) >= batchTimeout && flushBatch() {
+				lastFlush = time.Now()
+			}
+		case <-batchFullCh:
+			if flushBatch() {
+				lastFlush = time.Now()
+			}
+		case ctx := <-stopCh:
+			// Final flush, unless the stop deadline has passed
+			if ctx.Err() == nil {
+				flushBatch()
+			}
+
+			return
+		}
+	}
+}
diff --git a/internal/routewatch/batchwrite_internal_test.go b/internal/routewatch/batchwrite_internal_test.go
index c14ac79..f659a0f 100644
--- a/internal/routewatch/batchwrite_internal_test.go
+++ b/internal/routewatch/batchwrite_internal_test.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"fmt"
 	"log/slog"
 	"slices"
 	"sync"
@@ -168,9 +169,10 @@ func returnsInTime(f func()) bool {
 	}
 }
 
-// startBlockedWrite hands h the messages in fill, which fill its batch so that
-// the handler writes it to store, and waits until that write starts. store
-// holds it until the test calls releaseWrites.
+// startBlockedWrite hands h the messages in fill and waits until h starts to
+// write them to store. The prefix, AS and peer handlers start when fill fills
+// their batch, and the peering handler on its timer. store holds the write
+// until the test calls releaseWrites.
 func startBlockedWrite(
 	t *testing.T,
 	store *blockingStore,
@@ -193,7 +195,7 @@ func startBlockedWrite(
 	select {
 	case <-store.writeStarted:
 	case <-time.After(handleMessageTimeout):
-		t.Fatal("the handler did not write its full batch")
+		t.Fatal("the handler did not start to write the messages")
 	}
 }
 
@@ -273,7 +275,7 @@ func handleAtMaxBatchSize(
 }
 
 // stopWhileAWriteIsBlocked starts a blocked write with the messages in fill,
-// hands h the message extra, and stops h with a deadline that passes while
+// hands h the messages in extra, and stops h with a deadline that passes while
 // store holds the write. Stop must return all the same. Then it lets the write
 // finish and waits until done, closed when h's flush goroutine returns. Past
 // the deadline that goroutine skips its last write, so extra is not written,
@@ -287,7 +289,7 @@ func stopWhileAWriteIsBlocked(
 	},
 	done <-chan struct{},
 	fill []*ristypes.RISMessage,
-	extra *ristypes.RISMessage,
+	extra ...*ristypes.RISMessage,
 ) {
 	t.Helper()
 
@@ -296,7 +298,10 @@ func stopWhileAWriteIsBlocked(
 	defer store.releaseWrites()
 
 	startBlockedWrite(t, store, h, fill)
-	h.HandleMessage(extra)
+
+	for _, msg := range extra {
+		h.HandleMessage(msg)
+	}
 
 	ctx, cancel := context.WithTimeout(t.Context(), stopDeadline)
 	defer cancel()
@@ -428,7 +433,7 @@ func TestASHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	var logs bytes.Buffer
 
 	store := newBlockingStore()
-	h := NewASHandler(store, newErrorLogger(&logs))
+	h := newASHandler(store, newErrorLogger(&logs), longerThanAnyTest)
 	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
 	stopWhileAWriteIsBlocked(t, store, h, h.done,
 		[]*ristypes.RISMessage{{Path: fill}},
@@ -447,6 +452,29 @@ func TestASHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	}
 }
 
+// TestASHandlerSkipsAFullBatchAfterTheStopDeadline hands the AS handler a full
+// batch while a write is held, so the batch's wake-up of the flush goroutine
+// waits. The stop deadline passes meanwhile. When the write ends, the flush
+// goroutine must stop instead of writing the batch.
+func TestASHandlerSkipsAFullBatchAfterTheStopDeadline(t *testing.T) {
+	t.Parallel()
+
+	store := newBlockingStore()
+	h := newASHandler(store, logger.New(), longerThanAnyTest)
+	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
+	full := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNB}, asnBatchSize)}
+	stopWhileAWriteIsBlocked(t, store, h, h.done,
+		[]*ristypes.RISMessage{{Path: fill}},
+		&ristypes.RISMessage{Path: full})
+
+	store.mu.Lock()
+	defer store.mu.Unlock()
+
+	if store.asns[testASNB] {
+		t.Errorf("AS%d was written after the stop deadline", testASNB)
+	}
+}
+
 func TestPeerHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	t.Parallel()
 
@@ -455,7 +483,7 @@ func TestPeerHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	var logs bytes.Buffer
 
 	store := newBlockingStore()
-	h := NewPeerHandler(store, newErrorLogger(&logs))
+	h := newPeerHandler(store, newErrorLogger(&logs), longerThanAnyTest)
 	stopWhileAWriteIsBlocked(t, store, h, h.done,
 		slices.Repeat([]*ristypes.RISMessage{{Peer: testPeerIP}}, peerBatchSize),
 		&ristypes.RISMessage{Peer: otherPeerIP})
@@ -473,6 +501,30 @@ func TestPeerHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	}
 }
 
+// TestPeerHandlerSkipsAFullBatchAfterTheStopDeadline hands the peer handler a
+// full batch while a write is held, so the batch's wake-up of the flush
+// goroutine waits. The stop deadline passes meanwhile. When the write ends, the
+// flush goroutine must stop instead of writing the batch.
+func TestPeerHandlerSkipsAFullBatchAfterTheStopDeadline(t *testing.T) {
+	t.Parallel()
+
+	const otherPeerIP = "2001:db8:ffff::2"
+
+	store := newBlockingStore()
+	h := newPeerHandler(store, logger.New(), longerThanAnyTest)
+	stopWhileAWriteIsBlocked(t, store, h, h.done,
+		slices.Repeat([]*ristypes.RISMessage{{Peer: testPeerIP}}, peerBatchSize),
+		slices.Repeat([]*ristypes.RISMessage{{Peer: otherPeerIP}},
+			peerBatchSize)...)
+
+	store.mu.Lock()
+	defer store.mu.Unlock()
+
+	if store.peers[otherPeerIP] {
+		t.Errorf("peer %s was written after the stop deadline", otherPeerIP)
+	}
+}
+
 // TestPeeringHandlerStopCountsTheWriteInProgress hands a running peering
 // handler one AS path and stops it with a deadline that passes while store
 // holds Stop's last write, of that path's peering. Stop must log that path as
@@ -512,6 +564,35 @@ func TestPeeringHandlerStopCountsTheWriteInProgress(t *testing.T) {
 	}
 }
 
+// TestPeeringHandlerStartsNoWriteAfterTheStopDeadline has the peering handler's
+// timer start a write that store holds, and hands the handler one more AS path
+// meanwhile. The stop deadline passes, and a tick of the timer waits. When the
+// write ends, the processing goroutine must stop instead of writing the path.
+func TestPeeringHandlerStartsNoWriteAfterTheStopDeadline(t *testing.T) {
+	t.Parallel()
+
+	// The timer fires many times while Stop waits for its deadline.
+	const processInterval = 10 * time.Millisecond
+
+	store := newBlockingStore()
+	h := newPeeringHandler(store, logger.New(), processInterval)
+	stopWhileAWriteIsBlocked(t, store, h, h.done,
+		[]*ristypes.RISMessage{
+			{Path: ristypes.ASPath{Sequence: []int{testASNA, testASNB}}},
+		},
+		&ristypes.RISMessage{
+			Path: ristypes.ASPath{Sequence: []int{testASNC, testASND}},
+		})
+
+	store.mockStore.mu.Lock()
+	defer store.mockStore.mu.Unlock()
+
+	if store.Peerings[fmt.Sprintf("%d_%d", testASNC, testASND)] {
+		t.Errorf("the peering of AS%d and AS%d was written after the stop "+
+			"deadline", testASNC, testASND)
+	}
+}
+
 // prefixAnnouncement returns a message in which testPeerIP announces prefixes
 // with origin AS testASNA.
 func prefixAnnouncement(prefixes []string) *ristypes.RISMessage {
@@ -549,7 +630,7 @@ func TestASHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) {
 	t.Parallel()
 
 	store := newBlockingStore()
-	h := NewASHandler(store, logger.New())
+	h := newASHandler(store, logger.New(), longerThanAnyTest)
 	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
 	handleWhileAWriteIsBlocked(t, store, h,
 		[]*ristypes.RISMessage{{Path: fill}},
@@ -570,7 +651,7 @@ func TestPeerHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) {
 	const otherPeerIP = "2001:db8:ffff::2"
 
 	store := newBlockingStore()
-	h := NewPeerHandler(store, logger.New())
+	h := newPeerHandler(store, logger.New(), longerThanAnyTest)
 	handleWhileAWriteIsBlocked(t, store, h,
 		slices.Repeat([]*ristypes.RISMessage{{Peer: testPeerIP}}, peerBatchSize),
 		&ristypes.RISMessage{Peer: otherPeerIP})
@@ -611,7 +692,7 @@ func TestASHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 	t.Parallel()
 
 	store := newBlockingStore()
-	h := NewASHandler(store, logger.New())
+	h := newASHandler(store, logger.New(), longerThanAnyTest)
 	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
 	toMax := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchMaxSize)}
 	handleAtMaxBatchSize(t, store, h,
@@ -628,13 +709,17 @@ func TestASHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 	}
 }
 
+// TestPeerHandlerWaitsAtTheMaximumBatchSize must not have its flush timer fire.
+// A timer flush during the fill would take only part of it. The batch would
+// then reach its maximum size before the test's last message, and the test
+// would hang.
 func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 	t.Parallel()
 
 	const otherPeerIP = "2001:db8:ffff::2"
 
 	store := newBlockingStore()
-	h := NewPeerHandler(store, logger.New())
+	h := newPeerHandler(store, logger.New(), longerThanAnyTest)
 	message := []*ristypes.RISMessage{{Peer: testPeerIP}}
 	handleAtMaxBatchSize(t, store, h,
 		slices.Repeat(message, peerBatchSize),
diff --git a/internal/routewatch/peerhandler.go b/internal/routewatch/peerhandler.go
index 6fb3278..5800474 100644
--- a/internal/routewatch/peerhandler.go
+++ b/internal/routewatch/peerhandler.go
@@ -56,7 +56,6 @@ type PeerHandler struct {
 	// is &mu.
 	batchTaken  sync.Cond
 	batchFullCh chan struct{} // wakes the flush goroutine when peerBatch is full
-	lastFlush   time.Time     // used only by the flush goroutine
 	// stopCh hands Stop's context to the flush goroutine. It holds one, so
 	// Stop does not wait for a write in progress to hand it over.
 	stopCh chan context.Context
@@ -75,19 +74,27 @@ type peerUpdate struct {
 // periodically flushes accumulated peer updates to the database. The handler
 // should be stopped by calling Stop when it is no longer needed.
 func NewPeerHandler(db database.Store, logger *logger.Logger) *PeerHandler {
+	return newPeerHandler(db, logger, peerBatchTimeout)
+}
+
+// newPeerHandler is NewPeerHandler with the flush timer's interval,
+// batchTimeout, in place of peerBatchTimeout. A test that the timer must not
+// disturb gives an interval longer than the test runs.
+func newPeerHandler(
+	db database.Store, logger *logger.Logger, batchTimeout time.Duration,
+) *PeerHandler {
 	h := &PeerHandler{
 		db:          db,
 		logger:      logger,
 		peerBatch:   make([]peerUpdate, 0, peerBatchSize),
 		batchFullCh: make(chan struct{}, 1),
-		lastFlush:   time.Now(),
 		stopCh:      make(chan context.Context, 1),
 		done:        make(chan struct{}),
 	}
 	h.batchTaken.L = &h.mu
 
 	// Start the flush timer goroutine
-	go h.flushLoop()
+	go flushLoop(batchTimeout, h.batchFullCh, h.stopCh, h.done, h.flushBatch)
 
 	return h
 }
@@ -173,33 +180,6 @@ func (h *PeerHandler) Stop(ctx context.Context) {
 	}
 }
 
-// flushLoop runs in a goroutine, the only one that writes the batch: it
-// flushes when the batch is full, periodically, and once more on Stop.
-func (h *PeerHandler) flushLoop() {
-	defer close(h.done)
-
-	ticker := time.NewTicker(peerBatchTimeout)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-ticker.C:
-			if time.Since(h.lastFlush) >= peerBatchTimeout {
-				h.flushBatch()
-			}
-		case <-h.batchFullCh:
-			h.flushBatch()
-		case ctx := <-h.stopCh:
-			// Final flush, unless the stop deadline has passed
-			if ctx.Err() == nil {
-				h.flushBatch()
-			}
-
-			return
-		}
-	}
-}
-
 // takeBatch returns the pending updates and leaves an empty batch in their
 // place for HandleMessage to fill. They count in writing until flushBatch has
 // written them.
@@ -220,11 +200,11 @@ func (h *PeerHandler) takeBatch() []peerUpdate {
 }
 
 // flushBatch takes the pending batch and writes it to the database, without
-// holding the mutex during the write.
-func (h *PeerHandler) flushBatch() {
+// holding the mutex during the write. It reports whether there was a batch.
+func (h *PeerHandler) flushBatch() bool {
 	batch := h.takeBatch()
 	if len(batch) == 0 {
-		return
+		return false
 	}
 
 	// The earliest and the latest time the batch saw each peer, with the peer
@@ -255,5 +235,5 @@ func (h *PeerHandler) flushBatch() {
 	h.writing = 0
 	h.mu.Unlock()
 
-	h.lastFlush = time.Now()
+	return true
 }
diff --git a/internal/routewatch/peeringhandler.go b/internal/routewatch/peeringhandler.go
index 8e5f801..046a210 100644
--- a/internal/routewatch/peeringhandler.go
+++ b/internal/routewatch/peeringhandler.go
@@ -62,6 +62,15 @@ type PeeringHandler struct {
 // periodically processes accumulated AS paths into peering records. The
 // handler begins processing immediately upon creation.
 func NewPeeringHandler(db database.Store, logger *logger.Logger) *PeeringHandler {
+	return newPeeringHandler(db, logger, peeringProcessInterval)
+}
+
+// newPeeringHandler is NewPeeringHandler with the processing timer's interval,
+// processInterval, in place of peeringProcessInterval. A test gives a short
+// one to have the timer start a write.
+func newPeeringHandler(
+	db database.Store, logger *logger.Logger, processInterval time.Duration,
+) *PeeringHandler {
 	h := &PeeringHandler{
 		db:      db,
 		logger:  logger,
@@ -71,7 +80,7 @@ func NewPeeringHandler(db database.Store, logger *logger.Logger) *PeeringHandler
 	}
 
 	// Start the periodic processing goroutine
-	go h.processLoop()
+	go h.processLoop(processInterval)
 
 	return h
 }
@@ -163,15 +172,29 @@ func (h *PeeringHandler) Stop(ctx context.Context) {
 	}
 }
 
-// processLoop runs periodically to process AS paths into peerings, and once
+// processLoop processes AS paths into peerings every processInterval, and once
 // more on Stop.
-func (h *PeeringHandler) processLoop() {
+func (h *PeeringHandler) processLoop(processInterval time.Duration) {
 	defer close(h.done)
 
-	ticker := time.NewTicker(peeringProcessInterval)
+	ticker := time.NewTicker(processInterval)
 	defer ticker.Stop()
 
 	for {
+		// Stop goes first. A write can run past the stop deadline while a
+		// tick waits beside Stop's context. The select below picks one of
+		// them at random, but past the deadline no write may start.
+		select {
+		case ctx := <-h.stopCh:
+			// Final processing, unless the stop deadline has passed
+			if ctx.Err() == nil {
+				h.processPeerings()
+			}
+
+			return
+		default:
+		}
+
 		select {
 		case <-ticker.C:
 			h.processPeerings()
diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go
index e02f424..6835699 100644
--- a/internal/routewatch/prefixhandler.go
+++ b/internal/routewatch/prefixhandler.go
@@ -96,7 +96,6 @@ type PrefixHandler struct {
 	// &mu.
 	batchTaken  sync.Cond
 	batchFullCh chan struct{} // wakes the flush goroutine when batch is full
-	lastFlush   time.Time     // used only by the flush goroutine
 	// stopCh hands Stop's context to the flush goroutine. It holds one, so
 	// Stop does not wait for a write in progress to hand it over.
 	stopCh chan context.Context
@@ -147,14 +146,13 @@ func newPrefixHandler(
 		table:       table,
 		batch:       make([]prefixUpdate, 0, prefixBatchSize),
 		batchFullCh: make(chan struct{}, 1),
-		lastFlush:   time.Now(),
 		stopCh:      make(chan context.Context, 1),
 		done:        make(chan struct{}),
 	}
 	h.batchTaken.L = &h.mu
 
 	// Start the flush timer goroutine
-	go h.flushLoop(batchTimeout)
+	go flushLoop(batchTimeout, h.batchFullCh, h.stopCh, h.done, h.flushBatch)
 
 	return h
 }
@@ -324,48 +322,6 @@ func (h *PrefixHandler) peerSessionDown(peer string) {
 	}
 }
 
-// flushLoop runs in a goroutine, the only one that writes the batch: it
-// flushes when the batch is full, every batchTimeout, and once more on Stop.
-func (h *PrefixHandler) flushLoop(batchTimeout time.Duration) {
-	defer close(h.done)
-
-	ticker := time.NewTicker(batchTimeout)
-	defer ticker.Stop()
-
-	for {
-		// Stop goes first. A write can run past the stop deadline while a full
-		// batch's wake-up or a tick waits beside Stop's context. The select
-		// below picks one of them at random, but past the deadline no flush
-		// may start.
-		select {
-		case ctx := <-h.stopCh:
-			// Final flush, unless the stop deadline has passed
-			if ctx.Err() == nil {
-				h.flushBatch()
-			}
-
-			return
-		default:
-		}
-
-		select {
-		case <-ticker.C:
-			if time.Since(h.lastFlush) >= batchTimeout {
-				h.flushBatch()
-			}
-		case <-h.batchFullCh:
-			h.flushBatch()
-		case ctx := <-h.stopCh:
-			// Final flush, unless the stop deadline has passed
-			if ctx.Err() == nil {
-				h.flushBatch()
-			}
-
-			return
-		}
-	}
-}
-
 // takeBatch returns the pending updates, the held ones first, and the peers in
 // peersDown, and leaves an empty batch, no held updates and no peers in their
 // place for HandleMessage to fill. The updates count in writing until
@@ -396,12 +352,13 @@ func (h *PrefixHandler) takeBatch() ([]prefixUpdate, []string) {
 
 // flushBatch writes the routing table's changes, then takes the pending batch
 // and writes it to the database, without holding the mutex during the writes.
-func (h *PrefixHandler) flushBatch() {
+// It reports whether there was a batch, or a peer down to delete the routes of.
+func (h *PrefixHandler) flushBatch() bool {
 	h.writeTableChanges()
 
 	batch, peersDown := h.takeBatch()
 	if len(batch) == 0 && len(peersDown) == 0 {
-		return
+		return false
 	}
 
 	startTime := time.Now()
@@ -485,7 +442,7 @@ func (h *PrefixHandler) flushBatch() {
 		"duration_ms", elapsed.Milliseconds(),
 	)
 
-	h.lastFlush = time.Now()
+	return true
 }
 
 // applyToTable applies update to the routing table in memory: as an
diff --git a/internal/routewatch/prefixhandler_internal_test.go b/internal/routewatch/prefixhandler_internal_test.go
index 53c4fb6..2e95267 100644
--- a/internal/routewatch/prefixhandler_internal_test.go
+++ b/internal/routewatch/prefixhandler_internal_test.go
@@ -981,8 +981,8 @@ func TestPrefixHandlerStopLogsNothingAfterADeleteThatSucceeds(t *testing.T) {
 }
 
 // longerThanAnyTest is a flush timer interval that no test lasts. A test whose
-// prefix handler must flush only when its batch is full and on Stop gives it to
-// newPrefixHandler.
+// prefix, AS or peer handler must flush only when its batch is full and on Stop
+// gives it to newPrefixHandler, newASHandler or newPeerHandler.
 const longerThanAnyTest = time.Hour
 
 // prefixHandlerWithoutTimer returns a prefix handler that writes to store and