Write each handler batch from its flush goroutine, outside the mutex (closes #65)

The AS, peer and prefix handlers wrote their batch to the database while
holding the mutex HandleMessage needs, and HandleMessage wrote a full batch
itself. For as long as a write took, the handler took nothing off its queue,
and the streamer dropped updates once the queue was half full.

Now only the flush goroutine writes: it takes the batch under the mutex,
releases it, then writes. A full batch wakes the flush goroutine instead.
Stop still writes what the handler holds.

While a write runs, the next batch has no size limit; the README memory
section says so.

Model: opus-5-5

README.md

@@ -233,6 +233,10 @@ What happens at each limit:
   keeps running.
 - Handler queues: each of the four handler queues holds at most 20,000 messages.
   When a queue fills, the streamer drops messages instead of blocking.
+- Handler batches: while the AS, peer or prefix handler writes a batch to the
+  database, the updates that arrive collect in its next batch, which has no size
+  limit. A database that stays slower than the feed grows memory instead of
+  dropping updates.
 
 With `DEBUG=routewatch` the daemon logs a `System stats` line every 60 seconds
 with the goroutine count and Go memory figures.

internal/database/database.go

@@ -1632,16 +1632,17 @@ func (d *Database) GetASPeersContext(ctx context.Context, asn int) ([]ASPeer, er
 }
 
 // GetPrefixDetails returns detailed information about a prefix
-func (d *Database) GetPrefixDetails(prefix string) ([]LiveRoute, error) {
+func (d *Database) GetPrefixDetails(prefix string) ([]LiveRoute, map[int]ASN, error) {
 	return d.GetPrefixDetailsContext(context.Background(), prefix)
 }
 
-// GetPrefixDetailsContext returns detailed information about a prefix with context support
+// GetPrefixDetailsContext returns the live routes for a prefix and, keyed by
-func (d *Database) GetPrefixDetailsContext(ctx context.Context, prefix string) ([]LiveRoute, error) {
+// AS number, the handle and description of each AS that originates it.
+func (d *Database) GetPrefixDetailsContext(ctx context.Context, prefix string) ([]LiveRoute, map[int]ASN, error) {
 	// Determine if it's IPv4 or IPv6 by parsing the prefix
 	_, ipnet, err := net.ParseCIDR(prefix)
 	if err != nil {
-		return nil, fmt.Errorf("invalid prefix format: %w", err)
+		return nil, nil, fmt.Errorf("invalid prefix format: %w", err)
 	}
 
 	tableName := "live_routes_v4"
@@ -1663,11 +1664,12 @@ func (d *Database) GetPrefixDetailsContext(ctx context.Context, prefix string) (
 
 	rows, err := d.db.QueryContext(ctx, query, prefix)
 	if err != nil {
-		return nil, fmt.Errorf("failed to query prefix details: %w", err)
+		return nil, nil, fmt.Errorf("failed to query prefix details: %w", err)
 	}
 	defer func() { _ = rows.Close() }()
 
 	var routes []LiveRoute
+	origins := make(map[int]ASN)
 	for rows.Next() {
 		var route LiveRoute
 		var pathJSON string
@@ -1690,13 +1692,18 @@ func (d *Database) GetPrefixDetailsContext(ctx context.Context, prefix string) (
 		route.IPVersion = ipVersion
 		route.MaskLength, _ = ipnet.Mask.Size()
 		routes = append(routes, route)
+		origins[route.OriginASN] = ASN{
+			ASN:         route.OriginASN,
+			Handle:      handle.String,
+			Description: description.String,
+		}
 	}
 
 	if len(routes) == 0 {
-		return nil, fmt.Errorf("%w: %s", ErrNoRoute, prefix)
+		return nil, nil, fmt.Errorf("%w: %s", ErrNoRoute, prefix)
 	}
 
-	return routes, nil
+	return routes, origins, nil
 }
 
 // GetRandomPrefixesByLength returns a random sample of prefixes with the specified mask length

internal/database/interface.go

@@ -76,8 +76,8 @@ type Store interface {
 	GetASDetailsContext(ctx context.Context, asn int) (*ASN, []LiveRoute, error)
 	GetASPeers(asn int) ([]ASPeer, error)
 	GetASPeersContext(ctx context.Context, asn int) ([]ASPeer, error)
-	GetPrefixDetails(prefix string) ([]LiveRoute, error)
+	GetPrefixDetails(prefix string) ([]LiveRoute, map[int]ASN, error)
-	GetPrefixDetailsContext(ctx context.Context, prefix string) ([]LiveRoute, error)
+	GetPrefixDetailsContext(ctx context.Context, prefix string) ([]LiveRoute, map[int]ASN, error)
 	GetRandomPrefixesByLength(maskLength, ipVersion, limit int) ([]LiveRoute, error)
 	GetRandomPrefixesByLengthContext(ctx context.Context, maskLength, ipVersion, limit int) ([]LiveRoute, error)
 

internal/routewatch/app_integration_test.go

@@ -252,13 +252,15 @@ func (m *mockStore) GetASDetailsContext(ctx context.Context, asn int) (*database
 }
 
 // GetPrefixDetails mock implementation
-func (m *mockStore) GetPrefixDetails(prefix string) ([]database.LiveRoute, error) {
+func (m *mockStore) GetPrefixDetails(prefix string) ([]database.LiveRoute, map[int]database.ASN, error) {
 	// Return empty routes for now
-	return []database.LiveRoute{}, nil
+	return []database.LiveRoute{}, map[int]database.ASN{}, nil
 }
 
 // GetPrefixDetailsContext mock implementation with context support
-func (m *mockStore) GetPrefixDetailsContext(ctx context.Context, prefix string) ([]database.LiveRoute, error) {
+func (m *mockStore) GetPrefixDetailsContext(
+	ctx context.Context, prefix string,
+) ([]database.LiveRoute, map[int]database.ASN, error) {
 	return m.GetPrefixDetails(prefix)
 }
 

internal/routewatch/ashandler.go

@@ -32,12 +32,15 @@ type ASHandler struct {
 	db     database.Store
 	logger *logger.Logger
 
-	// Batching
+	// Batching. mu guards batch only: the flush goroutine takes the batch
-	mu        sync.Mutex
+	// under mu and writes it after releasing mu, so HandleMessage never
-	batch     []asnOp
+	// waits for a database write.
-	lastFlush time.Time
+	mu          sync.Mutex
-	stopCh    chan struct{}
+	batch       []asnOp
-	wg        sync.WaitGroup
+	batchFullCh chan struct{} // wakes the flush goroutine when batch is full
+	lastFlush   time.Time     // used only by the flush goroutine
+	stopCh      chan struct{}
+	wg          sync.WaitGroup
 }
 
 type asnOp struct {
@@ -52,11 +55,12 @@ type asnOp struct {
 // background goroutine is terminated.
 func NewASHandler(db database.Store, logger *logger.Logger) *ASHandler {
 	h := &ASHandler{
-		db:        db,
+		db:          db,
-		logger:    logger,
+		logger:      logger,
-		batch:     make([]asnOp, 0, asnBatchSize),
+		batch:       make([]asnOp, 0, asnBatchSize),
-		lastFlush: time.Now(),
+		batchFullCh: make(chan struct{}, 1),
-		stopCh:    make(chan struct{}),
+		lastFlush:   time.Now(),
+		stopCh:      make(chan struct{}),
 	}
 
 	// Start the flush timer goroutine
@@ -85,8 +89,8 @@ func (h *ASHandler) QueueCapacity() int {
 // HandleMessage processes a RIS Live BGP message by extracting all ASNs from
 // the AS path and queuing them for batch insertion into the database. The
 // origin ASN (last element in the path) and all transit ASNs are recorded
-// with their associated timestamps. The batch is automatically flushed when
+// with their associated timestamps. When the batch reaches the configured size
-// it reaches the configured size threshold.
+// threshold, the flush goroutine is woken to write it.
 func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
 	// Use the pre-parsed timestamp
 	timestamp := msg.ParsedTimestamp
@@ -118,13 +122,18 @@ func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
 		}
 	}
 
-	// Check if we need to flush
+	// A full batch wakes the flush goroutine. If it is writing, the wake-up
+	// waits in batchFullCh and the batch keeps growing until the write ends.
 	if len(h.batch) >= asnBatchSize {
-		h.flushBatchLocked()
+		select {
+		case h.batchFullCh <- struct{}{}:
+		default: // already woken
+		}
 	}
 }
 
-// flushLoop runs in a goroutine and periodically flushes batches
+// 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 h.wg.Done()
 	ticker := time.NewTicker(asnBatchTimeout)
@@ -133,31 +142,46 @@ func (h *ASHandler) flushLoop() {
 	for {
 		select {
 		case <-ticker.C:
-			h.mu.Lock()
 			if time.Since(h.lastFlush) >= asnBatchTimeout {
-				h.flushBatchLocked()
+				h.flushBatch()
 			}
-			h.mu.Unlock()
+		case <-h.batchFullCh:
+			h.flushBatch()
 		case <-h.stopCh:
 			// Final flush
-			h.mu.Lock()
+			h.flushBatch()
-			h.flushBatchLocked()
-			h.mu.Unlock()
 
 			return
 		}
 	}
 }
 
-// flushBatchLocked flushes the ASN batch to the database (must be called with mutex held)
+// takeBatch returns the pending operations and leaves an empty batch in their
-func (h *ASHandler) flushBatchLocked() {
+// place for HandleMessage to fill.
+func (h *ASHandler) takeBatch() []asnOp {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
 	if len(h.batch) == 0 {
+		return nil
+	}
+	batch := h.batch
+	h.batch = make([]asnOp, 0, asnBatchSize)
+
+	return batch
+}
+
+// flushBatch takes the pending batch and writes it to the database, without
+// holding the mutex during the write.
+func (h *ASHandler) flushBatch() {
+	batch := h.takeBatch()
+	if len(batch) == 0 {
 		return
 	}
 
 	// Process ASNs first (deduped)
 	asnMap := make(map[int]time.Time)
-	for _, op := range h.batch {
+	for _, op := range batch {
 		if existing, ok := asnMap[op.number]; !ok || op.timestamp.After(existing) {
 			asnMap[op.number] = op.timestamp
 		}
@@ -168,8 +192,6 @@ func (h *ASHandler) flushBatchLocked() {
 		h.logger.Error("Failed to process ASN batch", "error", err, "count", len(asnMap))
 	}
 
-	// Clear batch
-	h.batch = h.batch[:0]
 	h.lastFlush = time.Now()
 }
 

internal/routewatch/batchwrite_test.go

@@ -0,0 +1,192 @@
+package routewatch
+
+import (
+	"slices"
+	"sync"
+	"testing"
+	"time"
+
+	"git.eeqj.de/sneak/routewatch/internal/database"
+	"git.eeqj.de/sneak/routewatch/internal/logger"
+	"git.eeqj.de/sneak/routewatch/internal/ristypes"
+)
+
+// handleMessageTimeout is how long a test waits for HandleMessage to return,
+// or for a write to start, before it fails.
+const handleMessageTimeout = 2 * time.Second
+
+// blockingStore is a mockStore whose batch writes for the AS, peer and prefix
+// handlers wait until the test closes release. writeStarted is closed when the
+// first of them starts. It records what each write was given.
+type blockingStore struct {
+	*mockStore
+
+	release      chan struct{}
+	writeStarted chan struct{}
+	startOnce    sync.Once
+
+	mu       sync.Mutex
+	prefixes map[string]bool
+	asns     map[int]bool
+	peers    map[string]bool
+}
+
+func newBlockingStore() *blockingStore {
+	return &blockingStore{
+		mockStore:    newMockStore(),
+		release:      make(chan struct{}),
+		writeStarted: make(chan struct{}),
+		prefixes:     make(map[string]bool),
+		asns:         make(map[int]bool),
+		peers:        make(map[string]bool),
+	}
+}
+
+func (s *blockingStore) waitForRelease() {
+	s.startOnce.Do(func() { close(s.writeStarted) })
+	<-s.release
+}
+
+func (s *blockingStore) UpsertLiveRouteBatch(routes []*database.LiveRoute) error {
+	s.waitForRelease()
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for _, route := range routes {
+		s.prefixes[route.Prefix] = true
+	}
+
+	return nil
+}
+
+func (s *blockingStore) GetOrCreateASNBatch(asns map[int]time.Time) error {
+	s.waitForRelease()
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for asn := range asns {
+		s.asns[asn] = true
+	}
+
+	return nil
+}
+
+func (s *blockingStore) UpdatePeerBatch(peers map[string]database.PeerUpdate) error {
+	s.waitForRelease()
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for peerIP := range peers {
+		s.peers[peerIP] = true
+	}
+
+	return nil
+}
+
+// returnsInTime reports whether f returns within handleMessageTimeout.
+func returnsInTime(f func()) bool {
+	done := make(chan struct{})
+	go func() {
+		f()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		return true
+	case <-time.After(handleMessageTimeout):
+		return false
+	}
+}
+
+// handleWhileAWriteIsBlocked hands h the messages in fill, which fill its
+// batch so that the handler writes it to store, waits until the write starts,
+// and hands h the message extra while store holds the write. HandleMessage
+// must return each time. Then it lets the writes finish and stops h, which
+// writes what it holds.
+func handleWhileAWriteIsBlocked(
+	t *testing.T,
+	store *blockingStore,
+	h interface {
+		HandleMessage(*ristypes.RISMessage)
+		Stop()
+	},
+	fill []*ristypes.RISMessage,
+	extra *ristypes.RISMessage,
+) {
+	t.Helper()
+	defer h.Stop()
+	defer close(store.release)
+
+	if !returnsInTime(func() {
+		for _, msg := range fill {
+			h.HandleMessage(msg)
+		}
+	}) {
+		t.Fatal("HandleMessage did not return after filling the batch: it waits for the database write")
+	}
+
+	select {
+	case <-store.writeStarted:
+	case <-time.After(handleMessageTimeout):
+		t.Fatal("the handler did not write its full batch")
+	}
+
+	if !returnsInTime(func() { h.HandleMessage(extra) }) {
+		t.Fatal("HandleMessage did not return while a database write was blocked")
+	}
+}
+
+func TestPrefixHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) {
+	const (
+		fillPrefix  = "192.0.2.0/24"
+		extraPrefix = "198.51.100.0/24"
+	)
+
+	announce := func(prefixes []string) *ristypes.RISMessage {
+		return &ristypes.RISMessage{
+			Peer:          testPeerIP,
+			Path:          ristypes.ASPath{testASNA},
+			Announcements: []ristypes.RISAnnouncement{{NextHop: testPeerIP, Prefixes: prefixes}},
+		}
+	}
+
+	store := newBlockingStore()
+	h := NewPrefixHandler(store, logger.New())
+	handleWhileAWriteIsBlocked(t, store, h,
+		[]*ristypes.RISMessage{announce(slices.Repeat([]string{fillPrefix}, prefixBatchSize))},
+		announce([]string{extraPrefix}))
+
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	if !store.prefixes[extraPrefix] {
+		t.Errorf("Stop did not write %s, handled while a write was blocked", extraPrefix)
+	}
+}
+
+func TestASHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) {
+	store := newBlockingStore()
+	h := NewASHandler(store, logger.New())
+	handleWhileAWriteIsBlocked(t, store, h,
+		[]*ristypes.RISMessage{{Path: slices.Repeat(ristypes.ASPath{testASNA}, asnBatchSize)}},
+		&ristypes.RISMessage{Path: ristypes.ASPath{testASNB}})
+
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	if !store.asns[testASNB] {
+		t.Errorf("Stop did not write AS%d, handled while a write was blocked", testASNB)
+	}
+}
+
+func TestPeerHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) {
+	const otherPeerIP = "2001:db8:ffff::2"
+
+	store := newBlockingStore()
+	h := NewPeerHandler(store, logger.New())
+	handleWhileAWriteIsBlocked(t, store, h,
+		slices.Repeat([]*ristypes.RISMessage{{Peer: testPeerIP}}, peerBatchSize),
+		&ristypes.RISMessage{Peer: otherPeerIP})
+
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	if !store.peers[otherPeerIP] {
+		t.Errorf("Stop did not write peer %s, handled while a write was blocked", otherPeerIP)
+	}
+}

internal/routewatch/cli.go

@@ -20,6 +20,8 @@ const (
 	debugInterval = 60 * time.Second
 	// bytesPerMB is bytes per megabyte
 	bytesPerMB = 1024 * 1024
+	// failureExitCode is the exit status when RouteWatch stops on an error
+	failureExitCode = 1
 )
 
 // logDebugStats logs goroutine count and memory usage
@@ -83,6 +85,9 @@ func CLIEntry() {
 					go func() {
 						if err := rw.Run(ctx); err != nil {
 							logger.Error("RouteWatch error", "error", err)
+							if err := shutdowner.Shutdown(fx.ExitCode(failureExitCode)); err != nil {
+								logger.Error("Failed to shutdown gracefully", "error", err)
+							}
 						}
 					}()
 

internal/routewatch/peerhandler.go

@@ -35,12 +35,15 @@ type PeerHandler struct {
 	db     database.Store
 	logger *logger.Logger
 
-	// Batching
+	// Batching. mu guards peerBatch only: the flush goroutine takes the batch
-	mu        sync.Mutex
+	// under mu and writes it after releasing mu, so HandleMessage never
-	peerBatch []peerUpdate
+	// waits for a database write.
-	lastFlush time.Time
+	mu          sync.Mutex
-	stopCh    chan struct{}
+	peerBatch   []peerUpdate
-	wg        sync.WaitGroup
+	batchFullCh chan struct{} // wakes the flush goroutine when peerBatch is full
+	lastFlush   time.Time     // used only by the flush goroutine
+	stopCh      chan struct{}
+	wg          sync.WaitGroup
 }
 
 type peerUpdate struct {
@@ -56,11 +59,12 @@ type peerUpdate struct {
 // should be stopped by calling Stop when it is no longer needed.
 func NewPeerHandler(db database.Store, logger *logger.Logger) *PeerHandler {
 	h := &PeerHandler{
-		db:        db,
+		db:          db,
-		logger:    logger,
+		logger:      logger,
-		peerBatch: make([]peerUpdate, 0, peerBatchSize),
+		peerBatch:   make([]peerUpdate, 0, peerBatchSize),
-		lastFlush: time.Now(),
+		batchFullCh: make(chan struct{}, 1),
-		stopCh:    make(chan struct{}),
+		lastFlush:   time.Now(),
+		stopCh:      make(chan struct{}),
 	}
 
 	// Start the flush timer goroutine
@@ -87,8 +91,8 @@ func (h *PeerHandler) QueueCapacity() int {
 
 // HandleMessage processes a RIS message to track peer information.
 // It extracts the peer IP address and ASN from the message and adds
-// the update to an internal batch. When the batch reaches peerBatchSize
+// the update to an internal batch. The flush goroutine writes the batch to the
-// or the batch timeout expires, the batch is flushed to the database.
+// database when it reaches peerBatchSize or the batch timeout expires.
 func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) {
 	// Parse peer ASN from string
 	peerASN := 0
@@ -109,13 +113,18 @@ func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) {
 		timestamp:   msg.ParsedTimestamp,
 	})
 
-	// Check if we need to flush
+	// A full batch wakes the flush goroutine. If it is writing, the wake-up
+	// waits in batchFullCh and the batch keeps growing until the write ends.
 	if len(h.peerBatch) >= peerBatchSize {
-		h.flushBatchLocked()
+		select {
+		case h.batchFullCh <- struct{}{}:
+		default: // already woken
+		}
 	}
 }
 
-// flushLoop runs in a goroutine and periodically flushes batches
+// 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 h.wg.Done()
 	ticker := time.NewTicker(peerBatchTimeout)
@@ -124,31 +133,46 @@ func (h *PeerHandler) flushLoop() {
 	for {
 		select {
 		case <-ticker.C:
-			h.mu.Lock()
 			if time.Since(h.lastFlush) >= peerBatchTimeout {
-				h.flushBatchLocked()
+				h.flushBatch()
 			}
-			h.mu.Unlock()
+		case <-h.batchFullCh:
+			h.flushBatch()
 		case <-h.stopCh:
 			// Final flush
-			h.mu.Lock()
+			h.flushBatch()
-			h.flushBatchLocked()
-			h.mu.Unlock()
 
 			return
 		}
 	}
 }
 
-// flushBatchLocked flushes the peer batch to the database (must be called with mutex held)
+// takeBatch returns the pending updates and leaves an empty batch in their
-func (h *PeerHandler) flushBatchLocked() {
+// place for HandleMessage to fill.
+func (h *PeerHandler) takeBatch() []peerUpdate {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
 	if len(h.peerBatch) == 0 {
+		return nil
+	}
+	batch := h.peerBatch
+	h.peerBatch = make([]peerUpdate, 0, peerBatchSize)
+
+	return batch
+}
+
+// flushBatch takes the pending batch and writes it to the database, without
+// holding the mutex during the write.
+func (h *PeerHandler) flushBatch() {
+	batch := h.takeBatch()
+	if len(batch) == 0 {
 		return
 	}
 
 	// Deduplicate by peer IP, keeping the latest update for each peer
 	peerMap := make(map[string]peerUpdate)
-	for _, update := range h.peerBatch {
+	for _, update := range batch {
 		if existing, ok := peerMap[update.peerIP]; !ok || update.timestamp.After(existing.timestamp) {
 			peerMap[update.peerIP] = update
 		}
@@ -170,8 +194,6 @@ func (h *PeerHandler) flushBatchLocked() {
 		h.logger.Error("Failed to process peer batch", "error", err, "count", len(dbPeerMap))
 	}
 
-	// Clear batch
-	h.peerBatch = h.peerBatch[:0]
 	h.lastFlush = time.Now()
 }
 

internal/routewatch/prefixhandler.go

@@ -40,12 +40,15 @@ type PrefixHandler struct {
 	logger  *logger.Logger
 	metrics *metrics.Tracker
 
-	// Batching
+	// Batching. mu guards batch only: the flush goroutine takes the batch
-	mu        sync.Mutex
+	// under mu and writes it after releasing mu, so HandleMessage never
-	batch     []prefixUpdate
+	// waits for a database write.
-	lastFlush time.Time
+	mu          sync.Mutex
-	stopCh    chan struct{}
+	batch       []prefixUpdate
-	wg        sync.WaitGroup
+	batchFullCh chan struct{} // wakes the flush goroutine when batch is full
+	lastFlush   time.Time     // used only by the flush goroutine
+	stopCh      chan struct{}
+	wg          sync.WaitGroup
 }
 
 type prefixUpdate struct {
@@ -60,11 +63,12 @@ type prefixUpdate struct {
 // NewPrefixHandler creates a new batched prefix tracking handler
 func NewPrefixHandler(db database.Store, logger *logger.Logger) *PrefixHandler {
 	h := &PrefixHandler{
-		db:        db,
+		db:          db,
-		logger:    logger,
+		logger:      logger,
-		batch:     make([]prefixUpdate, 0, prefixBatchSize),
+		batch:       make([]prefixUpdate, 0, prefixBatchSize),
-		lastFlush: time.Now(),
+		batchFullCh: make(chan struct{}, 1),
-		stopCh:    make(chan struct{}),
+		lastFlush:   time.Now(),
+		stopCh:      make(chan struct{}),
 	}
 
 	// Start the flush timer goroutine
@@ -140,13 +144,18 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 		}
 	}
 
-	// Check if we need to flush
+	// A full batch wakes the flush goroutine. If it is writing, the wake-up
+	// waits in batchFullCh and the batch keeps growing until the write ends.
 	if len(h.batch) >= prefixBatchSize {
-		h.flushBatchLocked()
+		select {
+		case h.batchFullCh <- struct{}{}:
+		default: // already woken
+		}
 	}
 }
 
-// flushLoop runs in a goroutine and periodically flushes batches
+// 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 *PrefixHandler) flushLoop() {
 	defer h.wg.Done()
 	ticker := time.NewTicker(prefixBatchTimeout)
@@ -155,30 +164,45 @@ func (h *PrefixHandler) flushLoop() {
 	for {
 		select {
 		case <-ticker.C:
-			h.mu.Lock()
 			if time.Since(h.lastFlush) >= prefixBatchTimeout {
-				h.flushBatchLocked()
+				h.flushBatch()
 			}
-			h.mu.Unlock()
+		case <-h.batchFullCh:
+			h.flushBatch()
 		case <-h.stopCh:
 			// Final flush
-			h.mu.Lock()
+			h.flushBatch()
-			h.flushBatchLocked()
-			h.mu.Unlock()
 
 			return
 		}
 	}
 }
 
-// flushBatchLocked flushes the prefix batch to the database (must be called with mutex held)
+// takeBatch returns the pending updates and leaves an empty batch in their
-func (h *PrefixHandler) flushBatchLocked() {
+// place for HandleMessage to fill.
+func (h *PrefixHandler) takeBatch() []prefixUpdate {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
 	if len(h.batch) == 0 {
+		return nil
+	}
+	batch := h.batch
+	h.batch = make([]prefixUpdate, 0, prefixBatchSize)
+
+	return batch
+}
+
+// flushBatch takes the pending batch and writes it to the database, without
+// holding the mutex during the write.
+func (h *PrefixHandler) flushBatch() {
+	batch := h.takeBatch()
+	if len(batch) == 0 {
 		return
 	}
 
 	startTime := time.Now()
-	batchSize := len(h.batch)
+	batchSize := len(batch)
 
 	// For each prefix and peer, keep only the last update the feed delivered.
 	// Updates from different peers for the same prefix are all kept. Arrival
@@ -186,7 +210,7 @@ func (h *PrefixHandler) flushBatchLocked() {
 	// updates within one second carry the same one.
 	type prefixAndPeer struct{ prefix, peer string }
 	lastUpdates := make(map[prefixAndPeer]prefixUpdate)
-	for _, update := range h.batch {
+	for _, update := range batch {
 		lastUpdates[prefixAndPeer{update.prefix, update.peer}] = update
 	}
 
@@ -261,8 +285,6 @@ func (h *PrefixHandler) flushBatchLocked() {
 		"duration_ms", elapsed.Milliseconds(),
 	)
 
-	// Clear batch
-	h.batch = h.batch[:0]
 	h.lastFlush = time.Now()
 }
 

internal/routewatch/prefixhandler_test.go

@@ -30,11 +30,6 @@ func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) {
 
 	// Built without NewPrefixHandler's flush timer, so the test flushes itself.
 	h := &PrefixHandler{db: db, logger: logger.New()}
-	flush := func() {
-		h.mu.Lock()
-		defer h.mu.Unlock()
-		h.flushBatchLocked()
-	}
 
 	ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
 	h.HandleMessage(&ristypes.RISMessage{
@@ -46,7 +41,7 @@ func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) {
 			Prefixes: []string{"2001:db8:1::/48", "192.0.2.0/24"},
 		}},
 	})
-	flush()
+	h.flushBatch()
 
 	for ip, want := range map[string]string{
 		"2001:db8:1::1": "2001:db8:1::/48",
@@ -66,7 +61,7 @@ func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) {
 		ParsedTimestamp: ts.Add(time.Minute),
 		Withdrawals:     []string{"2001:db8:1:0:0:0:0:0/48", "192.0.2.0/24"},
 	})
-	flush()
+	h.flushBatch()
 
 	for _, ip := range []string{"2001:db8:1::1", "192.0.2.1"} {
 		if info, err := db.GetIPInfo(ip); !errors.Is(err, database.ErrNoRoute) {
@@ -164,13 +159,11 @@ func TestPrefixHandlerAppliesEachPeersLastUpdateInABatch(t *testing.T) {
 				for _, msg := range batch {
 					h.HandleMessage(msg)
 				}
-				h.mu.Lock()
+				h.flushBatch()
-				h.flushBatchLocked()
-				h.mu.Unlock()
 			}
 
 			ctx := context.Background()
-			routes, err := db.GetPrefixDetailsContext(ctx, prefix)
+			routes, _, err := db.GetPrefixDetailsContext(ctx, prefix)
 			if err != nil && !errors.Is(err, database.ErrNoRoute) {
 				t.Fatalf("GetPrefixDetailsContext: %v", err)
 			}

internal/server/handlers.go

@@ -759,7 +759,7 @@ func (s *Server) handlePrefixDetailJSON() http.HandlerFunc {
 		}
 		prefix := parsed.Masked().String()
 
-		routes, err := s.db.GetPrefixDetailsContext(r.Context(), prefix)
+		routes, _, err := s.db.GetPrefixDetailsContext(r.Context(), prefix)
 		if err != nil {
 			if errors.Is(err, database.ErrNoRoute) {
 				writeJSONError(w, http.StatusNotFound, err.Error())
@@ -930,7 +930,7 @@ func (s *Server) handlePrefixDetail() http.HandlerFunc {
 		}
 		prefix := parsed.Masked().String()
 
-		routes, err := s.db.GetPrefixDetailsContext(r.Context(), prefix)
+		routes, originASNs, err := s.db.GetPrefixDetailsContext(r.Context(), prefix)
 		if err != nil {
 			if errors.Is(err, database.ErrNoRoute) {
 				http.Error(w, "Prefix not found", http.StatusNotFound)
@@ -952,18 +952,10 @@ func (s *Server) handlePrefixDetail() http.HandlerFunc {
 		originMap := make(map[int]*ASNInfo)
 		for _, route := range routes {
 			if _, exists := originMap[route.OriginASN]; !exists {
-				// Get AS info from database
-				asInfo, _, _ := s.db.GetASDetailsContext(r.Context(), route.OriginASN)
-				handle := ""
-				description := ""
-				if asInfo != nil {
-					handle = asInfo.Handle
-					description = asInfo.Description
-				}
 				originMap[route.OriginASN] = &ASNInfo{
 					ASN:         route.OriginASN,
-					Handle:      handle,
+					Handle:      originASNs[route.OriginASN].Handle,
-					Description: description,
+					Description: originASNs[route.OriginASN].Description,
 					PeerCount:   0,
 				}
 			}

internal/server/handlers_test.go

@@ -3,10 +3,13 @@ package server
 import (
 	"context"
 	"encoding/json"
+	"errors"
+	"html"
 	"net/http"
 	"net/http/httptest"
 	"runtime"
 	"slices"
+	"strings"
 	"testing"
 	"time"
 
@@ -15,6 +18,7 @@ import (
 	"git.eeqj.de/sneak/routewatch/internal/logger"
 	"git.eeqj.de/sneak/routewatch/internal/metrics"
 	"git.eeqj.de/sneak/routewatch/internal/streamer"
+	"git.eeqj.de/sneak/routewatch/pkg/asinfo"
 	"github.com/google/uuid"
 )
 
@@ -200,6 +204,72 @@ func TestPrefixDetailHandlersParsePrefix(t *testing.T) {
 	}
 }
 
+// failingASDetailsDB wraps a real store and fails every AS details lookup.
+type failingASDetailsDB struct {
+	database.Store
+}
+
+func (failingASDetailsDB) GetASDetailsContext(_ context.Context, _ int) (*database.ASN, []database.LiveRoute, error) {
+	return nil, nil, errors.New("AS details lookup failed")
+}
+
+// TestPrefixDetailShowsOriginNames checks that the prefix page shows each
+// origin AS's handle and description without looking up the AS's details,
+// which also read every route the AS originates. Before the fix the page did
+// that lookup once per origin, so with it failing the names were missing
+// (https://git.eeqj.de/sneak/routewatch/issues/67). The AS path column shows
+// handles too, so the test looks for the origin list's "(handle)" form.
+func TestPrefixDetailShowsOriginNames(t *testing.T) {
+	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() })
+
+	ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
+	origins := []int{15169, 13335}
+	for _, asn := range origins {
+		if _, err := db.GetOrCreateASN(asn, ts); err != nil {
+			t.Fatalf("GetOrCreateASN(%d): %v", asn, err)
+		}
+	}
+	if err := db.UpsertLiveRouteBatch([]*database.LiveRoute{
+		{
+			ID: uuid.New(), Prefix: "192.0.2.0/24", MaskLength: 24, IPVersion: 4,
+			OriginASN: origins[0], PeerIP: "198.51.100.1", ASPath: []int{origins[0]}, NextHop: "198.51.100.1",
+			LastUpdated: ts,
+		},
+		{
+			ID: uuid.New(), Prefix: "192.0.2.0/24", MaskLength: 24, IPVersion: 4,
+			OriginASN: origins[1], PeerIP: "198.51.100.2", ASPath: []int{origins[1]}, NextHop: "198.51.100.2",
+			LastUpdated: ts,
+		},
+	}); err != nil {
+		t.Fatalf("UpsertLiveRouteBatch: %v", err)
+	}
+
+	s := New(failingASDetailsDB{db}, streamer.New(logger.New(), metrics.New()), logger.New(), &config.Config{})
+	rec := httptest.NewRecorder()
+	s.router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/prefix/192.0.2.0/24", nil))
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d, want %d; body %s", rec.Code, http.StatusOK, rec.Body)
+	}
+
+	page := rec.Body.String()
+	for _, asn := range origins {
+		info, ok := asinfo.Get(asn)
+		if !ok {
+			t.Fatalf("AS%d is not in the AS info data", asn)
+		}
+		if want := "(" + html.EscapeString(info.Handle) + ")"; !strings.Contains(page, want) {
+			t.Errorf("page does not show AS%d's handle %q", asn, want)
+		}
+		if want := html.EscapeString(info.Description); !strings.Contains(page, want) {
+			t.Errorf("page does not show AS%d's description %q", asn, want)
+		}
+	}
+}
+
 // settledGoroutineCount lets transient goroutines finish, then reports the
 // current count.
 func settledGoroutineCount() int {

internal/server/server.go

@@ -3,6 +3,8 @@ package server
 
 import (
 	"context"
+	"fmt"
+	"net"
 	"net/http"
 	"strconv"
 	"time"
@@ -53,9 +55,12 @@ func New(db database.Store, streamer *streamer.Streamer, logger *logger.Logger,
 	return s
 }
 
-// Start starts the HTTP server
+// Start listens on the configured port and serves HTTP in the background.
+// It returns an error if it cannot listen, for example because the port is
+// already in use.
 func (s *Server) Start() error {
 	const (
+		network           = "tcp"
 		readHeaderTimeout = 40 * time.Second
 		readTimeout       = 60 * time.Second
 		writeTimeout      = 60 * time.Second
@@ -73,10 +78,15 @@ func (s *Server) Start() error {
 
 	s.logger.Info("Starting HTTP server", "port", s.port, "addr", s.srv.Addr)
 
-	// Start in goroutine but log when actually listening
+	listener, err := net.Listen(network, s.srv.Addr)
+	if err != nil {
+		return fmt.Errorf("starting HTTP server: %w", err)
+	}
+
+	s.logger.Info("HTTP server listening", "addr", listener.Addr().String())
+
 	go func() {
-		s.logger.Info("HTTP server listening", "addr", s.srv.Addr)
+		if err := s.srv.Serve(listener); err != nil && err != http.ErrServerClosed {
-		if err := s.srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
 			s.logger.Error("HTTP server error", "error", err)
 		}
 	}()

internal/server/server_test.go

@@ -0,0 +1,33 @@
+package server
+
+import (
+	"context"
+	"net"
+	"testing"
+
+	"git.eeqj.de/sneak/routewatch/internal/config"
+	"git.eeqj.de/sneak/routewatch/internal/logger"
+	"git.eeqj.de/sneak/routewatch/internal/metrics"
+	"git.eeqj.de/sneak/routewatch/internal/streamer"
+)
+
+// TestStartFailsWhenPortIsHeld holds a port with a listener of its own and
+// checks that Start on that port returns an error. Before the fix Start
+// listened in a goroutine, only logged the failure and returned nil, so the
+// daemon kept running without its web server.
+func TestStartFailsWhenPortIsHeld(t *testing.T) {
+	held, err := net.Listen("tcp", ":0")
+	if err != nil {
+		t.Fatalf("failed to open the test listener: %v", err)
+	}
+	defer func() { _ = held.Close() }()
+
+	port := held.Addr().(*net.TCPAddr).Port
+	log := logger.New()
+	s := New(nil, streamer.New(log, metrics.New()), log, &config.Config{Port: port})
+	defer func() { _ = s.Stop(context.Background()) }()
+
+	if err := s.Start(); err == nil {
+		t.Fatalf("Start returned nil while port %d was held by another listener", port)
+	}
+}

internal/templates/prefix_detail.html

@@ -283,6 +283,7 @@
                 <div class="origin-item">
                     <a href="/as/{{.ASN}}" class="as-link">AS{{.ASN}}</a>
                     {{if .Handle}} ({{.Handle}}){{end}}
+                    {{if .Description}} {{.Description}}{{end}}
                     <span style="color: #7f8c8d; margin-left: 10px;">{{.PeerCount}} peer{{if ne .PeerCount 1}}s{{end}}</span>
                 </div>
                 {{end}}