package index
import (
"context"
"database/sql"
"fmt"
"path/filepath"
"strings"
"testing"
"time"
)
// shrinkRecycling makes every pool opened afterwards recycle at once.
func shrinkRecycling(t *testing.T, idle, life time.Duration) {
t.Helper()
oi, ol := connMaxIdleTime, connMaxLifetime
connMaxIdleTime, connMaxLifetime = idle, life
t.Cleanup(func() { connMaxIdleTime, connMaxLifetime = oi, ol })
}
type settings struct {
cache, temp, busy, fk int
heap int64
queryOnly int
journal, sync string
fts int
}
func tempStoreCode(s string) int {
if s == "FILE" {
return 1
}
return 2
}
func readSettings(t *testing.T, db *sql.DB) settings {
t.Helper()
ctx := context.Background()
c, err := db.Conn(ctx)
if err != nil {
t.Fatal(err)
}
defer c.Close()
var s settings
for q, dst := range map[string]any{
`PRAGMA cache_size`: &s.cache,
`PRAGMA temp_store`: &s.temp,
`PRAGMA busy_timeout`: &s.busy,
`PRAGMA foreign_keys`: &s.fk,
`PRAGMA query_only`: &s.queryOnly,
`PRAGMA journal_mode`: &s.journal,
`PRAGMA synchronous`: &s.sync,
`PRAGMA soft_heap_limit`: &s.heap,
} {
if err := c.QueryRowContext(ctx, q).Scan(dst); err != nil {
t.Fatalf("%s: %v", q, err)
}
}
// fts5 is registered per connection; a connection without it
// cannot even prepare a MATCH against the search index.
var n int
if err := c.QueryRowContext(ctx, `SELECT count(*) FROM search_fts WHERE search_fts MATCH 'x'`).Scan(&n); err != nil {
t.Errorf("FAIL: fts5 MATCH on a connection: %v", err)
}
s.fts = 1
return s
}
func TestPoolsCarryRecyclingPolicy(t *testing.T) {
ix, err := Open(filepath.Join(t.TempDir(), "index.db"))
if err != nil {
t.Fatal(err)
}
defer ix.Close()
ctx := context.Background()
// Hold three reader connections at once, release them: the pool
// may keep at most one idle.
var conns []*sql.Conn
for range 3 {
c, err := ix.db.Conn(ctx)
if err != nil {
t.Fatal(err)
}
conns = append(conns, c)
}
for _, c := range conns {
c.Close()
}
if got := ix.db.Stats().Idle; got > 1 {
t.Errorf("FAIL: reader pool keeps %d idle connections, want at most 1", got)
}
if got := ix.w.Stats().MaxOpenConnections; got != 1 {
t.Errorf("FAIL: writer pool max open = %d, want 1", got)
}
// Non-zero lifetime and idle time show up as closures when they are tiny.
shrinkRecycling(t, time.Nanosecond, time.Nanosecond)
ix2, err := Open(filepath.Join(t.TempDir(), "index.db"))
if err != nil {
t.Fatal(err)
}
defer ix2.Close()
for name, db := range map[string]*sql.DB{"reader": ix2.db, "writer": ix2.w} {
for range 3 {
if _, err := db.Exec(`SELECT 1`); err != nil {
t.Fatal(err)
}
time.Sleep(2 * time.Millisecond)
}
st := db.Stats()
if st.MaxLifetimeClosed == 0 && st.MaxIdleTimeClosed == 0 {
t.Errorf("FAIL: %s pool closed nothing for lifetime or idle time: policy not applied", name)
}
}
if !t.Failed() {
t.Log("ok")
}
}
func TestRecycledConnectionIsFullySetUp(t *testing.T) {
ix, err := Open(filepath.Join(t.TempDir(), "index.db"))
if err != nil {
t.Fatal(err)
}
defer ix.Close()
pools := map[string]*sql.DB{"reader": ix.db, "writer": ix.w}
first := map[string]settings{}
for n, db := range pools {
first[n] = readSettings(t, db)
}
if first["reader"].queryOnly != 0 || first["reader"].temp != tempStoreCode(readerTempStore) || first["writer"].temp != 1 {
t.Errorf("FAIL: unexpected first-connection settings %+v", first)
}
// Negative cache_size is KiB, per pool, and it must be the same on
// a connection the pool opened later.
if first["reader"].cache != -readerCacheKiB || first["writer"].cache != -writerCacheKiB {
t.Errorf("FAIL: cache_size reader %d writer %d, want %d and %d",
first["reader"].cache, first["writer"].cache, -readerCacheKiB, -writerCacheKiB)
}
for n, db := range pools {
before := db.Stats().MaxLifetimeClosed
db.SetConnMaxLifetime(time.Nanosecond)
time.Sleep(5 * time.Millisecond)
// Each use returns the connection expired, so the pool closes it
// and the next call opens a brand new one.
for range 3 {
readSettings(t, db)
}
if db.Stats().MaxLifetimeClosed <= before {
t.Errorf("FAIL: %s pool did not recycle any connection", n)
}
db.SetConnMaxLifetime(connMaxLifetime)
if got := readSettings(t, db); got != first[n] {
t.Errorf("FAIL: %s recycled connection differs: %+v, first %+v", n, got, first[n])
}
}
if !t.Failed() {
t.Log("ok")
}
}
func TestTransactionAcrossLifetimeBoundary(t *testing.T) {
ix, err := Open(filepath.Join(t.TempDir(), "index.db"))
if err != nil {
t.Fatal(err)
}
defer ix.Close()
ix.w.SetConnMaxLifetime(time.Nanosecond)
tx, err := ix.w.Begin()
if err != nil {
t.Fatal(err)
}
time.Sleep(5 * time.Millisecond)
if _, err := tx.Exec(`INSERT INTO meta (key, value) VALUES ('k', 'v')`); err != nil {
t.Errorf("FAIL: write after lifetime expiry inside a transaction: %v", err)
}
if _, err := tx.Exec(`INSERT INTO meta (key, value) VALUES ('k2', 'v')`); err != nil {
t.Errorf("FAIL: second write: %v", err)
}
if err := tx.Commit(); err != nil {
t.Errorf("FAIL: commit: %v", err)
}
var n int
if err := ix.w.QueryRow(`SELECT count(*) FROM meta WHERE key IN ('k','k2')`).Scan(&n); err != nil || n != 2 {
t.Errorf("FAIL: rows after commit = %d, err %v", n, err)
}
if !t.Failed() {
t.Log("ok")
}
}
func TestIdleConnectionsAreClosedAndReopened(t *testing.T) {
shrinkRecycling(t, 20*time.Millisecond, time.Hour)
ix, err := Open(filepath.Join(t.TempDir(), "index.db"))
if err != nil {
t.Fatal(err)
}
defer ix.Close()
if err := ix.db.QueryRow(`SELECT 1`).Scan(new(int)); err != nil {
t.Fatal(err)
}
deadline := time.Now().Add(5 * time.Second)
for ix.db.Stats().OpenConnections != 0 && time.Now().Before(deadline) {
time.Sleep(10 * time.Millisecond)
}
if got := ix.db.Stats().OpenConnections; got != 0 {
t.Errorf("FAIL: %d connections still open after idle time", got)
}
var n int
if err := ix.db.QueryRow(`SELECT count(*) FROM search_fts WHERE search_fts MATCH 'x'`).Scan(&n); err != nil {
t.Errorf("FAIL: query after the pool emptied: %v", err)
}
if !t.Failed() {
t.Log("ok")
}
}
// A pool's memory ceiling reaches every connection it opens, and the two
// pools carry their own.
func TestPoolsCarryTheirOwnMemoryCeiling(t *testing.T) {
ix := openIndex(t)
for n, c := range map[string]struct {
db *sql.DB
want int64
}{"reader": {ix.db, readerMaxMemory}, "writer": {ix.w, writerMaxMemory}} {
got := readSettings(t, c.db).heap
if got <= c.want/2 || got > c.want {
t.Errorf("FAIL: %s soft heap limit %d, want just under %d", n, got, c.want)
}
}
}
func TestReaderPoolBoundsOpenConnections(t *testing.T) {
ix := openIndex(t)
if got := ix.db.Stats().MaxOpenConnections; got != readerMaxConns {
t.Errorf("FAIL: reader pool max open = %d, want %d", got, readerMaxConns)
}
if readerMaxConns == 0 {
return
}
// One more concurrent reader than the bound waits for a connection
// instead of opening a fifth.
ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
defer cancel()
var held []*sql.Conn
for range readerMaxConns {
c, err := ix.db.Conn(context.Background())
if err != nil {
t.Fatal(err)
}
held = append(held, c)
}
if c, err := ix.db.Conn(ctx); err == nil {
c.Close()
t.Errorf("FAIL: reader pool handed out connection %d past its bound of %d", readerMaxConns+1, readerMaxConns)
}
for _, c := range held {
c.Close()
}
if got := ix.db.Stats().OpenConnections; got > readerMaxConns {
t.Errorf("FAIL: %d reader connections open, bound %d", got, readerMaxConns)
}
}
// A reader that needs more memory than its ceiling fails that query, as
// the driver's out-of-memory panic (the web handler turns it into one
// logged 500); the next query is answered and the writer is untouched.
func TestReaderQueryPastItsCeilingFailsAloneAndLeavesTheNextQueryAnswered(t *testing.T) {
old := readerMaxMemory
readerMaxMemory = 16 << 20
t.Cleanup(func() { readerMaxMemory = old })
oldTemp := readerTempStore
readerTempStore = "MEMORY"
t.Cleanup(func() { readerTempStore = oldTemp })
ix := openIndex(t)
if _, err := ix.w.Exec(`CREATE TABLE ceiling(b TEXT)`); err != nil {
t.Fatal(err)
}
if _, err := ix.w.Exec(`WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL SELECT x+1 FROM c WHERE x<100000)
INSERT INTO ceiling SELECT hex(randomblob(200)) FROM c`); err != nil {
t.Fatal(err)
}
var recovered any
func() {
defer func() { recovered = recover() }()
ix.db.QueryRow(`SELECT count(*) FROM (SELECT b FROM ceiling ORDER BY b DESC)`).Scan(new(int))
}()
if !strings.Contains(fmt.Sprint(recovered), "out of memory") {
t.Errorf("FAIL: a query past the ceiling: recovered %v, want the out-of-memory panic", recovered)
}
// The driver's panic escapes database/sql's own cleanup, so the
// failed query's connection stays checked out (InUse 1) for good:
// the reason the reader pool is not bounded. The pool still opens
// another connection for the next query.
var n int
if err := ix.db.QueryRow(`SELECT count(*) FROM ceiling`).Scan(&n); err != nil || n != 100000 {
t.Errorf("FAIL: next reader query: n=%d err=%v", n, err)
}
if err := ix.w.QueryRow(`SELECT count(*) FROM ceiling`).Scan(&n); err != nil || n != 100000 {
t.Errorf("FAIL: writer after reader failure: n=%d err=%v", n, err)
}
}