package index
import (
"context"
"sync"
"testing"
)
// A transaction that reads a row and then writes it must commit even
// when other writers land in between; with deferred transactions the
// late upgrade to a write lock failed with SQLITE_BUSY_SNAPSHOT. It
// goes through ix.w because every write transaction in the package does.
func TestConcurrentReadThenWriteTransactionsAllCommit(t *testing.T) {
ix := openIndex(t)
if _, err := ix.w.Exec(`CREATE TABLE counter (n INTEGER NOT NULL); INSERT INTO counter VALUES (0)`); err != nil {
t.Fatal(err)
}
const workers, rounds = 8, 25
var wg sync.WaitGroup
errs := make(chan error, workers*rounds)
for w := 0; w < workers; w++ {
wg.Add(1)
go func() {
defer wg.Done()
for r := 0; r < rounds; r++ {
errs <- func() error {
tx, err := ix.w.Begin()
if err != nil {
return err
}
defer tx.Rollback()
var n int
if err := tx.QueryRow(`SELECT n FROM counter`).Scan(&n); err != nil {
return err
}
if _, err := tx.Exec(`UPDATE counter SET n = ?`, n+1); err != nil {
return err
}
return tx.Commit()
}()
}
}()
}
wg.Wait()
close(errs)
failed := 0
for err := range errs {
if err == nil {
continue
}
if failed == 0 {
t.Errorf("FAIL: concurrent read-then-write transaction: %v", err)
}
failed++
}
if failed > 0 {
t.Fatalf("FAIL: %d of %d transactions failed", failed, workers*rounds)
}
var n int
if err := ix.db.QueryRow(`SELECT n FROM counter`).Scan(&n); err != nil {
t.Fatal(err)
}
if n != workers*rounds {
t.Fatalf("FAIL: counter is %d after %d increments; an update was lost", n, workers*rounds)
}
t.Logf("ok: %d concurrent read-then-write transactions committed, counter=%d", workers*rounds, n)
}
// Pragmas that live on a connection rather than in the file must be on
// every connection the pool opens, not only the first one.
func TestEveryPooledConnectionCarriesThePragmas(t *testing.T) {
ix := openIndex(t)
ctx := context.Background()
first, err := ix.db.Conn(ctx)
if err != nil {
t.Fatal(err)
}
defer first.Close()
// Holding the first connection forces the pool to open another.
second, err := ix.db.Conn(ctx)
if err != nil {
t.Fatal(err)
}
defer second.Close()
// mmap_size is not checked: this driver's SQLite is compiled with
// SQLITE_MAX_MMAP_SIZE=0, so it reads back as 0 on every connection.
var busy, sync, cache, temp int
var journal string
for _, p := range []struct {
name string
dst any
}{
{"busy_timeout", &busy}, {"journal_mode", &journal}, {"synchronous", &sync},
{"cache_size", &cache}, {"temp_store", &temp},
} {
if err := second.QueryRowContext(ctx, `PRAGMA `+p.name).Scan(p.dst); err != nil {
t.Fatalf("FAIL: PRAGMA %s: %v", p.name, err)
}
}
// synchronous 1 is NORMAL; temp_store 2 is MEMORY.
if busy != 5000 || journal != "wal" || sync != 1 || cache != -16000 || temp != 2 {
t.Fatalf("FAIL: second connection has busy_timeout=%d journal_mode=%s synchronous=%d cache_size=%d temp_store=%d",
busy, journal, sync, cache, temp)
}
t.Logf("ok: second connection has busy_timeout=%d journal_mode=%s synchronous=%d cache_size=%d temp_store=%d",
busy, journal, sync, cache, temp)
}
// A transaction begun on the reader pool must stay a reader: it may not
// take the write lock at BEGIN, or it would stall the writer queue and
// fail a writer that finds the lock held past the busy timeout.
func TestAReaderPoolTransactionDoesNotBlockTheWriter(t *testing.T) {
ix := openIndex(t)
if _, err := ix.w.Exec(`CREATE TABLE counter (n INTEGER NOT NULL); INSERT INTO counter VALUES (0)`); err != nil {
t.Fatal(err)
}
rtx, err := ix.db.Begin()
if err != nil {
t.Fatal(err)
}
defer rtx.Rollback()
var n int
if err := rtx.QueryRow(`SELECT n FROM counter`).Scan(&n); err != nil {
t.Fatal(err)
}
err = func() error {
tx, err := ix.w.Begin()
if err != nil {
return err
}
defer tx.Rollback()
if _, err := tx.Exec(`UPDATE counter SET n = n + 1`); err != nil {
return err
}
return tx.Commit()
}()
if err != nil {
t.Fatalf("FAIL: a writer transaction failed while a reader-pool transaction was open: %v", err)
}
t.Logf("ok: writer committed while a reader-pool transaction held a snapshot (n=%d)", n)
}