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

@@ -8,7 +8,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"math/rand/v2"
 	"net"
 	"net/netip"
 	"os"
@@ -62,17 +61,6 @@ const (
 	sqliteHardHeapLimitBytes = 1610612736
 )
 
-// Bounds on the work GetRandomPrefixesByLength does to pick its sample.
-const (
-	// maxPrefixesToShuffle is the most prefixes of one mask length that are
-	// read in full and shuffled to pick the sample. A mask length with more is
-	// sampled with random lookups.
-	maxPrefixesToShuffle = 5000
-	// sampleLookupsPerPrefix is the most random lookups made for each prefix
-	// asked for.
-	sampleLookupsPerPrefix = 4
-)
-
 // Common errors
 var (
 	// ErrInvalidIP is returned when an IP address is malformed
@@ -1723,183 +1711,45 @@ func (d *Database) GetRandomPrefixesByLength(maskLength, ipVersion, limit int) (
 	return d.GetRandomPrefixesByLengthContext(context.Background(), maskLength, ipVersion, limit)
 }
 
-// GetRandomPrefixesByLengthContext returns limit prefixes with the given mask
-// length picked at random, or all of them when there are fewer, each once with
-// its most recently updated live route, for a limit up to
-// maxPrefixesToShuffle. However many routes there are, it reads at most
-// maxPrefixesToShuffle+1 prefixes and makes at most sampleLookupsPerPrefix
-// lookups for each prefix asked for.
+// GetRandomPrefixesByLengthContext returns a random sample of prefixes with context support
 func (d *Database) GetRandomPrefixesByLengthContext(
 	ctx context.Context, maskLength, ipVersion, limit int) ([]LiveRoute, error) {
+	// Select unique prefixes with their most recent route information
 	tableName := "live_routes_v4"
 	if ipVersion == ipVersionV6 {
 		tableName = "live_routes_v6"
 	}
 
-	// Reading one prefix more than maxPrefixesToShuffle tells whether these
-	// are all the prefixes with this mask length.
-	prefixes, err := d.firstPrefixesByLength(ctx, tableName, maskLength, maxPrefixesToShuffle+1)
-	if err != nil {
-		return nil, err
-	}
-	rand.Shuffle(len(prefixes), func(i, j int) {
-		prefixes[i], prefixes[j] = prefixes[j], prefixes[i]
-	})
-	if len(prefixes) > maxPrefixesToShuffle {
-		prefixes, err = d.samplePrefixesByLength(ctx, tableName, maskLength, limit, prefixes)
-		if err != nil {
-			return nil, err
-		}
-	}
-	prefixes = prefixes[:min(limit, len(prefixes))]
-
-	return d.newestRouteOfEachPrefix(ctx, tableName, ipVersion, prefixes)
-}
-
-// firstPrefixesByLength returns the first n distinct prefixes, in text order,
-// with the given mask length. It reads only their entries in the mask length
-// and prefix index.
-func (d *Database) firstPrefixesByLength(
-	ctx context.Context, tableName string, maskLength, n int) ([]string, error) {
 	//nolint:gosec // Table name is hardcoded based on IP version
 	query := fmt.Sprintf(`
-		SELECT DISTINCT prefix FROM %s
-		WHERE mask_length = ?
-		ORDER BY prefix
-		LIMIT ?
-	`, tableName)
+		WITH unique_prefixes AS (
+			SELECT prefix, MAX(last_updated) as max_updated
+			FROM %s
+			WHERE mask_length = ?
+			GROUP BY prefix
+			ORDER BY RANDOM()
+			LIMIT ?
+		)
+		SELECT lr.prefix, lr.mask_length, lr.origin_asn, lr.as_path, 
+			lr.peer_ip, lr.last_updated
+		FROM %s lr
+		INNER JOIN unique_prefixes up ON lr.prefix = up.prefix AND lr.last_updated = up.max_updated
+		WHERE lr.mask_length = ?
+	`, tableName, tableName)
 
-	rows, err := d.db.QueryContext(ctx, query, maskLength, n)
+	rows, err := d.db.QueryContext(ctx, query, maskLength, limit, maskLength)
 	if err != nil {
-		return nil, fmt.Errorf("failed to list prefixes: %w", err)
+		return nil, fmt.Errorf("failed to query random prefixes: %w", err)
 	}
-	defer func() { _ = rows.Close() }()
+	defer func() {
+		_ = rows.Close()
+	}()
 
-	var prefixes []string
+	var routes []LiveRoute
 	for rows.Next() {
-		var prefix string
-		if err := rows.Scan(&prefix); err != nil {
-			return nil, fmt.Errorf("failed to scan prefix: %w", err)
-		}
-		prefixes = append(prefixes, prefix)
-	}
-	if err := rows.Err(); err != nil {
-		return nil, fmt.Errorf("failed to list prefixes: %w", err)
-	}
-
-	return prefixes, nil
-}
-
-// samplePrefixesByLength picks limit distinct prefixes with the given mask
-// length at random, in at most sampleLookupsPerPrefix lookups for each prefix
-// asked for. ORDER BY RANDOM() would read every route with that mask length.
-// Instead each lookup draws a row id between those of the first and last route
-// with that mask length and takes the prefix of the first such route at or
-// after it. SQLite keeps the row id at the end of every index entry, so the
-// mask length index is ordered by mask length, then row id, and the lookup is
-// one search of it.
-//
-// The sample is uneven, as the page needs a varied sample, not an even one: a
-// route is picked in proportion to the row ids between it and the previous
-// route with that mask length, including the row ids of routes with other mask
-// lengths and of withdrawn routes. A route that follows a long run of
-// withdrawn routes is picked far more often than one that directly follows
-// another route with that mask length.
-//
-// Should the lookups keep landing on prefixes already picked, the rest are
-// taken from shuffled, prefixes of that length in random order.
-func (d *Database) samplePrefixesByLength(
-	ctx context.Context, tableName string, maskLength, limit int, shuffled []string) ([]string, error) {
-	//nolint:gosec // Table name is hardcoded based on IP version
-	rangeQuery := fmt.Sprintf(`
-		SELECT
-			(SELECT MIN(rowid) FROM %[1]s WHERE mask_length = ?),
-			(SELECT MAX(rowid) FROM %[1]s WHERE mask_length = ?)
-	`, tableName)
-
-	var first, last sql.NullInt64
-	if err := d.db.QueryRowContext(ctx, rangeQuery, maskLength, maskLength).Scan(&first, &last); err != nil {
-		return nil, fmt.Errorf("failed to read row id range: %w", err)
-	}
-	if !first.Valid {
-		// Every route with this mask length was withdrawn since they were listed.
-		return nil, nil
-	}
-
-	//nolint:gosec // Table name is hardcoded based on IP version
-	lookupQuery := fmt.Sprintf(`
-		SELECT prefix FROM %s
-		WHERE mask_length = ? AND rowid >= ?
-		ORDER BY rowid
-		LIMIT 1
-	`, tableName)
-
-	stmt, err := d.db.PrepareContext(ctx, lookupQuery)
-	if err != nil {
-		return nil, fmt.Errorf("failed to prepare prefix lookup: %w", err)
-	}
-	defer func() { _ = stmt.Close() }()
-
-	picked := make(map[string]bool, limit)
-	prefixes := make([]string, 0, limit)
-	pick := func(prefix string) {
-		if !picked[prefix] {
-			picked[prefix] = true
-			prefixes = append(prefixes, prefix)
-		}
-	}
-
-	for lookups := 0; lookups < limit*sampleLookupsPerPrefix && len(prefixes) < limit; lookups++ {
-		//nolint:gosec // The sample needs variety, not unpredictability.
-		rowID := first.Int64 + rand.Int64N(last.Int64-first.Int64+1)
-
-		var prefix string
-		err := stmt.QueryRowContext(ctx, maskLength, rowID).Scan(&prefix)
-		if errors.Is(err, sql.ErrNoRows) {
-			// The routes from rowID on were withdrawn since the range was read.
-			continue
-		}
-		if err != nil {
-			return nil, fmt.Errorf("failed to look up prefix: %w", err)
-		}
-		pick(prefix)
-	}
-
-	for _, prefix := range shuffled {
-		if len(prefixes) == limit {
-			break
-		}
-		pick(prefix)
-	}
-
-	return prefixes, nil
-}
-
-// newestRouteOfEachPrefix returns the most recently updated live route of each
-// prefix, one lookup on the prefix index per prefix. A prefix whose routes were
-// all withdrawn since it was picked is left out.
-func (d *Database) newestRouteOfEachPrefix(
-	ctx context.Context, tableName string, ipVersion int, prefixes []string) ([]LiveRoute, error) {
-	//nolint:gosec // Table name is hardcoded based on IP version
-	query := fmt.Sprintf(`
-		SELECT prefix, mask_length, origin_asn, as_path, peer_ip, last_updated
-		FROM %s
-		WHERE prefix = ?
-		ORDER BY last_updated DESC
-		LIMIT 1
-	`, tableName)
-
-	stmt, err := d.db.PrepareContext(ctx, query)
-	if err != nil {
-		return nil, fmt.Errorf("failed to prepare route query: %w", err)
-	}
-	defer func() { _ = stmt.Close() }()
-
-	routes := make([]LiveRoute, 0, len(prefixes))
-	for _, prefix := range prefixes {
-		route := LiveRoute{IPVersion: ipVersion}
+		var route LiveRoute
 		var pathJSON string
-		err := stmt.QueryRowContext(ctx, prefix).Scan(
+		err := rows.Scan(
 			&route.Prefix,
 			&route.MaskLength,
 			&route.OriginASN,
@@ -1907,16 +1757,17 @@ func (d *Database) newestRouteOfEachPrefix(
 			&route.PeerIP,
 			&route.LastUpdated,
 		)
-		if errors.Is(err, sql.ErrNoRows) {
+		// Set IP version based on which table we queried
+		route.IPVersion = ipVersion
+		if err != nil {
 			continue
 		}
-		if err != nil {
-			return nil, fmt.Errorf("failed to query route of %s: %w", prefix, err)
-		}
 
+		// Decode AS path
 		if err := json.Unmarshal([]byte(pathJSON), &route.ASPath); err != nil {
 			route.ASPath = []int{}
 		}
+
 		routes = append(routes, route)
 	}
 

internal/database/database_test.go

@@ -6,7 +6,6 @@ import (
 	"errors"
 	"fmt"
 	"net/netip"
-	"slices"
 	"sync"
 	"testing"
 	"time"
@@ -46,32 +45,6 @@ const (
 	asnSecondBand = 100
 )
 
-// Parameters for the prefix-length sample tests.
-const (
-	// sampleLimit is how many prefixes the tests ask the sample for.
-	sampleLimit = 10
-	// sampledPrefixCount is how many prefixes the sampling test stores, more
-	// than sampleLimit so the sample has to choose among them.
-	sampledPrefixCount = 30
-	// secondV4PeerIP and secondV6PeerIP are the second peer each prefix is
-	// learned from, besides the one mkV4Route and mkV6Route set.
-	secondV4PeerIP = "192.0.2.2"
-	secondV6PeerIP = "2001:db8::2"
-	// largeSamplePrefixCount is how many prefixes the test of a mask length
-	// with more than maxPrefixesToShuffle prefixes stores: twice that, so about
-	// half of them come after the first maxPrefixesToShuffle+1 in text order.
-	largeSamplePrefixCount = 2 * maxPrefixesToShuffle
-	// largeSampleLimit is how many prefixes that test asks for. Each random
-	// lookup lands after the first maxPrefixesToShuffle+1 in text order about
-	// half the time, so none of them doing so is about as likely as
-	// largeSampleLimit coin tosses all coming up heads.
-	largeSampleLimit = 30
-	// liveRoutesV4Table and liveRoutesV6Table are the live route tables the
-	// tests pass to the sampling steps.
-	liveRoutesV4Table = "live_routes_v4"
-	liveRoutesV6Table = "live_routes_v6"
-)
-
 // TestGetIPInfoFindsMostSpecificLiveRoute stores nested live prefixes for both
 // families and checks that a lookup returns the most specific one covering the
 // address, ErrNoRoute when none covers it, and the next less specific prefix
@@ -416,281 +389,3 @@ func TestBatchWriteDuringCheckpoint(t *testing.T) {
 	cancel()
 	wg.Wait()
 }
-
-// TestGetRandomPrefixesByLengthListsPrefixOnce stores one prefix of each
-// family learned from two peers in the same second and checks that the
-// prefix-length sample lists each of them once.
-func TestGetRandomPrefixesByLengthListsPrefixOnce(t *testing.T) {
-	cfg := &config.Config{StateDir: t.TempDir()}
-
-	db, err := New(cfg, logger.New())
-	if err != nil {
-		t.Fatalf("failed to create database: %v", err)
-	}
-	defer func() { _ = db.Close() }()
-
-	ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
-	v4First := mkV4Route(t, "198.51.100.0/24", 64500, ts)
-	v4Second := mkV4Route(t, "198.51.100.0/24", 64500, ts)
-	v4Second.PeerIP = secondV4PeerIP
-	v6First := mkV6Route("2001:db8::/32", 64500, ts)
-	v6Second := mkV6Route("2001:db8::/32", 64500, ts)
-	v6Second.PeerIP = secondV6PeerIP
-	if err := db.UpsertLiveRouteBatch([]*LiveRoute{v4First, v4Second, v6First, v6Second}); err != nil {
-		t.Fatalf("UpsertLiveRouteBatch: %v", err)
-	}
-
-	for _, want := range []*LiveRoute{v4First, v6First} {
-		routes, err := db.GetRandomPrefixesByLength(want.MaskLength, want.IPVersion, sampleLimit)
-		if err != nil {
-			t.Fatalf("GetRandomPrefixesByLength(%d, %d): %v", want.MaskLength, want.IPVersion, err)
-		}
-		if len(routes) != 1 || routes[0].Prefix != want.Prefix {
-			t.Errorf("GetRandomPrefixesByLength(%d, %d) = %+v, want %s listed once",
-				want.MaskLength, want.IPVersion, routes, want.Prefix)
-		}
-	}
-}
-
-// TestGetRandomPrefixesByLengthListsNewestRoute stores a prefix learned from
-// two peers at different times, the older route first, and checks that the
-// prefix-length sample lists the newer route.
-func TestGetRandomPrefixesByLengthListsNewestRoute(t *testing.T) {
-	cfg := &config.Config{StateDir: t.TempDir()}
-
-	db, err := New(cfg, logger.New())
-	if err != nil {
-		t.Fatalf("failed to create database: %v", err)
-	}
-	defer func() { _ = db.Close() }()
-
-	older := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
-	olderRoute := mkV4Route(t, "198.51.100.0/24", 64500, older)
-	newerRoute := mkV4Route(t, "198.51.100.0/24", 64500, older.Add(time.Minute))
-	newerRoute.PeerIP = secondV4PeerIP
-	if err := db.UpsertLiveRouteBatch([]*LiveRoute{olderRoute, newerRoute}); err != nil {
-		t.Fatalf("UpsertLiveRouteBatch: %v", err)
-	}
-
-	routes, err := db.GetRandomPrefixesByLength(newerRoute.MaskLength, ipVersionV4, sampleLimit)
-	if err != nil {
-		t.Fatalf("GetRandomPrefixesByLength: %v", err)
-	}
-	if len(routes) != 1 || routes[0].PeerIP != newerRoute.PeerIP {
-		t.Errorf("GetRandomPrefixesByLength = %+v, want only the newer route, from %s",
-			routes, newerRoute.PeerIP)
-	}
-}
-
-// TestGetRandomPrefixesByLengthSamplesManyPrefixes stores more /24 prefixes
-// than the sample asks for, each followed by a /16 so that routes of another
-// mask length sit between the /24 rows, and each prefix learned from two peers
-// in the same second. It checks both the sample the pages get, picked from the
-// shuffled prefixes, and the random lookups used for a mask length with more
-// than maxPrefixesToShuffle prefixes.
-func TestGetRandomPrefixesByLengthSamplesManyPrefixes(t *testing.T) {
-	cfg := &config.Config{StateDir: t.TempDir()}
-
-	db, err := New(cfg, logger.New())
-	if err != nil {
-		t.Fatalf("failed to create database: %v", err)
-	}
-	defer func() { _ = db.Close() }()
-
-	ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
-	var routes []*LiveRoute
-	var stored []string
-	for i := range sampledPrefixCount {
-		prefix := fmt.Sprintf("10.0.%d.0/24", i)
-		for _, p := range []string{prefix, fmt.Sprintf("172.%d.0.0/16", i)} {
-			first := mkV4Route(t, p, 64500, ts)
-			second := mkV4Route(t, p, 64500, ts)
-			second.PeerIP = secondV4PeerIP
-			routes = append(routes, first, second)
-		}
-		stored = append(stored, prefix)
-	}
-	if err := db.UpsertLiveRouteBatch(routes); err != nil {
-		t.Fatalf("UpsertLiveRouteBatch: %v", err)
-	}
-	slices.Sort(stored)
-
-	maskLength := routes[0].MaskLength
-	sample, err := db.GetRandomPrefixesByLength(maskLength, ipVersionV4, sampleLimit)
-	if err != nil {
-		t.Fatalf("GetRandomPrefixesByLength: %v", err)
-	}
-	var listed []string
-	for _, route := range sample {
-		listed = append(listed, route.Prefix)
-	}
-	checkPrefixSample(t, "GetRandomPrefixesByLength", listed, stored)
-
-	looked, err := db.samplePrefixesByLength(
-		context.Background(), liveRoutesV4Table, maskLength, sampleLimit, nil)
-	if err != nil {
-		t.Fatalf("samplePrefixesByLength: %v", err)
-	}
-	checkPrefixSample(t, "samplePrefixesByLength", looked, stored)
-}
-
-// checkPrefixSample checks that sample lists sampleLimit prefixes, each once
-// and each one of stored, and that it is not just the first sampleLimit of
-// stored, which is in text order.
-func checkPrefixSample(t *testing.T, name string, sample, stored []string) {
-	t.Helper()
-
-	if len(sample) != sampleLimit {
-		t.Errorf("%s listed %d prefixes, want %d", name, len(sample), sampleLimit)
-	}
-	listed := make(map[string]bool)
-	for _, prefix := range sample {
-		if listed[prefix] || !slices.Contains(stored, prefix) {
-			t.Errorf("%s lists %s twice or it is not one of the /24s", name, prefix)
-		}
-		listed[prefix] = true
-	}
-	if slices.Equal(slices.Sorted(slices.Values(sample)), stored[:sampleLimit]) {
-		t.Errorf("%s lists just the first %d prefixes in text order", name, sampleLimit)
-	}
-}
-
-// TestGetRandomPrefixesByLengthSamplesPastFirstPrefixes stores more prefixes
-// of one mask length than maxPrefixesToShuffle. It checks that
-// firstPrefixesByLength reads just the first maxPrefixesToShuffle+1 in text
-// order, and that the sample lists the prefixes asked for, each once, some of
-// them after those first ones, which only the random lookups reach.
-func TestGetRandomPrefixesByLengthSamplesPastFirstPrefixes(t *testing.T) {
-	cfg := &config.Config{StateDir: t.TempDir()}
-
-	db, err := New(cfg, logger.New())
-	if err != nil {
-		t.Fatalf("failed to create database: %v", err)
-	}
-	defer func() { _ = db.Close() }()
-
-	ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
-	var routes []*LiveRoute
-	var stored []string
-	for i := range largeSamplePrefixCount {
-		route := mkV6Route(fmt.Sprintf("2001:%x::/32", i), 64500, ts)
-		routes = append(routes, route)
-		stored = append(stored, route.Prefix)
-	}
-	if err := db.UpsertLiveRouteBatch(routes); err != nil {
-		t.Fatalf("UpsertLiveRouteBatch: %v", err)
-	}
-	slices.Sort(stored)
-	firstPrefixes := stored[:maxPrefixesToShuffle+1]
-
-	maskLength := routes[0].MaskLength
-	read, err := db.firstPrefixesByLength(
-		context.Background(), liveRoutesV6Table, maskLength, maxPrefixesToShuffle+1)
-	if err != nil {
-		t.Fatalf("firstPrefixesByLength: %v", err)
-	}
-	if !slices.Equal(read, firstPrefixes) {
-		t.Errorf("firstPrefixesByLength read %d prefixes, want the first %d in text order",
-			len(read), len(firstPrefixes))
-	}
-
-	sample, err := db.GetRandomPrefixesByLength(maskLength, ipVersionV6, largeSampleLimit)
-	if err != nil {
-		t.Fatalf("GetRandomPrefixesByLength: %v", err)
-	}
-	if len(sample) != largeSampleLimit {
-		t.Errorf("GetRandomPrefixesByLength listed %d prefixes, want %d", len(sample), largeSampleLimit)
-	}
-	listed := make(map[string]bool)
-	pastFirstPrefixes := 0
-	for _, route := range sample {
-		if listed[route.Prefix] {
-			t.Errorf("GetRandomPrefixesByLength lists %s twice", route.Prefix)
-		}
-		listed[route.Prefix] = true
-		if !slices.Contains(firstPrefixes, route.Prefix) {
-			pastFirstPrefixes++
-		}
-	}
-	if pastFirstPrefixes == 0 {
-		t.Errorf("GetRandomPrefixesByLength lists only prefixes among the first %d in text order",
-			len(firstPrefixes))
-	}
-}
-
-// TestSamplePrefixesByLengthFillsUpFromShuffled stores a single prefix, so
-// every random lookup lands on it, and checks that the sample still lists
-// sampleLimit prefixes: the stored one, then the shuffled ones in order. The
-// stored prefix is the second shuffled one, so the fill-up must skip it.
-func TestSamplePrefixesByLengthFillsUpFromShuffled(t *testing.T) {
-	cfg := &config.Config{StateDir: t.TempDir()}
-
-	db, err := New(cfg, logger.New())
-	if err != nil {
-		t.Fatalf("failed to create database: %v", err)
-	}
-	defer func() { _ = db.Close() }()
-
-	var shuffled []string
-	for i := range sampledPrefixCount {
-		shuffled = append(shuffled, fmt.Sprintf("10.0.%d.0/24", i))
-	}
-	route := mkV4Route(t, shuffled[1], 64500, time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC))
-	if err := db.UpsertLiveRouteBatch([]*LiveRoute{route}); err != nil {
-		t.Fatalf("UpsertLiveRouteBatch: %v", err)
-	}
-
-	sample, err := db.samplePrefixesByLength(
-		context.Background(), liveRoutesV4Table, route.MaskLength, sampleLimit, shuffled)
-	if err != nil {
-		t.Fatalf("samplePrefixesByLength: %v", err)
-	}
-	want := append([]string{shuffled[1], shuffled[0]}, shuffled[2:sampleLimit]...)
-	if !slices.Equal(sample, want) {
-		t.Errorf("samplePrefixesByLength = %v, want %v", sample, want)
-	}
-}
-
-// TestGetRandomPrefixesByLengthSkipsWithdrawnRoutes withdraws a route after
-// storing it, as can happen between the steps of GetRandomPrefixesByLength. It
-// checks that the random lookups pick nothing for a mask length with no route
-// left, and that only the prefixes that still have a route get their newest
-// route listed.
-func TestGetRandomPrefixesByLengthSkipsWithdrawnRoutes(t *testing.T) {
-	cfg := &config.Config{StateDir: t.TempDir()}
-
-	db, err := New(cfg, logger.New())
-	if err != nil {
-		t.Fatalf("failed to create database: %v", err)
-	}
-	defer func() { _ = db.Close() }()
-
-	ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
-	kept := mkV4Route(t, "198.51.100.0/24", 64500, ts)
-	withdrawn := mkV4Route(t, "172.16.0.0/16", 64500, ts)
-	if err := db.UpsertLiveRouteBatch([]*LiveRoute{kept, withdrawn}); err != nil {
-		t.Fatalf("UpsertLiveRouteBatch: %v", err)
-	}
-	if err := db.DeleteLiveRoute(withdrawn.Prefix, withdrawn.OriginASN, withdrawn.PeerIP); err != nil {
-		t.Fatalf("DeleteLiveRoute: %v", err)
-	}
-
-	ctx := context.Background()
-	sample, err := db.samplePrefixesByLength(
-		ctx, liveRoutesV4Table, withdrawn.MaskLength, sampleLimit, []string{withdrawn.Prefix})
-	if err != nil {
-		t.Fatalf("samplePrefixesByLength: %v", err)
-	}
-	if len(sample) != 0 {
-		t.Errorf("samplePrefixesByLength = %v, want none for a mask length with no route left", sample)
-	}
-
-	routes, err := db.newestRouteOfEachPrefix(
-		ctx, liveRoutesV4Table, ipVersionV4, []string{withdrawn.Prefix, kept.Prefix})
-	if err != nil {
-		t.Fatalf("newestRouteOfEachPrefix: %v", err)
-	}
-	if len(routes) != 1 || routes[0].Prefix != kept.Prefix {
-		t.Errorf("newestRouteOfEachPrefix = %+v, want only the route of %s", routes, kept.Prefix)
-	}
-}

internal/routewatch/ashandler.go

@@ -32,12 +32,15 @@ type ASHandler struct {
 	db     database.Store
 	logger *logger.Logger
 
-	// Batching
-	mu        sync.Mutex
-	batch     []asnOp
-	lastFlush time.Time
-	stopCh    chan struct{}
-	wg        sync.WaitGroup
+	// Batching. mu guards batch only: the flush goroutine takes the batch
+	// under mu and writes it after releasing mu, so HandleMessage never
+	// waits for a database write.
+	mu          sync.Mutex
+	batch       []asnOp
+	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,
-		logger:    logger,
-		batch:     make([]asnOp, 0, asnBatchSize),
-		lastFlush: time.Now(),
-		stopCh:    make(chan struct{}),
+		db:          db,
+		logger:      logger,
+		batch:       make([]asnOp, 0, asnBatchSize),
+		batchFullCh: make(chan struct{}, 1),
+		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
-// it reaches the configured size threshold.
+// with their associated timestamps. When the batch reaches the configured size
+// 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.flushBatchLocked()
-			h.mu.Unlock()
+			h.flushBatch()
 
 			return
 		}
 	}
 }
 
-// flushBatchLocked flushes the ASN batch to the database (must be called with mutex held)
-func (h *ASHandler) flushBatchLocked() {
+// takeBatch returns the pending operations and leaves an empty batch in their
+// 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/peerhandler.go

@@ -35,12 +35,15 @@ type PeerHandler struct {
 	db     database.Store
 	logger *logger.Logger
 
-	// Batching
-	mu        sync.Mutex
-	peerBatch []peerUpdate
-	lastFlush time.Time
-	stopCh    chan struct{}
-	wg        sync.WaitGroup
+	// Batching. mu guards peerBatch only: the flush goroutine takes the batch
+	// under mu and writes it after releasing mu, so HandleMessage never
+	// waits for a database write.
+	mu          sync.Mutex
+	peerBatch   []peerUpdate
+	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,
-		logger:    logger,
-		peerBatch: make([]peerUpdate, 0, peerBatchSize),
-		lastFlush: time.Now(),
-		stopCh:    make(chan struct{}),
+		db:          db,
+		logger:      logger,
+		peerBatch:   make([]peerUpdate, 0, peerBatchSize),
+		batchFullCh: make(chan struct{}, 1),
+		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
-// or the batch timeout expires, the batch is flushed to the database.
+// the update to an internal batch. The flush goroutine writes the batch to the
+// 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.flushBatchLocked()
-			h.mu.Unlock()
+			h.flushBatch()
 
 			return
 		}
 	}
 }
 
-// flushBatchLocked flushes the peer batch to the database (must be called with mutex held)
-func (h *PeerHandler) flushBatchLocked() {
+// takeBatch returns the pending updates and leaves an empty batch in their
+// 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
-	mu        sync.Mutex
-	batch     []prefixUpdate
-	lastFlush time.Time
-	stopCh    chan struct{}
-	wg        sync.WaitGroup
+	// Batching. mu guards batch only: the flush goroutine takes the batch
+	// under mu and writes it after releasing mu, so HandleMessage never
+	// waits for a database write.
+	mu          sync.Mutex
+	batch       []prefixUpdate
+	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,
-		logger:    logger,
-		batch:     make([]prefixUpdate, 0, prefixBatchSize),
-		lastFlush: time.Now(),
-		stopCh:    make(chan struct{}),
+		db:          db,
+		logger:      logger,
+		batch:       make([]prefixUpdate, 0, prefixBatchSize),
+		batchFullCh: make(chan struct{}, 1),
+		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.flushBatchLocked()
-			h.mu.Unlock()
+			h.flushBatch()
 
 			return
 		}
 	}
 }
 
-// flushBatchLocked flushes the prefix batch to the database (must be called with mutex held)
-func (h *PrefixHandler) flushBatchLocked() {
+// takeBatch returns the pending updates and leaves an empty batch in their
+// 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,9 +159,7 @@ func TestPrefixHandlerAppliesEachPeersLastUpdateInABatch(t *testing.T) {
 				for _, msg := range batch {
 					h.HandleMessage(msg)
 				}
-				h.mu.Lock()
-				h.flushBatchLocked()
-				h.mu.Unlock()
+				h.flushBatch()
 			}
 
 			ctx := context.Background()