From 5ae25a1d48da6f21ab8f37aced6fce4be662dc4c Mon Sep 17 00:00:00 2001 From: sneak Date: Mon, 19 Oct 2026 12:50:03 +0000 Subject: [PATCH] Write batches through one connection with a 1 GiB cache, in key order (closes #96) Every write now uses one connection taken from the pool when the database opens and used under the database lock. Its 1 GiB page cache keeps the pages the route writes touch from one batch to the next. A reading connection's cache is emptied whenever another connection has written, so the 9 reading connections drop to 16 MiB each. The soft and hard heap limits rise to 1.5 and 2 GiB so they do not shrink the caches, and README "Memory" adds up the new budget. Batch writes sort their rows by peer, then prefix. The benchmark also times the largest prefix handler batch, 250,000 updates. Model: opus-5-5 --- README.md | 17 +- internal/database/database.go | 122 +++++++++---- internal/database/database_internal_test.go | 181 ++++++++++++++++---- internal/database/slowquery.go | 4 +- 4 files changed, 252 insertions(+), 72 deletions(-) diff --git a/README.md b/README.md index 84a7bce..e5cd97a 100644 --- a/README.md +++ b/README.md @@ -117,7 +117,7 @@ routewatch/ - Signal handlers registered for graceful shutdown 2. **Initialization** (`internal/routewatch/app.go`) - - Database created with SQLite schema (WAL mode, 3GB cache) + - Database created with SQLite schema (WAL mode, page caches as in "Memory") - Message handlers registered with the streamer - HTTP server started on configured port @@ -215,8 +215,12 @@ data it tracks. The image sets ceilings that keep it inside a 5 GiB container. Budget: - Go heap: a 1.5 GiB soft limit (`GOMEMLIMIT=1536MiB`, set in the image). -- SQLite: at most 640 MiB of page cache across the connection pool (64 MiB per - connection, 10 connections) and a 1.5 GiB hard heap limit for the C library. +- SQLite: 1 GiB of page cache for the one connection every write uses, which + keeps the pages the route writes touch from one batch to the next, and 16 MiB + for each of the 9 reading connections, whose cache SQLite empties whenever + another connection has written: about 1.15 GiB in all. A 1.5 GiB soft heap + limit sits above the caches, so it does not shrink them, and a 2 GiB hard heap + limit caps the C library. - glibc allocator: the SQLite C library runs on glibc `malloc`, which frees page-cache chunks back to per-arena free lists rather than to the kernel, so process RSS tracks the high-water mark of those arenas, not SQLite's live @@ -226,6 +230,9 @@ Budget: database writes are already serialized. - About 0.2 GiB for everything else in the runtime. +The ceilings add up to 3.7 GiB: the Go heap's 1.5 GiB, SQLite's 2 GiB and the +0.2 GiB for the rest. + Run the container with a memory limit of 5 GiB and swap disabled: ```bash @@ -244,8 +251,8 @@ What happens at each limit: - Go soft limit: as the heap approaches `GOMEMLIMIT`, the runtime runs garbage collection more aggressively rather than growing further. -- SQLite: at a 1 GiB soft heap limit it recycles its page cache instead of - allocating more; at the 1.5 GiB hard heap limit a statement fails with an +- SQLite: at the 1.5 GiB soft heap limit it recycles its page cache instead of + allocating more; at the 2 GiB hard heap limit a statement fails with an out-of-memory error, and the handler logs it and drops that batch. The process keeps running. - Handler queues: each of the four handler queues holds at most 20,000 messages. diff --git a/internal/database/database.go b/internal/database/database.go index 08ac7d1..30c4722 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -1,18 +1,21 @@ package database import ( + "cmp" "context" "database/sql" _ "embed" "encoding/json" "errors" "fmt" + "maps" "math/rand/v2" "net" "net/netip" "os" "path/filepath" "runtime" + "slices" "sync" "time" @@ -51,20 +54,31 @@ const ( minLookupMaskLengthV6 = 16 ) -// SQLite memory tuning. cache_size and busy_timeout go in the DSN so every -// pooled connection gets them; the heap limits are process-wide and set once. +// SQLite memory tuning (README "Memory"). cache_size and busy_timeout go in the +// DSN so every pooled connection gets them; the connection kept for writing +// then sets its own cache size. The heap limits are process-wide and set once. const ( - // sqliteCacheSizeKiB is the per-connection page cache; negative means KiB. - // -65536 = 64 MiB, so at most 640 MiB across the 10-connection pool. - sqliteCacheSizeKiB = -65536 + // maxConns is the size of the connection pool: the connection kept for + // writing and 9 reading connections. + maxConns = 10 + // sqliteReadCacheSizeKiB is the page cache of each reading connection; + // negative means KiB. -16384 = 16 MiB, at most 144 MiB for the 9. A reading + // connection's cache is emptied whenever another connection has written + // since its last read, many times a minute, so a large one is wasted. + sqliteReadCacheSizeKiB = -16384 + // sqliteWriteCacheSizeKiB is the page cache of the connection kept for + // writing, -1048576 = 1 GiB. It keeps its pages from one write to the next, + // so the index pages the route writes touch stay in memory. + sqliteWriteCacheSizeKiB = -1048576 // sqliteBusyTimeoutMs is how long a connection waits on a locked database. sqliteBusyTimeoutMs = 5000 - // sqliteSoftHeapLimitBytes (1 GiB) makes SQLite recycle its cache rather - // than allocate once its C heap passes this size. - sqliteSoftHeapLimitBytes = 1073741824 - // sqliteHardHeapLimitBytes (1.5 GiB) fails a statement with SQLITE_NOMEM + // sqliteSoftHeapLimitBytes (1.5 GiB) makes SQLite recycle its cache rather + // than allocate once its C heap passes this size. It is above the caches + // together, so it does not shrink them. + sqliteSoftHeapLimitBytes = 1610612736 + // sqliteHardHeapLimitBytes (2 GiB) fails a statement with SQLITE_NOMEM // instead of growing the C heap without bound. - sqliteHardHeapLimitBytes = 1610612736 + sqliteHardHeapLimitBytes = 2147483648 ) // Settings of the ANALYZE that Analyze runs. @@ -105,7 +119,10 @@ var ( // Database manages the SQLite database connection and operations. type Database struct { - db *sql.DB + db *sql.DB + // writer is the connection kept for writing, taken from db when the + // database opens. Every write uses it while holding mu. + writer *sql.Conn logger *logger.Logger path string mu sync.Mutex @@ -143,7 +160,7 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) { "file:%s?_cache_size=%d&_synchronous=OFF&_busy_timeout=%d"+ "&_auto_vacuum=incremental&_journal_mode=WAL&_txlock=immediate", dbPath, - sqliteCacheSizeKiB, + sqliteReadCacheSizeKiB, sqliteBusyTimeoutMs, ) @@ -152,20 +169,34 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) { return nil, fmt.Errorf("failed to open database: %w", err) } - err = db.PingContext(context.Background()) + ctx := context.Background() + + err = db.PingContext(ctx) if err != nil { return nil, fmt.Errorf("failed to ping database: %w", err) } // Set connection pool parameters // Multiple connections allow concurrent reads while writes are serialized - const maxConns = 10 db.SetMaxOpenConns(maxConns) db.SetMaxIdleConns(maxConns) db.SetConnMaxLifetime(0) + writer, err := db.Conn(ctx) + if err != nil { + return nil, fmt.Errorf("failed to open the writing connection: %w", err) + } + + _, err = writer.ExecContext(ctx, + fmt.Sprintf("PRAGMA cache_size=%d", sqliteWriteCacheSizeKiB)) + if err != nil { + return nil, fmt.Errorf("failed to set the writing connection's cache: %w", + err) + } + database := &Database{ - db: db, logger: logger, path: dbPath, counts: &liveCounts{}, + db: db, writer: writer, logger: logger, path: dbPath, + counts: &liveCounts{}, } err = database.Initialize() @@ -187,10 +218,11 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) { // Initialize creates the database schema if it doesn't exist. func (d *Database) Initialize() error { // Set SQLite pragmas for performance. Per-connection settings (cache_size, - // synchronous, busy_timeout, journal_mode) live in the DSN; temp_store is - // left at its default so DISTINCT temp B-trees spill to disk instead of C - // heap. The heap limits below are process-wide, so setting them once here is - // enough for the whole pool. + // synchronous, busy_timeout, journal_mode) live in the DSN, and New sets the + // writing connection's cache_size; temp_store is left at its default so + // DISTINCT temp B-trees spill to disk instead of C heap. The heap limits + // below are process-wide, so setting them once here is enough for the whole + // pool. pragmas := []string{ "PRAGMA journal_mode=WAL", // Write-Ahead Logging fmt.Sprintf("PRAGMA soft_heap_limit=%d", sqliteSoftHeapLimitBytes), @@ -207,7 +239,7 @@ func (d *Database) Initialize() error { // Run WAL checkpoint on startup to consolidate any existing WAL var walPages, checkpointedPages, movedPages int - err := d.db.QueryRowContext(context.Background(), + err := d.writer.QueryRowContext(context.Background(), "PRAGMA wal_checkpoint(TRUNCATE)", ).Scan(&walPages, &checkpointedPages, &movedPages) if err != nil { @@ -228,9 +260,9 @@ func (d *Database) Initialize() error { return nil } -// Close closes the database connection. +// Close closes the writing connection and then the connection pool. func (d *Database) Close() error { - return d.db.Close() + return errors.Join(d.writer.Close(), d.db.Close()) } // A live-route upsert sets the origin AS of the peer's route for the prefix @@ -428,9 +460,18 @@ func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error { } defer d.rollback(tx) + // The rows are written in key order, peer then prefix, so consecutive writes + // land on the same index pages. The sort is stable, so of two updates of + // one route the later one is still written last. + sorted := slices.SortedStableFunc(slices.Values(routes), + func(a, b *LiveRoute) int { + return cmp.Or(cmp.Compare(a.PeerIP, b.PeerIP), + cmp.Compare(a.Prefix, b.Prefix)) + }) + var routesV4, routesV6 []*LiveRoute - for _, route := range routes { + for _, route := range sorted { if route.IPVersion == ipVersionV4 { routesV4 = append(routesV4, route) } else { @@ -574,9 +615,17 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error { } defer d.rollback(tx) + // The rows are deleted in key order, peer then prefix, as + // UpsertLiveRouteBatch writes them. + sorted := slices.SortedFunc(slices.Values(deletions), + func(a, b LiveRouteDeletion) int { + return cmp.Or(cmp.Compare(a.PeerIP, b.PeerIP), + cmp.Compare(a.Prefix, b.Prefix)) + }) + var deletionsV4, deletionsV6 []LiveRouteDeletion - for _, del := range deletions { + for _, del := range sorted { if del.IPVersion == ipVersionV4 { deletionsV4 = append(deletionsV4, del) } else { @@ -780,9 +829,10 @@ func upsertPrefixesInTable( var newPrefixes int - for prefix, timestamp := range prefixes { + // The rows are written in key order, as UpsertLiveRouteBatch writes routes. + for _, prefix := range slices.Sorted(maps.Keys(prefixes)) { inserted, err := upsertPrefix(ctx, updateStmt, insertStmt, prefix, - timestamp) + prefixes[prefix]) if err != nil { return 0, err } @@ -1411,25 +1461,25 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error { // The write lock is held, so no other writer can insert this key between the // update and the insert even though they are separate autocommit statements. - setOrigin, err := d.db.PrepareContext(ctx, originSQL) + setOrigin, err := d.writer.PrepareContext(ctx, originSQL) if err != nil { return fmt.Errorf("failed to prepare origin statement: %w", err) } defer func() { _ = setOrigin.Close() }() - upd, err := d.db.PrepareContext(ctx, updateSQL) + upd, err := d.writer.PrepareContext(ctx, updateSQL) if err != nil { return fmt.Errorf("failed to prepare update statement: %w", err) } defer func() { _ = upd.Close() }() - ins, err := d.db.PrepareContext(ctx, insertSQL) + ins, err := d.writer.PrepareContext(ctx, insertSQL) if err != nil { return fmt.Errorf("failed to prepare insert statement: %w", err) } defer func() { _ = ins.Close() }() - has, err := d.db.PrepareContext(ctx, lookupSQL) + has, err := d.writer.PrepareContext(ctx, lookupSQL) if err != nil { return fmt.Errorf("failed to prepare prefix lookup statement: %w", err) } @@ -2025,7 +2075,7 @@ func (d *Database) UpdateASNWHOIS( WHERE asn = ? ` - _, err := d.db.ExecContext(ctx, query, + _, err := d.writer.ExecContext(ctx, query, update.ASName, update.OrgName, update.OrgID, @@ -2191,7 +2241,7 @@ func (d *Database) Vacuum(ctx context.Context) error { d.lock("Vacuum") defer d.unlock() - rows, err := d.db.QueryContext(ctx, + rows, err := d.writer.QueryContext(ctx, fmt.Sprintf("PRAGMA incremental_vacuum(%d)", pagesToFree)) if err != nil { return fmt.Errorf("failed to run incremental vacuum: %w", err) @@ -2223,7 +2273,7 @@ func (d *Database) Analyze(ctx context.Context) error { // analysis_limit is a setting of the connection, so it is set in the same // call as PRAGMA optimize, which runs both statements on one connection. - _, err := d.db.ExecContext(ctx, fmt.Sprintf( + _, err := d.writer.ExecContext(ctx, fmt.Sprintf( "PRAGMA analysis_limit=%d; PRAGMA optimize=%#x", analysisLimit, optimizeAllTables)) if err != nil { @@ -2240,7 +2290,7 @@ func (d *Database) Checkpoint(ctx context.Context) error { d.lock("Checkpoint") defer d.unlock() - _, err := d.db.ExecContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)") + _, err := d.writer.ExecContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)") if err != nil { return fmt.Errorf("failed to checkpoint WAL: %w", err) } @@ -2296,9 +2346,9 @@ func (d *Database) unlock() { "held_by", lockedBy, "duration_ms", holdDuration.Milliseconds()) } -// beginTx starts a new transaction with logging +// beginTx starts a new transaction on the writing connection, with logging func (d *Database) beginTx() (*loggingTx, error) { - tx, err := d.db.BeginTx(context.Background(), nil) + tx, err := d.writer.BeginTx(context.Background(), nil) if err != nil { return nil, err } diff --git a/internal/database/database_internal_test.go b/internal/database/database_internal_test.go index 2be88da..639e9fb 100644 --- a/internal/database/database_internal_test.go +++ b/internal/database/database_internal_test.go @@ -9,6 +9,7 @@ import ( "math/rand/v2" "net/netip" "slices" + "strconv" "sync" "testing" "time" @@ -33,6 +34,9 @@ const vacuumTestRoutes = 2000 // once so each is a distinct SQLite connection that parsed the DSN. const heldConnections = 5 +// bytesPerKiB turns a cache_size in KiB into bytes. +const bytesPerKiB = 1024 + // testPeerIP is the peer every route in the IP lookup test is learned from. const testPeerIP = "192.0.2.254" @@ -336,8 +340,10 @@ func writeOriginRoutes(t *testing.T, db *Database, origins map[string]int) { // TestConnectionPoolPragmas holds several pooled connections open at once and // checks each one carries the per-connection settings from the DSN, plus the -// process-wide hard heap limit, and sees the new file with auto_vacuum -// incremental. +// process-wide heap limits, and sees the new file with auto_vacuum +// incremental. The connection kept for writing must have the large cache, and +// the soft heap limit must leave room for every connection's cache: SQLite +// shrinks the caches to stay under it. func TestConnectionPoolPragmas(t *testing.T) { t.Parallel() @@ -371,9 +377,9 @@ func TestConnectionPoolPragmas(t *testing.T) { for i, c := range conns { cacheSize := readPragma(ctx, t, c, i, "cache_size") - if cacheSize != sqliteCacheSizeKiB { + if cacheSize != sqliteReadCacheSizeKiB { t.Errorf("conn %d: cache_size = %d, want %d", - i, cacheSize, sqliteCacheSizeKiB) + i, cacheSize, sqliteReadCacheSizeKiB) } busyTimeout := readPragma(ctx, t, c, i, "busy_timeout") @@ -388,6 +394,12 @@ func TestConnectionPoolPragmas(t *testing.T) { i, tempStore, tempStoreMemory) } + softHeapLimit := readPragma(ctx, t, c, i, "soft_heap_limit") + if softHeapLimit != sqliteSoftHeapLimitBytes { + t.Errorf("conn %d: soft_heap_limit = %d, want %d", + i, softHeapLimit, sqliteSoftHeapLimitBytes) + } + hardHeapLimit := readPragma(ctx, t, c, i, "hard_heap_limit") if hardHeapLimit != sqliteHardHeapLimitBytes { t.Errorf("conn %d: hard_heap_limit = %d, want %d", @@ -400,6 +412,29 @@ func TestConnectionPoolPragmas(t *testing.T) { i, autoVacuum, autoVacuumIncremental) } } + + var writeCacheSize int64 + + err = db.writer.QueryRowContext(ctx, "PRAGMA cache_size"). + Scan(&writeCacheSize) + if err != nil { + t.Fatalf("failed to read the writing connection's cache_size: %v", err) + } + + if writeCacheSize != sqliteWriteCacheSizeKiB { + t.Errorf("writing connection: cache_size = %d, want %d", + writeCacheSize, sqliteWriteCacheSizeKiB) + } + + readCacheSize := readPragma(ctx, t, conns[0], 0, "cache_size") + softHeapLimit := readPragma(ctx, t, conns[0], 0, "soft_heap_limit") + cachesBytes := -bytesPerKiB * + (writeCacheSize + (maxConns-1)*readCacheSize) + + if softHeapLimit <= cachesBytes { + t.Errorf("soft_heap_limit = %d, not above the %d bytes of the caches", + softHeapLimit, cachesBytes) + } } // readPragma reads the integer value of PRAGMA name on connection i, c. @@ -1229,6 +1264,85 @@ func TestUpdatePrefixesBatchKeepsFirstSeen(t *testing.T) { } } +// TestUpsertLiveRouteBatchWritesInKeyOrder writes new routes of two peers out +// of key order, the last one a second announcement of the first, and checks +// that the rows were inserted in key order, peer then prefix (each new row's id +// is one above the last), and that the route announced twice carries the later +// announcement. +func TestUpsertLiveRouteBatchWritesInKeyOrder(t *testing.T) { + t.Parallel() + + db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + defer func() { _ = db.Close() }() + + const firstOrigin, laterOrigin = 64500, 64501 + + type row struct { + peer, prefix string + origin int + } + + ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + + var routes []*LiveRoute + + for _, announced := range []row{ + {secondV4PeerIP, "198.51.100.0/24", firstOrigin}, + {testV4PeerIP, "203.0.113.0/24", firstOrigin}, + {secondV4PeerIP, "192.0.2.0/24", firstOrigin}, + {testV4PeerIP, "198.51.100.0/24", firstOrigin}, + {secondV4PeerIP, "198.51.100.0/24", laterOrigin}, + } { + route := mkV4Route(t, announced.prefix, announced.origin, ts) + route.PeerIP = announced.peer + routes = append(routes, route) + } + + err = db.UpsertLiveRouteBatch(routes) + if err != nil { + t.Fatalf("UpsertLiveRouteBatch: %v", err) + } + + rows, err := db.db.QueryContext(t.Context(), + "SELECT peer_ip, prefix, origin_asn FROM live_routes_v4 ORDER BY id") + if err != nil { + t.Fatalf("reading the routes: %v", err) + } + defer func() { _ = rows.Close() }() + + var stored []row + + for rows.Next() { + var r row + + err := rows.Scan(&r.peer, &r.prefix, &r.origin) + if err != nil { + t.Fatalf("reading a route: %v", err) + } + + stored = append(stored, r) + } + + err = rows.Err() + if err != nil { + t.Fatalf("reading the routes: %v", err) + } + + want := []row{ + {testV4PeerIP, "198.51.100.0/24", firstOrigin}, + {testV4PeerIP, "203.0.113.0/24", firstOrigin}, + {secondV4PeerIP, "192.0.2.0/24", firstOrigin}, + {secondV4PeerIP, "198.51.100.0/24", laterOrigin}, + } + + if !slices.Equal(stored, want) { + t.Errorf("routes in insertion order = %+v, want %+v", stored, want) + } +} + // Parameters for BenchmarkPrefixBatchOnFiveMillionRoutes. const ( // benchPeers and benchPrefixes size the table the batches are written to: @@ -1238,6 +1352,10 @@ const ( // benchBatchSize is how many updates each batch holds, as many as a full // prefix handler batch (prefixBatchSize in internal/routewatch). benchBatchSize = 25000 + // benchMaxBatchSize is the largest batch the prefix handler writes, the + // one it holds while a slow write runs (prefixBatchMaxSize in + // internal/routewatch). + benchMaxBatchSize = 250000 // benchWithdrawalEvery makes every fifth update of a timed batch a // withdrawal; the others are announcements. benchWithdrawalEvery = 5 @@ -1250,13 +1368,14 @@ const ( ) // BenchmarkPrefixBatchOnFiveMillionRoutes times the writes the prefix handler -// makes for one full batch, 25,000 announcements and withdrawals of routes -// picked at random, to a database that already holds 5 million live routes: -// 500,000 IPv4 prefixes, each from 10 peers. It builds that database first, -// one batch at a time, which takes some minutes and a few GB of disk under -// TMPDIR. It is run by hand, not by make test, for example in the test image: +// makes for one batch of announcements and withdrawals of routes picked at +// random, to a database that already holds 5 million live routes: 500,000 +// IPv4 prefixes, each from 10 peers. It times a full batch of 25,000 updates, +// then the largest batch, 250,000. It builds that database first, one batch at +// a time, which takes some minutes and a few GB of disk under TMPDIR. It is +// run by hand, not by make test, for example in the test image: // -// go test -run '^$' -bench FiveMillion -benchtime 20x ./internal/database/ +// go test -run '^$' -bench FiveMillion -benchtime 10x ./internal/database/ func BenchmarkPrefixBatchOnFiveMillionRoutes(b *testing.B) { db, err := New(&config.Config{StateDir: b.TempDir()}, logger.New()) if err != nil { @@ -1280,28 +1399,32 @@ func BenchmarkPrefixBatchOnFiveMillionRoutes(b *testing.B) { } } - for b.Loop() { - ts = ts.Add(time.Second) + for _, batchSize := range []int{benchBatchSize, benchMaxBatchSize} { + b.Run(strconv.Itoa(batchSize), func(b *testing.B) { + for b.Loop() { + ts = ts.Add(time.Second) - var ( - routes []*LiveRoute - withdrawals []LiveRouteDeletion - ) + var ( + routes []*LiveRoute + withdrawals []LiveRouteDeletion + ) - for i := range benchBatchSize { - route := benchRoute(rng, rng.IntN(benchPrefixes), - rng.IntN(benchPeers), ts) - if i%benchWithdrawalEvery == 0 { - withdrawals = append(withdrawals, LiveRouteDeletion{ - Prefix: route.Prefix, PeerIP: route.PeerIP, - IPVersion: ipVersionV4, - }) - } else { - routes = append(routes, route) + for i := range batchSize { + route := benchRoute(rng, rng.IntN(benchPrefixes), + rng.IntN(benchPeers), ts) + if i%benchWithdrawalEvery == 0 { + withdrawals = append(withdrawals, LiveRouteDeletion{ + Prefix: route.Prefix, PeerIP: route.PeerIP, + IPVersion: ipVersionV4, + }) + } else { + routes = append(routes, route) + } + } + + writeBenchBatch(b, db, routes, withdrawals, ts) } - } - - writeBenchBatch(b, db, routes, withdrawals, ts) + }) } } diff --git a/internal/database/slowquery.go b/internal/database/slowquery.go index d027211..13a0ac8 100644 --- a/internal/database/slowquery.go +++ b/internal/database/slowquery.go @@ -18,12 +18,12 @@ func logSlowQuery(logger *logger.Logger, query string, start time.Time) { } } -// exec wraps Exec with slow query logging +// exec runs a statement on the writing connection, with slow query logging func (d *Database) exec(query string, args ...any) error { start := time.Now() defer logSlowQuery(d.logger, query, start) - _, err := d.db.ExecContext(context.Background(), query, args...) + _, err := d.writer.ExecContext(context.Background(), query, args...) return err }