package index
import (
"context"
"fmt"
"path/filepath"
"testing"
"time"
"github.com/ncruces/go-sqlite3/driver"
)
// attachment.head runs on every ticket page that shows a file, and it
// records what it learned in the index. That record must not make the
// page wait for the writer: a mirror fold of a large tracker holds it
// for seconds, and before this the page sat behind the fold once per
// attachment card. The verdict is readable at once, and lands in the
// table when the writer frees up.
func TestAHeadMemoNeverWaitsForTheWriter(t *testing.T) {
ix := openIndex(t)
tx, err := ix.w.Begin()
if err != nil {
t.Fatal(err)
}
const held = 2 * time.Second
released := make(chan struct{})
go func() {
time.Sleep(held)
tx.Rollback()
close(released)
}()
want := AttachmentHead{Size: 9, Kind: "text", Lines: 1, Text: "nine byte"}
start := time.Now()
ix.RememberAttachmentHead("blob1", want)
if el := time.Since(start); el > held/4 {
t.Fatalf("FAIL: remembering a verdict waited %v on a held writer", el.Round(time.Millisecond))
}
got, ok, err := ix.AttachmentHeadFor("blob1")
if err != nil || !ok || got != want {
t.Fatalf("FAIL: a verdict remembered behind a held writer read back ok=%v err=%v %+v", ok, err, got)
}
if el := time.Since(start); el > held/4 {
t.Fatalf("FAIL: reading the verdict back waited %v on a held writer", el.Round(time.Millisecond))
}
<-released
deadline := time.Now().Add(5 * time.Second)
for {
var n int
if err := ix.db.QueryRow(`SELECT count(*) FROM attachment_head WHERE src = 'blob1'`).Scan(&n); err != nil {
t.Fatal(err)
}
if n == 1 {
break
}
if time.Now().After(deadline) {
t.Fatal("FAIL: the verdict never reached the table once the writer was free")
}
time.Sleep(10 * time.Millisecond)
}
t.Log("ok: a head verdict is remembered without waiting on the writer, and written once it is free")
}
// A full pending set turns a new verdict away without waiting, and
// keeps every verdict it already holds: a crawler during a long fold
// costs later readers a memo miss, never a wait or a lost verdict.
func TestAFullHeadMemoDropsNewVerdictsWithoutWaiting(t *testing.T) {
ix := openIndex(t)
tx, err := ix.w.Begin()
if err != nil {
t.Fatal(err)
}
defer tx.Rollback()
start := time.Now()
for i := 0; i < maxPendingHeads; i++ {
ix.RememberAttachmentHead(fmt.Sprintf("blob%d", i), AttachmentHead{Size: i, Kind: "other"})
}
ix.RememberAttachmentHead("one-too-many", AttachmentHead{Size: 1, Kind: "other"})
if el := time.Since(start); el > time.Second {
t.Fatalf("FAIL: filling the pending set waited %v on a held writer", el.Round(time.Millisecond))
}
if _, ok, _ := ix.AttachmentHeadFor("one-too-many"); ok {
t.Fatal("FAIL: a verdict past the bound was kept")
}
for _, i := range []int{0, maxPendingHeads / 2, maxPendingHeads - 1} {
if h, ok, err := ix.AttachmentHeadFor(fmt.Sprintf("blob%d", i)); err != nil || !ok || h.Size != i {
t.Fatalf("FAIL: pending verdict %d lost: ok=%v err=%v %+v", i, ok, err, h)
}
}
t.Log("ok: a full pending set drops new verdicts at once and keeps the ones it holds")
}
// A Reset that races a flush leaves the memo empty: the flush either
// commits before the drop or sees the generation move and writes
// nothing. Meant to be run under -race too.
func TestAResetRacingAHeadFlushLeavesTheMemoEmpty(t *testing.T) {
ix := openIndex(t)
for round := 0; round < 20; round++ {
for i := 0; i < 50; i++ {
ix.RememberAttachmentHead(fmt.Sprintf("r%d-%d", round, i), AttachmentHead{Size: i, Kind: "other"})
}
if err := ix.Reset(); err != nil {
t.Fatal(err)
}
deadline := time.Now().Add(5 * time.Second)
for {
ix.headMu.Lock()
idle := !ix.headFlusher
ix.headMu.Unlock()
if idle {
break
}
if time.Now().After(deadline) {
t.Fatal("FAIL: the flush never finished")
}
time.Sleep(time.Millisecond)
}
var n int
if err := ix.db.QueryRow(`SELECT count(*) FROM attachment_head`).Scan(&n); err != nil {
t.Fatal(err)
}
if n != 0 {
t.Fatalf("FAIL: round %d: %d verdicts survived a Reset that raced their flush", round, n)
}
}
t.Log("ok: a Reset racing a head flush leaves the memo empty")
}
// Close stops the flush even while verdicts keep arriving: the stop
// budget is seconds, and a flush that looped for as long as traffic
// lasted would spend it. Pending verdicts are dropped, which is a miss.
func TestCloseStopsTheHeadFlushUnderTraffic(t *testing.T) {
ix, err := Open(filepath.Join(t.TempDir(), "i.db"))
if err != nil {
t.Fatal(err)
}
stop := make(chan struct{})
done := make(chan struct{})
go func() {
defer close(done)
for i := 0; ; i++ {
select {
case <-stop:
return
default:
}
ix.RememberAttachmentHead(fmt.Sprintf("t%d", i), AttachmentHead{Size: i, Kind: "other"})
}
}()
time.Sleep(100 * time.Millisecond)
start := time.Now()
closed := make(chan error, 1)
go func() { closed <- ix.Close() }()
select {
case err := <-closed:
if err != nil {
t.Fatal(err)
}
case <-time.After(2 * time.Second):
close(stop)
<-done
<-closed
t.Fatal("FAIL: Close waited on a flush that kept looping while verdicts arrived")
}
close(stop)
<-done
t.Logf("ok: Close returned in %v with verdicts still arriving", time.Since(start).Round(time.Millisecond))
}
// A flush whose writer cannot be had — another handle holds the write
// lock past the busy timeout — must not retry row by row: each retry
// would wait the timeout again, and Close waits behind all of them.
func TestCloseDoesNotWaitOnRetriesOfAnUnavailableWriter(t *testing.T) {
path := filepath.Join(t.TempDir(), "i.db")
ix, err := Open(path)
if err != nil {
t.Fatal(err)
}
if _, err := ix.w.Exec(`PRAGMA busy_timeout = 100`); err != nil {
t.Fatal(err)
}
other, err := driver.Open("file:" + path + "?_pragma=busy_timeout(0)")
if err != nil {
t.Fatal(err)
}
defer other.Close()
conn, err := other.Conn(context.Background())
if err != nil {
t.Fatal(err)
}
defer conn.Close()
if _, err := conn.ExecContext(context.Background(), `BEGIN IMMEDIATE`); err != nil {
t.Fatal(err)
}
defer conn.ExecContext(context.Background(), `ROLLBACK`)
for i := 0; i < 50; i++ {
ix.RememberAttachmentHead(fmt.Sprintf("b%d", i), AttachmentHead{Size: i, Kind: "other"})
}
// Past the first batch's failure, into what used to be the retries.
time.Sleep(250 * time.Millisecond)
start := time.Now()
if err := ix.Close(); err != nil {
t.Fatal(err)
}
if el := time.Since(start); el > 500*time.Millisecond {
t.Fatalf("FAIL: Close waited %v behind retries against an unavailable writer", el.Round(time.Millisecond))
}
t.Log("ok: an unavailable writer is not retried row by row, and Close does not wait on it")
}