diff --git a/dbkit/sqlite/CHANGELOG.md b/dbkit/sqlite/CHANGELOG.md index 94ed152..990da53 100644 --- a/dbkit/sqlite/CHANGELOG.md +++ b/dbkit/sqlite/CHANGELOG.md @@ -16,9 +16,17 @@ Releases of this module are tagged `dbkit/sqlite/vX.Y.Z`. ### Changed +- **Breaking.** `Options.BusyPoll` is required, how often a waiting connection tries a lock again, from 1ms up to `BusyTimeout`. Set it from a setting of your own. Keep it small, because each opening try can block for about one `BusyPoll`. +- Each connection waits for a lock through dbkit's own busy handler, so `PRAGMA busy_timeout` reads 0, and setting that pragma brings back a wait that ignores the context. +- `sql.Conn.Raw` hands out a dbkit wrapper, so the driver's methods outside `database/sql`'s interfaces, such as `Serialize`, `NewBackup` and the hooks, are no longer reachable through it. - The module needs Go 1.27.2. - Requires `dbkit` 0.3.0. +### Fixed + +- A statement, transaction or snapshot checkpoint waiting for a lock stops once its context ends. A prepared statement's run stops within one `BusyPoll` of its context ending. A `BeginTx` error after its context ends also matches the context error. +- A first connection to a file that another process is switching to WAL tries again every `BusyPoll` until `BusyTimeout` passes or its context ends, instead of failing busy at once. + ## [0.1.0] - 2026-10-08 ### Added diff --git a/dbkit/sqlite/busy.go b/dbkit/sqlite/busy.go new file mode 100644 index 0000000..aca91d6 --- /dev/null +++ b/dbkit/sqlite/busy.go @@ -0,0 +1,288 @@ +// SPDX-License-Identifier: Apache-2.0 + +package sqlite + +import ( + "context" + "database/sql/driver" + "fmt" + "io" + "reflect" + "sync" + "sync/atomic" + "time" + "unsafe" + + "modernc.org/libc" + sqlite3 "modernc.org/sqlite/lib" +) + +// busyWait is how one connection waits for a lock. +type busyWait struct { + // db is the connection's SQLite handle. + db uintptr + // timeout is BusyTimeout. + timeout time.Duration + // poll is BusyPoll. + poll time.Duration + // began is when the current wait outside a call began. + began time.Time + // span is the call of busyConn running on the connection, or the query whose rows are open, nil between calls. + span atomic.Pointer[callSpan] +} + +// callSpan is one call of busyConn on a connection. +type callSpan struct { + // done is the Done channel of the call's context, nil for a context that never ends. + done <-chan struct{} + // began is when the call's current wait began, zero until the call first waits. + began time.Time +} + +var ( + // busyWaits maps the id SQLite hands busyHandler to the wait of its connection. + busyWaits sync.Map + // lastWaitID is the id of the newest wait. + lastWaitID atomic.Uintptr +) + +// busyHandler is SQLite's busy callback for a dbkit connection, 1 to try the lock again and 0 to answer busy. +func busyHandler(tls *libc.TLS, id uintptr, count int32) int32 { + v, ok := busyWaits.Load(id) + if !ok { + return 0 + } + w := v.(*busyWait) + if sqlite3.Xsqlite3_is_interrupted(tls, w.db) != 0 || !w.retry(count) { + return 0 + } + return 1 +} + +// retry waits one busy poll at most and reports whether to try the lock again. +func (w *busyWait) retry(count int32) bool { + began, done := w.current() + if count == 0 || began.IsZero() { + *began = time.Now() + } + left := w.timeout - time.Since(*began) + if left <= 0 { + return false + } + poll := time.NewTimer(min(w.poll, left)) + defer poll.Stop() + select { + case <-done: + return false + case <-poll.C: + return true + } +} + +// current returns the start of the running wait and the Done channel of the running call, nil outside a call. +func (w *busyWait) current() (*time.Time, <-chan struct{}) { + if s := w.span.Load(); s != nil { + return &s.began, s.done + } + return &w.began, nil +} + +// busyHandlerFP is busyHandler as the function pointer SQLite calls. +var busyHandlerFP = *(*uintptr)(unsafe.Pointer(&struct { + f func(*libc.TLS, uintptr, int32) int32 +}{busyHandler})) + +// handleOf returns the SQLite handle of a connection the driver opened, read from the driver's private field db. +func handleOf(conn driver.Conn) (uintptr, error) { + v := reflect.ValueOf(conn) + if v.Kind() == reflect.Pointer { + v = v.Elem() + } + if v.Kind() != reflect.Struct { + return 0, fmt.Errorf("dbkit: the SQLite driver connection %T is not a struct, so a busy wait cannot end "+ + "with its context", conn) + } + field, ok := v.Type().FieldByName("db") + if !ok || len(field.Index) != 1 { + return 0, fmt.Errorf("dbkit: the SQLite driver connection %T holds no handle in its field db, so a busy wait "+ + "cannot end with its context", conn) + } + f := v.Field(field.Index[0]) + if f.Kind() != reflect.Uintptr || f.Uint() == 0 { + return 0, fmt.Errorf("dbkit: the SQLite driver connection %T holds no handle in its field db, so a busy wait "+ + "cannot end with its context", conn) + } + return uintptr(f.Uint()), nil +} + +// driverConn is every interface the driver's connection offers to database/sql. +type driverConn interface { + driver.Conn + driver.ConnBeginTx + driver.ConnPrepareContext + driver.ExecerContext + driver.QueryerContext + driver.Pinger + driver.SessionResetter + driver.Validator +} + +// waitOn points the busy callback of conn at busyHandler and returns conn wrapped so its waits end with their context. +func (c *connector) waitOn(conn driver.Conn) (driver.Conn, error) { + db, err := handleOf(conn) + if err != nil { + return nil, err + } + full, ok := conn.(driverConn) + if !ok { + return nil, fmt.Errorf("dbkit: the SQLite driver connection %T lacks an interface database/sql uses", conn) + } + id := lastWaitID.Add(1) + wait := &busyWait{db: db, timeout: c.busyTimeout, poll: c.busyPoll} + busyWaits.Store(id, wait) + tls := libc.NewTLS() + defer tls.Close() + sqlite3.Xsqlite3_busy_handler(tls, db, busyHandlerFP, id) + return &busyConn{driverConn: full, id: id, wait: wait}, nil +} + +// busyConn is a driver connection whose waits for a lock end with their context. +type busyConn struct { + // driverConn is the driver's own connection. + driverConn + // id is the key of the connection's wait in busyWaits. + id uintptr + // wait is the connection's wait, the one busyWaits holds under id. + wait *busyWait +} + +// during hands ctx's Done channel to the busy handler for one call and returns what ends that span, never a later one. +func (c *busyConn) during(ctx context.Context) func() { + s := &callSpan{done: ctx.Done()} + c.wait.span.Store(s) + return func() { c.wait.span.CompareAndSwap(s, nil) } +} + +// BeginTx starts a transaction, and an error after ctx ended also matches the error of ctx. +func (c *busyConn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, error) { + defer c.during(ctx)() + tx, err := c.driverConn.BeginTx(ctx, opts) + if err != nil { + return nil, ended(ctx, err) + } + return tx, nil +} + +// ExecContext runs a statement, and answers the error of ctx alone when it failed after ctx ended. +func (c *busyConn) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) { + defer c.during(ctx)() + r, err := c.driverConn.ExecContext(ctx, query, args) + if err != nil { + return nil, interrupted(ctx, err) + } + return r, nil +} + +// QueryContext runs a query whose waits end with ctx until its rows close, and answers ctx's error alone once it ended. +func (c *busyConn) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) { + endSpan := c.during(ctx) + rows, err := c.driverConn.QueryContext(ctx, query, args) + if err != nil { + endSpan() + return nil, interrupted(ctx, err) + } + if end := ctx.Err(); end != nil { + _ = rows.Close() + endSpan() + return nil, end + } + return keepSpan(ctx, rows, endSpan), nil +} + +// driverRows is every interface the driver's rows offer to database/sql. +type driverRows interface { + driver.Rows + driver.RowsColumnTypeDatabaseTypeName + driver.RowsColumnTypeLength + driver.RowsColumnTypeNullable + driver.RowsColumnTypePrecisionScale + driver.RowsColumnTypeScanType +} + +// keepSpan returns rows that end the span at Close, or ends it and returns rows as they are if they lack an interface. +func keepSpan(ctx context.Context, rows driver.Rows, endSpan func()) driver.Rows { + full, ok := rows.(driverRows) + if !ok { + endSpan() + return rows + } + return &busyRows{driverRows: full, ctx: ctx, endSpan: endSpan} +} + +// busyRows is the driver's rows of a query whose waits end with the query's context until the rows close. +type busyRows struct { + // driverRows is the driver's own rows. + driverRows + // ctx is the context of the query. + ctx context.Context + // endSpan ends the span of the query. + endSpan func() +} + +// Next reads the next row, and answers the error of ctx alone when it failed after ctx ended. +func (r *busyRows) Next(dest []driver.Value) error { + err := r.driverRows.Next(dest) + if err == nil || err == io.EOF { + return err + } + return interrupted(r.ctx, err) +} + +// Close closes the rows and ends the span of their query. +func (r *busyRows) Close() error { + err := r.driverRows.Close() + r.endSpan() + return err +} + +// PrepareContext prepares a statement, and answers the error of ctx alone when it failed after ctx ended. +func (c *busyConn) PrepareContext(ctx context.Context, query string) (driver.Stmt, error) { + defer c.during(ctx)() + stmt, err := c.driverConn.PrepareContext(ctx, query) + if err != nil { + return nil, interrupted(ctx, err) + } + return stmt, nil +} + +// Ping checks the connection, and answers the error of ctx alone when it failed after ctx ended. +func (c *busyConn) Ping(ctx context.Context) error { + defer c.during(ctx)() + if err := c.driverConn.Ping(ctx); err != nil { + return interrupted(ctx, err) + } + return nil +} + +// Close closes the connection and forgets its wait. +func (c *busyConn) Close() error { + err := c.driverConn.Close() + busyWaits.Delete(c.id) + return err +} + +// ended returns err marked with the error of ctx once ctx has ended, and err as it is before. +func ended(ctx context.Context, err error) error { + if end := ctx.Err(); end != nil { + return fmt.Errorf("%w: %w", err, end) + } + return err +} + +// interrupted returns the error of ctx in place of err once ctx has ended, and err as it is before. +func interrupted(ctx context.Context, err error) error { + if end := ctx.Err(); end != nil { + return end + } + return err +} diff --git a/dbkit/sqlite/busy_internal_test.go b/dbkit/sqlite/busy_internal_test.go new file mode 100644 index 0000000..a68a0a5 --- /dev/null +++ b/dbkit/sqlite/busy_internal_test.go @@ -0,0 +1,799 @@ +// SPDX-License-Identifier: Apache-2.0 + +package sqlite + +import ( + "cmp" + "context" + "database/sql/driver" + "errors" + "fmt" + "io" + "path/filepath" + "reflect" + "testing" + "time" + + "modernc.org/libc" +) + +const ( + // internalCount is a busy count past the first call of a wait. + internalCount = 7 + // internalFakeHandle is a handle number no SQLite connection has. + internalFakeHandle = 1 + // internalUpstreamMethod is the method a driver connection gains with an upstream busy handler. + internalUpstreamMethod = "RegisterBusyHandler" + // internalKnownMethod is a method of the driver connection the lookup must find. + internalKnownMethod = "Ping" + // internalTypeName is the database type name the typed rows answer. + internalTypeName = "DECIMAL" + // internalLength is the column length the typed rows answer. + internalLength = 20 + // internalPrecision is the precision the typed rows answer. + internalPrecision = 10 + // internalScale is the scale the typed rows answer. + internalScale = 2 +) + +func TestEndedMarksAnErrorOnlyOnceItsContextEnded(t *testing.T) { + t.Parallel() + + cancelled, cancel := context.WithCancel(t.Context()) + cancel() + + if got := ended(t.Context(), errStepFailed); got != errStepFailed { + t.Errorf("ended() under a live context = %v, want the error as it is", got) + } + got := ended(cancelled, errStepFailed) + if !errors.Is(got, errStepFailed) || !errors.Is(got, context.Canceled) { + t.Errorf("ended() under a cancelled context = %v, want an error matching both the error and context.Canceled", + got) + } + if want := errStepFailed.Error() + ": " + context.Canceled.Error(); got == nil || got.Error() != want { + t.Errorf("ended() under a cancelled context = %v, want %q", got, want) + } +} + +// internalHandleConn is a driver connection with a field db and only the methods of internalConn. +type internalHandleConn struct { + // internalConn gives the connection the methods of driver.Conn and ExecContext. + internalConn + // db is the handle number the connection holds. + db uintptr +} + +// internalEmbeddedConn is a driver connection whose field db comes through an embedded pointer. +type internalEmbeddedConn struct { + // internalHandleConn gives the connection a field db, left nil. + *internalHandleConn +} + +// internalWrongHandleConn is a driver connection whose field db is no number. +type internalWrongHandleConn struct { + // internalConn gives the connection the methods of driver.Conn and ExecContext. + internalConn + // db is a handle of the wrong type. + db string +} + +// internalNumberConn is a driver connection that is a number, not a struct. +type internalNumberConn int + +// Prepare refuses every statement. +func (internalNumberConn) Prepare(string) (driver.Stmt, error) { + return nil, errors.New("busy test: no statement is prepared on an internalNumberConn") +} + +// Close does nothing. +func (internalNumberConn) Close() error { + return nil +} + +// Begin refuses every transaction. +func (internalNumberConn) Begin() (driver.Tx, error) { + return nil, errors.New("busy test: no transaction runs on an internalNumberConn") +} + +// internalNoHandle returns the error text of handleOf for a connection of the type named kind with no handle. +func internalNoHandle(kind string) string { + return "dbkit: the SQLite driver connection " + kind + + " holds no handle in its field db, so a busy wait cannot end with its context" +} + +func TestHandleOfRefusesAConnectionWithoutAHandle(t *testing.T) { + t.Parallel() + + cases := []struct { + // name names the connection. + name string + // conn is the connection handleOf reads. + conn driver.Conn + // want is the error handleOf answers. + want string + }{ + {"a struct with no field db", &internalConn{}, internalNoHandle("*sqlite.internalConn")}, + {"a field db of another type", &internalWrongHandleConn{db: "site.db"}, + internalNoHandle("*sqlite.internalWrongHandleConn")}, + {"a field db of zero", &internalHandleConn{}, internalNoHandle("*sqlite.internalHandleConn")}, + {"a field db behind an embedded pointer", &internalEmbeddedConn{}, internalNoHandle("*sqlite.internalEmbeddedConn")}, + {"a value that is no struct", internalNumberConn(0), + "dbkit: the SQLite driver connection sqlite.internalNumberConn is not a struct, so a busy wait cannot end " + + "with its context"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + t.Parallel() + + db, err := handleOf(c.conn) + + if db != 0 { + t.Errorf("handleOf() handle = %d, want 0", db) + } + if got := fmt.Sprint(err); got != c.want { + t.Errorf("handleOf() error = %s, want %q", got, c.want) + } + }) + } +} + +func TestConnectFailsClosedWithoutAHandle(t *testing.T) { + t.Parallel() + + fake := &internalConn{} + c, _ := internalConnector(fake, nil) + c.wait = c.waitOn + + conn, err := c.Connect(t.Context()) + + if conn != nil { + t.Errorf("Connect() connection = %v, want nil", conn) + } + if got, want := fmt.Sprint(err), internalNoHandle("*sqlite.internalConn"); got != want { + t.Errorf("Connect() error = %s, want %q", got, want) + } + if !fake.closed { + t.Errorf("closed = %v, want the connection closed", fake.closed) + } +} + +// internalWaitHolds reports whether busyWaits holds a wait on the handle db. +func internalWaitHolds(db uintptr) bool { + held := false + busyWaits.Range(func(_, v any) bool { + held = v.(*busyWait).db == db + return !held + }) + return held +} + +func TestWaitOnRefusesAConnectionMissingAnInterface(t *testing.T) { + t.Parallel() + + c, _ := internalConnector(nil, nil) + + conn, err := c.waitOn(&internalHandleConn{db: internalFakeHandle}) + + want := "dbkit: the SQLite driver connection *sqlite.internalHandleConn lacks an interface database/sql uses" + if conn != nil { + t.Errorf("waitOn() connection = %v, want nil", conn) + } + if got := fmt.Sprint(err); got != want { + t.Errorf("waitOn() error = %s, want %q", got, want) + } + if internalWaitHolds(internalFakeHandle) { + t.Errorf("busyWaits holds a wait on the handle %d, want SQLite never handed the number", internalFakeHandle) + } +} + +// internalRealConn returns a connection Connect opened to a fresh file, closed when the test ends. +func internalRealConn(t *testing.T) *busyConn { + t.Helper() + if err := ofdOnce(); err != nil { + t.Fatalf("ofdOnce() error = %v, want nil", err) + } + c := newConnector(internalDriver(t), filepath.Join(t.TempDir(), "site.db"), internalOptions()) + conn, err := c.Connect(t.Context()) + if err != nil { + t.Fatalf("Connect() error = %v, want nil", err) + } + t.Cleanup(func() { _ = conn.Close() }) + waiting, ok := conn.(*busyConn) + if !ok { + t.Fatalf("Connect() = %T, want a *busyConn", conn) + } + return waiting +} + +// internalHandler calls busyHandler with a fresh libc TLS for the wait of id at count. +func internalHandler(id uintptr, count int32) int32 { + tls := libc.NewTLS() + defer tls.Close() + return busyHandler(tls, id, count) +} + +func TestAClosedConnectionForgetsItsWait(t *testing.T) { + t.Parallel() + + conn := internalRealConn(t) + if _, ok := busyWaits.Load(conn.id); !ok { + t.Fatalf("busyWaits holds no wait for the open connection %d, want its wait", conn.id) + } + + if err := conn.Close(); err != nil { + t.Fatalf("Close() error = %v, want nil", err) + } + + if _, ok := busyWaits.Load(conn.id); ok { + t.Errorf("busyWaits still holds the wait of the closed connection %d, want it forgotten", conn.id) + } + if got := internalHandler(conn.id, 0); got != 0 { + t.Errorf("busyHandler() for a forgotten wait = %d, want 0 to answer busy", got) + } +} + +func TestAWaitThatStartsAboveCountZeroStartsFresh(t *testing.T) { + t.Parallel() + + conn := internalRealConn(t) + conn.wait.began = time.Now().Add(-internalUnreachedWait) + conn.wait.span.Store(&callSpan{}) + + got := internalHandler(conn.id, internalCount) + + if got != 1 { + t.Errorf("busyHandler() at count %d of a call's first wait = %d, want 1 to try the lock again", internalCount, + got) + } +} + +func TestAWaitOutsideACallKeepsItsStartPastCountZero(t *testing.T) { + t.Parallel() + + conn := internalRealConn(t) + conn.wait.began = time.Now().Add(-internalUnreachedWait) + + got := internalHandler(conn.id, internalCount) + + if got != 0 { + t.Errorf("busyHandler() at count %d of a wait an hour old = %d, want 0 to answer busy", internalCount, got) + } +} + +func TestAWaitOutsideACallStartsFreshAtCountZero(t *testing.T) { + t.Parallel() + + conn := internalRealConn(t) + conn.wait.began = time.Now().Add(-internalUnreachedWait) + + got := internalHandler(conn.id, 0) + + if got != 1 { + t.Errorf("busyHandler() at count 0 after a wait an hour old = %d, want 1 to try the lock again", got) + } +} + +// internalSpanConn is a driver connection that records the Done channel the wait's span holds during each call. +type internalSpanConn struct { + // driverConn is left nil. + driverConn + // wait is the wait whose span each call reads. + wait *busyWait + // seen maps each call to the Done channel of the span it ran in, nil for a call with no span. + seen map[string]<-chan struct{} +} + +// record notes the Done channel of the span the call name runs in. +func (c *internalSpanConn) record(name string) { + c.seen[name] = nil + if s := c.wait.span.Load(); s != nil { + c.seen[name] = s.done + } +} + +// BeginTx records its span and starts no transaction. +func (c *internalSpanConn) BeginTx(context.Context, driver.TxOptions) (driver.Tx, error) { + c.record("BeginTx") + return nil, errStepFailed +} + +// ExecContext records its span and runs nothing. +func (c *internalSpanConn) ExecContext(context.Context, string, []driver.NamedValue) (driver.Result, error) { + c.record("ExecContext") + return nil, errStepFailed +} + +// QueryContext records its span and runs nothing. +func (c *internalSpanConn) QueryContext(context.Context, string, []driver.NamedValue) (driver.Rows, error) { + c.record("QueryContext") + return nil, errStepFailed +} + +// PrepareContext records its span and prepares nothing. +func (c *internalSpanConn) PrepareContext(context.Context, string) (driver.Stmt, error) { + c.record("PrepareContext") + return nil, errStepFailed +} + +// Ping records its span. +func (c *internalSpanConn) Ping(context.Context) error { + c.record("Ping") + return nil +} + +func TestEveryCallHandsItsDoneChannelToTheWait(t *testing.T) { + t.Parallel() + + wait := &busyWait{} + fake := &internalSpanConn{wait: wait, seen: map[string]<-chan struct{}{}} + conn := &busyConn{driverConn: fake, wait: wait} + ctx := t.Context() + + _, _ = conn.BeginTx(ctx, driver.TxOptions{}) + _, _ = conn.ExecContext(ctx, internalOptimize, nil) + _, _ = conn.QueryContext(ctx, internalOptimize, nil) + _, _ = conn.PrepareContext(ctx, internalOptimize) + _ = conn.Ping(ctx) + + for _, name := range []string{"BeginTx", "ExecContext", "QueryContext", "PrepareContext", "Ping"} { + if done, ok := fake.seen[name]; !ok || done != ctx.Done() { + t.Errorf("the span during %s holds %v, want the Done channel of the call's context", name, done) + } + } + if s := wait.span.Load(); s != nil { + t.Errorf("the span after every call = %+v, want none between calls", s) + } +} + +// internalBusyConn is a driver connection whose every call answers a busy error. +type internalBusyConn struct { + // driverConn is left nil. + driverConn + // busy is the error every call answers with. + busy error +} + +// BeginTx answers the busy error. +func (c *internalBusyConn) BeginTx(context.Context, driver.TxOptions) (driver.Tx, error) { + return nil, c.busy +} + +// ExecContext answers the busy error. +func (c *internalBusyConn) ExecContext(context.Context, string, []driver.NamedValue) (driver.Result, error) { + return nil, c.busy +} + +// QueryContext answers the busy error. +func (c *internalBusyConn) QueryContext(context.Context, string, []driver.NamedValue) (driver.Rows, error) { + return nil, c.busy +} + +// PrepareContext answers the busy error. +func (c *internalBusyConn) PrepareContext(context.Context, string) (driver.Stmt, error) { + return nil, c.busy +} + +// Ping answers the busy error. +func (c *internalBusyConn) Ping(context.Context) error { + return c.busy +} + +// internalStatementCalls returns each statement call of conn as a function of its context. +func internalStatementCalls(conn *busyConn) map[string]func(ctx context.Context) error { + return map[string]func(ctx context.Context) error{ + "ExecContext": func(ctx context.Context) error { + _, err := conn.ExecContext(ctx, internalOptimize, nil) + return err + }, + "QueryContext": func(ctx context.Context) error { + _, err := conn.QueryContext(ctx, internalOptimize, nil) + return err + }, + "PrepareContext": func(ctx context.Context) error { + _, err := conn.PrepareContext(ctx, internalOptimize) + return err + }, + "Ping": conn.Ping, + } +} + +func TestAStatementAfterItsContextEndedAnswersTheContextError(t *testing.T) { + t.Parallel() + + busy := internalBusyError(t) + conn := &busyConn{driverConn: &internalBusyConn{busy: busy}, wait: &busyWait{}} + cancelled, cancel := context.WithCancel(t.Context()) + cancel() + + for name, call := range internalStatementCalls(conn) { + if err := call(cancelled); err != context.Canceled { + t.Errorf("%s under a cancelled context error = %v, want exactly context.Canceled", name, err) + } + if err := call(t.Context()); err != busy { + t.Errorf("%s under a live context error = %v, want the busy error as it is", name, err) + } + } +} + +func TestABeginAfterItsContextEndedAlsoAnswersTheContextError(t *testing.T) { + t.Parallel() + + busy := internalBusyError(t) + conn := &busyConn{driverConn: &internalBusyConn{busy: busy}, wait: &busyWait{}} + cancelled, cancel := context.WithCancel(t.Context()) + cancel() + + _, err := conn.BeginTx(cancelled, driver.TxOptions{}) + _, live := conn.BeginTx(t.Context(), driver.TxOptions{}) + + if !errors.Is(err, busy) || !errors.Is(err, context.Canceled) { + t.Errorf("BeginTx() under a cancelled context error = %v, want one matching both busy and context.Canceled", err) + } + if live != busy { + t.Errorf("BeginTx() under a live context error = %v, want the busy error as it is", live) + } +} + +// internalRows is a result set with no rows that records its Close. +type internalRows struct { + // closed reports that Close ran. + closed bool +} + +// Columns names no column. +func (*internalRows) Columns() []string { + return nil +} + +// Close records that it ran. +func (r *internalRows) Close() error { + r.closed = true + return nil +} + +// Next answers the end of the rows. +func (*internalRows) Next([]driver.Value) error { + return io.EOF +} + +// internalRowsConn is a driver connection whose every query answers its rows. +type internalRowsConn struct { + // driverConn is left nil. + driverConn + // rows is the result set every query answers. + rows driver.Rows +} + +// QueryContext answers the rows of the connection. +func (c *internalRowsConn) QueryContext(context.Context, string, []driver.NamedValue) (driver.Rows, error) { + return c.rows, nil +} + +// internalQuery runs a query under ctx through a wrapper whose driver answers rows, and returns what it answered. +func internalQuery(ctx context.Context, rows *internalRows) (driver.Rows, error) { + conn := &busyConn{driverConn: &internalRowsConn{rows: rows}, wait: &busyWait{}} + return conn.QueryContext(ctx, internalOptimize, nil) +} + +func TestAQueryThatAnswersAfterItsContextEndedAnswersTheContextError(t *testing.T) { + t.Parallel() + + cancelled, cancel := context.WithCancel(t.Context()) + cancel() + ended := &internalRows{} + + rows, err := internalQuery(cancelled, ended) + + if rows != nil || err != context.Canceled { + t.Errorf("QueryContext() under a cancelled context = %v, %v, want no rows and exactly context.Canceled", rows, err) + } + if !ended.closed { + t.Errorf("closed = %v, want the rows of a query whose context ended closed", ended.closed) + } +} + +func TestAQueryUnderALiveContextAnswersItsRows(t *testing.T) { + t.Parallel() + + live := &internalRows{} + + rows, err := internalQuery(t.Context(), live) + + if rows != driver.Rows(live) || err != nil { + t.Errorf("QueryContext() under a live context = %v, %v, want its rows and nil", rows, err) + } + if live.closed { + t.Errorf("closed = %v, want the rows of a live query left open", live.closed) + } +} + +// internalTypedRows is a result set with no rows that answers each column type question with a value of its own. +type internalTypedRows struct { + // internalRows gives the rows Columns, Next and a recorded Close. + internalRows +} + +// ColumnTypeDatabaseTypeName answers internalTypeName. +func (*internalTypedRows) ColumnTypeDatabaseTypeName(int) string { + return internalTypeName +} + +// ColumnTypeLength answers internalLength. +func (*internalTypedRows) ColumnTypeLength(int) (int64, bool) { + return internalLength, true +} + +// ColumnTypeNullable answers that the column holds no NULL. +func (*internalTypedRows) ColumnTypeNullable(int) (bool, bool) { + return false, true +} + +// ColumnTypePrecisionScale answers internalPrecision and internalScale. +func (*internalTypedRows) ColumnTypePrecisionScale(int) (int64, int64, bool) { + return internalPrecision, internalScale, true +} + +// ColumnTypeScanType answers the type of a bool. +func (*internalTypedRows) ColumnTypeScanType(int) reflect.Type { + return reflect.TypeFor[bool]() +} + +// internalTypedQuery returns a wrapper on wait whose driver answers typed to every query. +func internalTypedQuery(wait *busyWait, typed *internalTypedRows) *busyConn { + return &busyConn{driverConn: &internalRowsConn{rows: typed}, wait: wait} +} + +// internalSpanDone returns the Done channel the span of wait holds, nil when it holds no span. +func internalSpanDone(wait *busyWait) <-chan struct{} { + if s := wait.span.Load(); s != nil { + return s.done + } + return nil +} + +func TestTheRowsOfAQueryKeepItsSpanUntilTheyClose(t *testing.T) { + t.Parallel() + + wait, typed := &busyWait{}, &internalTypedRows{} + ctx := t.Context() + + rows, err := internalTypedQuery(wait, typed).QueryContext(ctx, internalOptimize, nil) + + if err != nil { + t.Fatalf("QueryContext() error = %v, want nil", err) + } + if done := internalSpanDone(wait); done != ctx.Done() { + t.Errorf("the span while the rows are open holds %v, want the Done channel of the query's context", done) + } + if err := rows.Close(); err != nil { + t.Fatalf("Close() error = %v, want nil", err) + } + if s := wait.span.Load(); s != nil { + t.Errorf("the span after Close = %+v, want none", s) + } + if !typed.closed { + t.Errorf("closed = %v, want the driver's rows closed", typed.closed) + } +} + +func TestClosingRowsTwiceLeavesTheSpanOfALaterQuery(t *testing.T) { + t.Parallel() + + wait := &busyWait{} + conn := internalTypedQuery(wait, &internalTypedRows{}) + later, cancel := context.WithCancel(t.Context()) + defer cancel() + first, err := conn.QueryContext(t.Context(), internalOptimize, nil) + if err != nil { + t.Fatalf("QueryContext() error = %v, want nil", err) + } + second, err := conn.QueryContext(later, internalOptimize, nil) + if err != nil { + t.Fatalf("QueryContext() under the later context error = %v, want nil", err) + } + + closed, again := first.Close(), first.Close() + + if err := errors.Join(closed, again); err != nil { + t.Fatalf("Close() twice error = %v, want nil", err) + } + if done := internalSpanDone(wait); done != later.Done() { + t.Errorf("the span after the first rows closed twice holds %v, want the Done channel of the later query", done) + } + if err := second.Close(); err != nil { + t.Fatalf("Close() of the later rows error = %v, want nil", err) + } + if s := wait.span.Load(); s != nil { + t.Errorf("the span after the later rows closed = %+v, want none", s) + } +} + +func TestAQueryAnsweringNoWrappedRowsEndsItsSpan(t *testing.T) { + t.Parallel() + + cancelled, cancel := context.WithCancel(t.Context()) + cancel() + cases := []struct { + // name names the query. + name string + // ctx is the context of the query. + ctx context.Context + // driverConn is the driver connection that answers the query. + driverConn driverConn + }{ + {"a query the driver failed", t.Context(), &internalBusyConn{busy: errStepFailed}}, + {"a query whose context ended", cancelled, &internalRowsConn{rows: &internalTypedRows{}}}, + {"rows lacking a column type interface", t.Context(), &internalRowsConn{rows: &internalRows{}}}, + } + for _, c := range cases { + wait := &busyWait{} + + rows, _ := (&busyConn{driverConn: c.driverConn, wait: wait}).QueryContext(c.ctx, internalOptimize, nil) + + if s := wait.span.Load(); s != nil { + t.Errorf("the span after %s = %+v, want none", c.name, s) + } + if rows != nil { + _ = rows.Close() + } + } +} + +// internalSpanRows is typed rows whose Close records the Done channel the span of wait holds and fails. +type internalSpanRows struct { + // internalTypedRows answers every column type question. + internalTypedRows + // wait is the wait whose span Close records. + wait *busyWait + // closedUnder is the Done channel the span held when Close ran. + closedUnder <-chan struct{} +} + +// Close records the Done channel the span of wait holds and fails with errStepFailed. +func (r *internalSpanRows) Close() error { + r.closedUnder = internalSpanDone(r.wait) + return errStepFailed +} + +func TestTheDriversRowsCloseUnderTheSpanOfTheirQuery(t *testing.T) { + t.Parallel() + + cancelled, cancel := context.WithCancel(t.Context()) + cancel() + for _, ctx := range []context.Context{t.Context(), cancelled} { + wait := &busyWait{} + spanRows := &internalSpanRows{wait: wait} + conn := &busyConn{driverConn: &internalRowsConn{rows: spanRows}, wait: wait} + + rows, err := conn.QueryContext(ctx, internalOptimize, nil) + if err == nil { + err = rows.Close() + } + + if spanRows.closedUnder != ctx.Done() { + t.Errorf("the span when the driver's rows closed held %v, want the Done channel %v of their query", + spanRows.closedUnder, ctx.Done()) + } + if want := cmp.Or(ctx.Err(), errStepFailed); err != want { + t.Errorf("QueryContext() then Close() error = %v, want %v as it is", err, want) + } + if s := wait.span.Load(); s != nil { + t.Errorf("the span after the driver's rows closed = %+v, want none", s) + } + } +} + +// internalColumnProbes asks rows each column type question and reports whether they answer as internalTypedRows. +var internalColumnProbes = []struct { + // name names the interface. + name string + // answers reports whether rows offer the interface and answer it as internalTypedRows. + answers func(rows driver.Rows) bool +}{ + {"driver.RowsColumnTypeDatabaseTypeName", func(rows driver.Rows) bool { + r, ok := rows.(driver.RowsColumnTypeDatabaseTypeName) + return ok && r.ColumnTypeDatabaseTypeName(0) == internalTypeName + }}, + {"driver.RowsColumnTypeLength", func(rows driver.Rows) bool { + r, ok := rows.(driver.RowsColumnTypeLength) + if !ok { + return false + } + length, known := r.ColumnTypeLength(0) + return known && length == internalLength + }}, + {"driver.RowsColumnTypeNullable", func(rows driver.Rows) bool { + r, ok := rows.(driver.RowsColumnTypeNullable) + if !ok { + return false + } + nullable, known := r.ColumnTypeNullable(0) + return known && !nullable + }}, + {"driver.RowsColumnTypePrecisionScale", func(rows driver.Rows) bool { + r, ok := rows.(driver.RowsColumnTypePrecisionScale) + if !ok { + return false + } + precision, scale, known := r.ColumnTypePrecisionScale(0) + return known && precision == internalPrecision && scale == internalScale + }}, + {"driver.RowsColumnTypeScanType", func(rows driver.Rows) bool { + r, ok := rows.(driver.RowsColumnTypeScanType) + return ok && r.ColumnTypeScanType(0) == reflect.TypeFor[bool]() + }}, +} + +func TestTheRowsOfAQueryKeepEveryColumnTypeInterface(t *testing.T) { + t.Parallel() + + rows, err := internalTypedQuery(&busyWait{}, &internalTypedRows{}).QueryContext(t.Context(), internalOptimize, nil) + if err != nil { + t.Fatalf("QueryContext() error = %v, want nil", err) + } + defer func() { _ = rows.Close() }() + + for _, p := range internalColumnProbes { + if !p.answers(rows) { + t.Errorf("%s on the rows of a query is missing or answers otherwise, want the driver's answer", p.name) + } + } +} + +// internalDriverConn returns a connection the driver itself opened to a fresh file, closed when the test ends. +func internalDriverConn(t *testing.T) driver.Conn { + t.Helper() + if err := ofdOnce(); err != nil { + t.Fatalf("ofdOnce() error = %v, want nil", err) + } + conn, err := internalDriver(t).Open(filepath.Join(t.TempDir(), "site.db")) + if err != nil { + t.Fatalf("Open() error = %v, want nil", err) + } + t.Cleanup(func() { _ = conn.Close() }) + return conn +} + +// internalOffers reports whether the type of conn has the method name, pointer receivers included. +func internalOffers(conn driver.Conn, name string) bool { + _, ok := reflect.TypeOf(conn).MethodByName(name) + return ok +} + +func TestTheDriverOffersNoBusyHandlerYet(t *testing.T) { + t.Parallel() + + raw := internalDriverConn(t) + if _, err := handleOf(raw); err != nil { + t.Fatalf("handleOf(%T) error = %v, want the driver's own connection", raw, err) + } + if !internalOffers(raw, internalKnownMethod) { + t.Fatalf("the lookup finds no %s on %T, want it to see every method of the driver connection", + internalKnownMethod, raw) + } + + if internalOffers(raw, internalUpstreamMethod) { + t.Errorf("modernc.org/sqlite now offers %s on %T, so replace the busy handler bridge in busy.go with it", + internalUpstreamMethod, raw) + } +} + +func TestTheDriversRowsOfferEveryInterfaceTheWrapperKeepsAndNoOther(t *testing.T) { + t.Parallel() + + rows, err := internalDriverConn(t).(driver.QueryerContext).QueryContext(t.Context(), "SELECT 1", nil) + if err != nil { + t.Fatalf("QueryContext() error = %v, want nil", err) + } + defer func() { _ = rows.Close() }() + + if _, ok := rows.(driverRows); !ok { + t.Errorf("the driver's rows %T lack an interface of driverRows, so every query answers them unwrapped", rows) + } + if _, ok := rows.(driver.RowsNextResultSet); ok { + t.Errorf("modernc.org/sqlite now offers driver.RowsNextResultSet on %T, so keep it in busyRows", rows) + } + if _, ok := rows.(driver.RowsColumnScanner); ok { + t.Errorf("modernc.org/sqlite now offers driver.RowsColumnScanner on %T, so keep it in busyRows", rows) + } +} diff --git a/dbkit/sqlite/busy_test.go b/dbkit/sqlite/busy_test.go new file mode 100644 index 0000000..5f0f4c1 --- /dev/null +++ b/dbkit/sqlite/busy_test.go @@ -0,0 +1,519 @@ +// SPDX-License-Identifier: Apache-2.0 + +package sqlite_test + +import ( + "context" + "database/sql" + "errors" + "math" + "path/filepath" + "slices" + "testing" + "time" + + "github.com/gopherium/framework/dbkit" + "github.com/gopherium/framework/dbkit/sqlite" +) + +const ( + // busyPoll is how often a call behind a held writer tries the lock again. + busyPoll = 5 * time.Millisecond + // busyDeadline is the deadline, and the cancel, of a call behind a held writer. + busyDeadline = 100 * time.Millisecond + // busyBound is the longest a call behind a held writer may run, a tenth of the busy timeout. + busyBound = optionsBusyTimeout / 10 + // busyLateHold is how long a holder keeps the write lock after the deadline of the write it blocks. + busyLateHold = 400 * time.Millisecond + // busyHold is how long a holder keeps the write lock while a write with no deadline waits for it. + busyHold = 50 * time.Millisecond + // busySchema creates the table the busy tests write to. + busySchema = "CREATE TABLE busy_rows (a INTEGER NOT NULL)" + // busyInsert is the write a held writer makes wait. + busyInsert = "INSERT INTO busy_rows VALUES (1)" + // busyOtherInsert is a write that is not the one under test. + busyOtherInsert = "INSERT INTO busy_rows VALUES (0)" + // busyWritten counts the rows busyInsert wrote. + busyWritten = "SELECT count(*) FROM busy_rows WHERE a = 1" + // busyLateSchema is the schema change that leaves a connection's cached schema stale. + busyLateSchema = "CREATE TABLE late_schema (a INTEGER)" + // busyAuxSchema creates the table of the file a connection attaches as aux. + busyAuxSchema = "CREATE TABLE aux_rows (a INTEGER)" + // busyAttach attaches the file named by its one argument as aux. + busyAttach = "ATTACH DATABASE ? AS aux" + // busyLaterStep answers one row on its first step and reads aux only on its second. + busyLaterStep = "SELECT 1 UNION ALL SELECT page_count FROM pragma_page_count('aux')" + // busyAuxRead reads aux, so a holder inside a transaction keeps its read lock. + busyAuxRead = "SELECT count(*) FROM aux_rows" + // busyAuxReturning writes one row to aux on its first step and commits it at the end of its rows. + busyAuxReturning = "INSERT INTO aux.aux_rows VALUES (1) RETURNING a" + // busyTypedSchema creates a table whose columns declare their types. + busyTypedSchema = "CREATE TABLE typed_rows (amount INTEGER NOT NULL, label VARCHAR(20) NOT NULL)" + // busyTypedInsert writes one row of synthetic values to the typed table. + busyTypedInsert = "INSERT INTO typed_rows VALUES (5, 'synthetic')" + // busyTypedRead reads every column of the typed table. + busyTypedRead = "SELECT amount, label FROM typed_rows" +) + +// busyShape is one call that waits for the write lock. +type busyShape struct { + // name names the call. + name string + // run makes the call on db under ctx, with prepared holding busyInsert prepared before the writer took the lock. + run func(ctx context.Context, db *sql.DB, prepared *sql.Stmt) error + // busy reports that the call's error also matches dbkit.ErrBusy through Classify. + busy bool + // longPoll reports that the call also runs on a handle whose busy poll is the busy timeout. + longPoll bool +} + +// busyShapes are the calls a held writer makes wait. +var busyShapes = []busyShape{ + {"an autocommit insert", func(ctx context.Context, db *sql.DB, _ *sql.Stmt) error { + _, err := db.ExecContext(ctx, busyInsert) + return err + }, false, true}, + {"an insert returning its row", func(ctx context.Context, db *sql.DB, _ *sql.Stmt) error { + var a int64 + return db.QueryRowContext(ctx, busyInsert+" RETURNING a").Scan(&a) + }, false, true}, + {"BeginTx", func(ctx context.Context, db *sql.DB, _ *sql.Stmt) error { + tx, err := db.BeginTx(ctx, nil) + if err == nil { + _ = tx.Rollback() + } + return err + }, true, true}, + {"InTx", func(ctx context.Context, db *sql.DB, _ *sql.Stmt) error { + return dbkit.InTx(ctx, db, func(tx *sql.Tx) error { + _, err := tx.ExecContext(ctx, busyInsert) + return err + }) + }, true, false}, + {"a statement prepared before the writer", func(ctx context.Context, _ *sql.DB, prepared *sql.Stmt) error { + _, err := prepared.ExecContext(ctx) + return err + }, false, false}, +} + +// busyCase is one shape run on a handle with one busy poll. +type busyCase struct { + // name names the shape and its poll. + name string + // shape is the call. + shape busyShape + // poll is the busy poll of the handle. + poll time.Duration +} + +// busyCases returns every shape at busyPoll, and the long poll shapes again at a busy poll of the busy timeout. +func busyCases() []busyCase { + var cases []busyCase + for _, shape := range busyShapes { + cases = append(cases, busyCase{name: shape.name, shape: shape, poll: busyPoll}) + if shape.longPoll { + cases = append(cases, busyCase{name: shape.name + " at a busy poll of the busy timeout", shape: shape, + poll: optionsBusyTimeout}) + } + } + return cases +} + +// busyOptions returns testOptions with the busy poll given. +func busyOptions(poll time.Duration) sqlite.Options { + opts := testOptions() + opts.BusyPoll = poll + return opts +} + +// busyBehindAWriter returns a handle with poll, busyInsert prepared on it and a writer holding its lock until cleanup. +func busyBehindAWriter(t *testing.T, poll time.Duration) (*sql.DB, *sql.Stmt) { + t.Helper() + db := rulesOpen(t, busyOptions(poll)) + rulesExec(t, db, busySchema) + busyWarmUp(t, db) + prepared, err := db.PrepareContext(t.Context(), busyInsert) + if err != nil { + t.Fatalf("PrepareContext(%q) error = %v, want nil", busyInsert, err) + } + t.Cleanup(func() { _ = prepared.Close() }) + rulesHoldWriter(t, db, busyOtherInsert) + return db, prepared +} + +// busyWarmUp opens two connections of db at once and returns both to the pool, so a call behind the writer opens none. +func busyWarmUp(t *testing.T, db *sql.DB) { + t.Helper() + first, second := busyPin(t, db), busyPin(t, db) + if err := errors.Join(first.Close(), second.Close()); err != nil { + t.Fatalf("Close() error = %v, want nil", err) + } +} + +// busyEnd is how a call behind a held writer ended. +type busyEnd struct { + // err is the error of the call. + err error + // waited is how long the call ran. + waited time.Duration +} + +// busyRun makes the call of shape on db under ctx and returns how it ended. +func busyRun(ctx context.Context, shape busyShape, db *sql.DB, prepared *sql.Stmt) busyEnd { + began := time.Now() + err := shape.run(ctx, db, prepared) + return busyEnd{err: err, waited: time.Since(began)} +} + +// busyMustEnd fails the test unless the call of shape ended with want within busyBound, busy as shape says. +func busyMustEnd(t *testing.T, shape busyShape, end busyEnd, want error) { + t.Helper() + if !errors.Is(end.err, want) { + t.Errorf("%s error = %v, want one matching %v", shape.name, end.err, want) + } + if busy := errors.Is(sqlite.Classify(end.err), dbkit.ErrBusy); busy != shape.busy { + t.Errorf("%s error %v matches dbkit.ErrBusy = %v, want %v", shape.name, end.err, busy, shape.busy) + } + if end.waited > busyBound { + t.Errorf("%s answered after %v, want within a tenth of the busy timeout %v", shape.name, end.waited, + optionsBusyTimeout) + } +} + +// busyMustEndEvery runs every case behind a held writer under the context end makes and expects want. +func busyMustEndEvery(t *testing.T, end func(context.Context) (context.Context, context.CancelFunc), want error) { + t.Helper() + for _, c := range busyCases() { + t.Run(c.name, func(t *testing.T) { + t.Parallel() + db, prepared := busyBehindAWriter(t, c.poll) + ctx, cancel := end(t.Context()) + defer cancel() + + got := busyRun(ctx, c.shape, db, prepared) + + busyMustEnd(t, c.shape, got, want) + }) + } +} + +// busyDeadlineOf returns ctx with busyDeadline as its deadline. +func busyDeadlineOf(ctx context.Context) (context.Context, context.CancelFunc) { + return context.WithTimeout(ctx, busyDeadline) +} + +// busyCancelOf returns ctx cancelled once busyDeadline has passed, and what stops the cancel and cancels it. +func busyCancelOf(ctx context.Context) (context.Context, context.CancelFunc) { + ctx, cancel := context.WithCancel(ctx) + later := time.AfterFunc(busyDeadline, cancel) + return ctx, func() { + later.Stop() + cancel() + } +} + +func TestABusyWaitEndsAtItsDeadline(t *testing.T) { + t.Parallel() + + busyMustEndEvery(t, busyDeadlineOf, context.DeadlineExceeded) +} + +func TestABusyWaitEndsWhenCancelled(t *testing.T) { + t.Parallel() + + busyMustEndEvery(t, busyCancelOf, context.Canceled) +} + +// busyPin returns a connection of db pinned for the rest of the test. +func busyPin(t *testing.T, db *sql.DB) *sql.Conn { + t.Helper() + conn, err := db.Conn(t.Context()) + if err != nil { + t.Fatalf("Conn() error = %v, want nil", err) + } + t.Cleanup(func() { _ = conn.Close() }) + return conn +} + +// busyPinExec runs query on conn and fails the test when it cannot. +func busyPinExec(t *testing.T, conn *sql.Conn, query string) { + t.Helper() + if _, err := conn.ExecContext(t.Context(), query); err != nil { + t.Fatalf("ExecContext(%q) error = %v, want nil", query, err) + } +} + +// busyBegin takes the write lock on db, released by the test or at cleanup. +func busyBegin(t *testing.T, db *sql.DB) *sql.Tx { + t.Helper() + tx, err := db.BeginTx(t.Context(), nil) + if err != nil { + t.Fatalf("BeginTx() error = %v, want nil", err) + } + t.Cleanup(func() { _ = tx.Rollback() }) + return tx +} + +// busyMustReceive returns what ch receives and fails the test when nothing arrives within busyBound. +func busyMustReceive(t *testing.T, ch <-chan error, what string) error { + t.Helper() + select { + case err := <-ch: + return err + case <-time.After(busyBound): + t.Fatalf("%s still runs after %v, want it ended", what, busyBound) + return nil + } +} + +// busyRows returns how many rows busyInsert wrote to db. +func busyRows(t *testing.T, db *sql.DB) int64 { + t.Helper() + var n int64 + if err := db.QueryRowContext(t.Context(), busyWritten).Scan(&n); err != nil { + t.Fatalf("%s error = %v, want nil", busyWritten, err) + } + return n +} + +func TestALateLockLeavesACancelledWriteUnwritten(t *testing.T) { + t.Parallel() + + db := rulesOpen(t, busyOptions(busyPoll)) + rulesExec(t, db, busySchema) + pinned, other := busyPin(t, db), busyPin(t, db) + busyPinExec(t, pinned, busyOtherInsert) + busyPinExec(t, other, busyLateSchema) + holder, err := other.BeginTx(t.Context(), nil) + if err != nil { + t.Fatalf("BeginTx() error = %v, want nil", err) + } + t.Cleanup(func() { _ = holder.Rollback() }) + ctx, cancel := context.WithTimeout(t.Context(), busyDeadline) + defer cancel() + wrote := make(chan error, 1) + + go func() { + _, err := pinned.ExecContext(ctx, busyInsert) + wrote <- err + }() + <-ctx.Done() + <-time.After(busyLateHold) + if err := holder.Rollback(); err != nil { + t.Fatalf("Rollback() error = %v, want nil", err) + } + + if err := busyMustReceive(t, wrote, "the cancelled write"); !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("the write error = %v, want one matching context.DeadlineExceeded", err) + } + if n := busyRows(t, db); n != 0 { + t.Errorf("rows written = %d once the late lock ended, want 0 from a write whose deadline came first", n) + } +} + +// busyPrepareThenEndAWait returns busyInsert prepared on pinned once it and a write of pinned ended at their deadlines. +func busyPrepareThenEndAWait(t *testing.T, db *sql.DB, pinned *sql.Conn) *sql.Stmt { + t.Helper() + prepared, err := pinned.PrepareContext(t.Context(), busyInsert) + if err != nil { + t.Fatalf("PrepareContext(%q) error = %v, want nil", busyInsert, err) + } + t.Cleanup(func() { _ = prepared.Close() }) + first := busyBegin(t, db) + ctx, cancel := context.WithTimeout(t.Context(), busyDeadline) + defer cancel() + if _, err := pinned.ExecContext(ctx, busyOtherInsert); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("the write behind the first holder error = %v, want one matching context.DeadlineExceeded", err) + } + again, stop := context.WithTimeout(t.Context(), busyDeadline) + defer stop() + if _, err := prepared.ExecContext(again); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("the prepared insert behind the first holder error = %v, want one matching context.DeadlineExceeded", err) + } + if err := first.Rollback(); err != nil { + t.Fatalf("Rollback() error = %v, want nil", err) + } + return prepared +} + +// busyHoldFor takes the write lock on db for hold and returns the channel that receives the error of its release. +func busyHoldFor(t *testing.T, db *sql.DB, hold time.Duration) <-chan error { + t.Helper() + holder := busyBegin(t, db) + released := make(chan error, 1) + time.AfterFunc(hold, func() { released <- holder.Rollback() }) + return released +} + +func TestAConnectionWritesAgainAfterItsWaitEnded(t *testing.T) { + t.Parallel() + + db := rulesOpen(t, busyOptions(busyPoll)) + rulesExec(t, db, busySchema) + prepared := busyPrepareThenEndAWait(t, db, busyPin(t, db)) + began := time.Now() + released := busyHoldFor(t, db, busyHold) + + _, err := prepared.ExecContext(t.Context()) + + waited := time.Since(began) + if err != nil { + t.Errorf("the prepared insert error = %v, want it written once the second holder let go", err) + } + if waited < busyHold || waited > busyBound { + t.Errorf("the prepared insert answered after %v, want after the hold %v and within %v", waited, busyHold, + busyBound) + } + if err := busyMustReceive(t, released, "the second holder"); err != nil { + t.Errorf("Rollback() error = %v, want nil", err) + } + if n := busyRows(t, db); n != 1 { + t.Errorf("rows written = %d, want 1 from the prepared insert", n) + } +} + +func TestEveryConnectionEndsItsWaitAtItsDeadline(t *testing.T) { + t.Parallel() + + db := rulesHeldElsewhere(t, testOptions()) + + for i, conn := range rulesConns(t, db) { + ctx, cancel := context.WithTimeout(t.Context(), busyDeadline) + waited, err := rulesBegin(ctx, conn) + cancel() + + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("BeginTx() on connection %d error = %v, want one matching context.DeadlineExceeded", i, err) + } + if waited > busyBound { + t.Errorf("BeginTx() on connection %d answered after %v, want within a tenth of the busy timeout %v", i, + waited, optionsBusyTimeout) + } + } +} + +// busyAttachHeld returns a connection of db with aux, a rollback journal file a plain handle locks by running hold. +func busyAttachHeld(t *testing.T, db *sql.DB, hold ...string) *sql.Conn { + t.Helper() + path := filepath.Join(t.TempDir(), "aux.db") + plain, err := sql.Open("sqlite", path) + if err != nil { + t.Fatalf("sql.Open() error = %v, want nil", err) + } + t.Cleanup(func() { _ = plain.Close() }) + holder, pinned := busyPin(t, plain), busyPin(t, db) + busyPinExec(t, holder, busyAuxSchema) + if _, err := pinned.ExecContext(t.Context(), busyAttach, path); err != nil { + t.Fatalf("ExecContext(%q) error = %v, want nil", busyAttach, err) + } + for _, query := range hold { + busyPinExec(t, holder, query) + } + return pinned +} + +func TestALaterStepOfAQueryEndsItsWaitWithItsContext(t *testing.T) { + t.Parallel() + + pinned := busyAttachHeld(t, rulesOpen(t, busyOptions(busyPoll)), "BEGIN EXCLUSIVE") + ctx, cancel := context.WithTimeout(t.Context(), busyDeadline) + defer cancel() + began := time.Now() + rows, err := pinned.QueryContext(ctx, busyLaterStep) + if err != nil { + t.Fatalf("QueryContext(%q) error = %v, want its first row", busyLaterStep, err) + } + defer func() { _ = rows.Close() }() + + first, second := rows.Next(), rows.Next() + + waited := time.Since(began) + if !first || second { + t.Errorf("Next() = %v, then %v, want the first row and then no second while aux is held", first, second) + } + if err := rows.Err(); err != context.DeadlineExceeded { + t.Errorf("Err() = %v, want exactly the deadline once the second step stopped", err) + } + if waited > busyBound { + t.Errorf("the second step answered after %v, want within a tenth of the busy timeout %v", waited, + optionsBusyTimeout) + } +} + +func TestClosingTheRowsOfAWriteEndsItsWaitWithItsContext(t *testing.T) { + t.Parallel() + + pinned := busyAttachHeld(t, rulesOpen(t, busyOptions(busyPoll)), "BEGIN", busyAuxRead) + ctx, cancel := context.WithTimeout(t.Context(), busyDeadline) + defer cancel() + began := time.Now() + var a int64 + + err := pinned.QueryRowContext(ctx, busyAuxReturning).Scan(&a) + + waited := time.Since(began) + if !errors.Is(err, context.DeadlineExceeded) && !errors.Is(sqlite.Classify(err), dbkit.ErrBusy) { + t.Errorf("Scan() error = %v, want the deadline or busy once the commit at Close stopped", err) + } + if waited > busyBound { + t.Errorf("the commit at Close answered after %v, want within a tenth of the busy timeout %v", waited, + optionsBusyTimeout) + } +} + +// busyColumn is what database/sql reports of one column. +type busyColumn struct { + // name is the column's name. + name string + // typeName is the column's database type name. + typeName string + // scan names the column's scan type. + scan string + // length is the column's length. + length int64 + // hasLength reports that the length is known. + hasLength bool + // nullable reports that the column may hold NULL. + nullable bool + // knowsNullable reports that nullable is known. + knowsNullable bool +} + +// busyColumnOf returns what database/sql reports of the column c. +func busyColumnOf(c *sql.ColumnType) busyColumn { + length, hasLength := c.Length() + nullable, knowsNullable := c.Nullable() + return busyColumn{name: c.Name(), typeName: c.DatabaseTypeName(), scan: c.ScanType().String(), length: length, + hasLength: hasLength, nullable: nullable, knowsNullable: knowsNullable} +} + +func TestTheRowsOfAQueryKeepTheDriversColumnTypes(t *testing.T) { + t.Parallel() + + db := rulesOpen(t, testOptions()) + rulesExec(t, db, busyTypedSchema) + rulesExec(t, db, busyTypedInsert) + rows, err := db.QueryContext(t.Context(), busyTypedRead) + if err != nil { + t.Fatalf("QueryContext(%q) error = %v, want nil", busyTypedRead, err) + } + defer func() { _ = rows.Close() }() + + types, err := rows.ColumnTypes() + + if err != nil { + t.Fatalf("ColumnTypes() error = %v, want nil", err) + } + got := make([]busyColumn, len(types)) + for i, c := range types { + got[i] = busyColumnOf(c) + } + want := []busyColumn{ + {name: "amount", typeName: "INTEGER", scan: "int64", nullable: true, knowsNullable: true}, + {name: "label", typeName: "VARCHAR(20)", scan: "string", length: math.MaxInt64, hasLength: true, nullable: true, + knowsNullable: true}, + } + if !slices.Equal(got, want) { + t.Errorf("ColumnTypes() = %+v, want the driver's %+v", got, want) + } +} diff --git a/dbkit/sqlite/connector.go b/dbkit/sqlite/connector.go index 71a16ee..2e17ef5 100644 --- a/dbkit/sqlite/connector.go +++ b/dbkit/sqlite/connector.go @@ -6,9 +6,9 @@ import ( "context" "database/sql/driver" "errors" + "fmt" "log/slog" - "strconv" - "sync" + "time" modernc "modernc.org/sqlite" @@ -31,14 +31,18 @@ type connector struct { driver *modernc.Driver // open opens one connection by the driver's name of the file. open func(name string) (driver.Conn, error) - // opening lets one call of open run at a time. - opening sync.Mutex + // opening is the one slot a try of open holds until it answers, so one try runs at a time. + opening chan struct{} // name is the driver's name of the file with every rule. name string // path is the absolute path of the file. path string - // restore sets a connection's busy timeout back to the option. - restore string + // busyTimeout is how long a busy open or a connection waiting for a lock keeps trying. + busyTimeout time.Duration + // busyPoll is how often a busy open or a connection waiting for a lock tries again. + busyPoll time.Duration + // wait makes the waits of a new connection end with their context. + wait func(conn driver.Conn) (driver.Conn, error) // logger receives the skipped optimize notes. logger *slog.Logger } @@ -49,31 +53,132 @@ func newConnector(drv *modernc.Driver, path string, opts Options) *connector { if logger == nil { logger = slog.New(slog.DiscardHandler) } - return &connector{ - driver: drv, - open: drv.Open, - name: connectionString(path, opts), - path: path, - restore: "PRAGMA busy_timeout=" + strconv.FormatInt(opts.BusyTimeout.Milliseconds(), 10), - logger: logger, + c := &connector{ + driver: drv, + open: drv.Open, + opening: make(chan struct{}, 1), + name: connectionString(path, opts), + path: path, + busyTimeout: opts.BusyTimeout, + busyPoll: opts.BusyPoll, + logger: logger, } + c.wait = c.waitOn + return c } -// Connect opens one connection to the file, one open at a time, optimizes it and passes it through the path's seam. +// Connect opens one connection, waiting while busy, optimizes it, ends its waits with ctx and passes it to the seam. func (c *connector) Connect(ctx context.Context) (driver.Conn, error) { if err := ctx.Err(); err != nil { return nil, err } - c.opening.Lock() - conn, err := c.open(c.name) - c.opening.Unlock() + conn, err := c.openWaiting(ctx) if err != nil { return nil, err } if err := c.optimize(ctx, conn.(driver.ExecerContext)); err != nil { return nil, errors.Join(err, conn.Close()) } - return seam.Wrap(c.path, conn), nil + waiting, err := c.wait(conn) + if err != nil { + return nil, errors.Join(err, conn.Close()) + } + return seam.Wrap(c.path, waiting), nil +} + +// openWaiting opens one connection, trying again every busy poll while busy, until the busy timeout passes or ctx ends. +func (c *connector) openWaiting(ctx context.Context) (driver.Conn, error) { + deadline := time.NewTimer(c.busyTimeout) + defer deadline.Stop() + poll := time.NewTicker(c.busyPoll) + defer poll.Stop() + var busy error + for { + got, answered := c.openOnce(ctx) + if !answered { + return nil, gaveUp(ctx, busy) + } + if got.err == nil { + return got.conn, nil + } + if !errors.Is(Classify(got.err), dbkit.ErrBusy) { + return nil, got.err + } + busy = got.err + poll.Reset(c.busyPoll) + select { + case <-ctx.Done(): + return nil, ended(ctx, busy) + case <-deadline.C: + return nil, busy + case <-poll.C: + } + } +} + +// openAnswer is what one try of open answered. +type openAnswer struct { + // conn is the connection the try opened, nil after an error. + conn driver.Conn + // err is the error of the try. + err error +} + +// openOnce runs one try of open in the opening slot and reports false when ctx ended before the try answered. +func (c *connector) openOnce(ctx context.Context) (openAnswer, bool) { + if !c.takeSlot(ctx) { + return openAnswer{}, false + } + answer := make(chan openAnswer, 1) + go c.try(answer) + select { + case got := <-answer: + <-c.opening + return got, true + case <-ctx.Done(): + go c.leave(answer) + return openAnswer{}, false + } +} + +// takeSlot takes the opening slot at once when it is free, or waits for it, and reports false when ctx ended first. +func (c *connector) takeSlot(ctx context.Context) bool { + select { + case c.opening <- struct{}{}: + return true + default: + } + select { + case c.opening <- struct{}{}: + return true + case <-ctx.Done(): + return false + } +} + +// try sends on answer what one call of open answers. +func (c *connector) try(answer chan<- openAnswer) { + conn, err := c.open(c.name) + answer <- openAnswer{conn: conn, err: err} +} + +// leave closes the connection a given-up try opened once it answers on answer, then frees the opening slot. +func (c *connector) leave(answer <-chan openAnswer) { + if got := <-answer; got.err == nil { + _ = got.conn.Close() + } + <-c.opening +} + +// gaveUp returns the error of an open whose context ended before its try answered, marking the last busy error if any. +func gaveUp(ctx context.Context, busy error) error { + if busy != nil { + return ended(ctx, busy) + } + if cause := context.Cause(ctx); !errors.Is(cause, ctx.Err()) { + return fmt.Errorf("dbkit: open the database: %w: %w", cause, ctx.Err()) + } + return fmt.Errorf("dbkit: open the database: %w", ctx.Err()) } // Driver returns the private driver value of the handle. @@ -81,15 +186,12 @@ func (c *connector) Driver() driver.Driver { return c.driver } -// optimize runs PRAGMA optimize on conn and skips a busy answer with a note. +// optimize lowers the busy timeout of conn, runs PRAGMA optimize and skips a busy answer with a note. func (c *connector) optimize(ctx context.Context, conn driver.ExecerContext) error { _, err := conn.ExecContext(ctx, lowerBusyTimeout, nil) if err == nil { _, err = conn.ExecContext(ctx, optimizeStatement, nil) } - if _, restoreErr := conn.ExecContext(context.WithoutCancel(ctx), c.restore, nil); restoreErr != nil { - return errors.Join(err, restoreErr) - } if errors.Is(Classify(err), dbkit.ErrBusy) { c.logger.InfoContext(ctx, optimizeSkipped, slog.String("path", c.path), slog.Any("error", err)) return nil diff --git a/dbkit/sqlite/connector_internal_test.go b/dbkit/sqlite/connector_internal_test.go index 65e0c73..ec5bb77 100644 --- a/dbkit/sqlite/connector_internal_test.go +++ b/dbkit/sqlite/connector_internal_test.go @@ -6,11 +6,19 @@ import ( "context" "database/sql/driver" "errors" + "math" + "path/filepath" + "runtime" "slices" + "strings" + "sync" + "sync/atomic" "testing" "time" modernc "modernc.org/sqlite" + + "github.com/gopherium/framework/dbkit" ) const ( @@ -18,8 +26,30 @@ const ( internalLower = "PRAGMA busy_timeout=0" // internalOptimize is the optimize statement every new connection runs. internalOptimize = "PRAGMA optimize=0x10002" - // internalRestore is the statement that sets internalOptions' busy timeout back. - internalRestore = "PRAGMA busy_timeout=1500" + // internalBusyPoll is the busy poll of internalOptions, the smallest Open accepts. + internalBusyPoll = time.Millisecond + // internalLongBusyTimeout is a busy timeout no internal test waits out. + internalLongBusyTimeout = 20017 * time.Millisecond + // internalShortBusyTimeout is a busy timeout a test waits out on purpose. + internalShortBusyTimeout = 50 * time.Millisecond + // internalShortBusyPoll is the busy poll of a test that waits out internalShortBusyTimeout. + internalShortBusyPoll = 5 * time.Millisecond + // internalUnreachedWait is a busy timeout and busy poll no test reaches. + internalUnreachedWait = time.Hour + // internalAtOnce is how long a test waits for a call that must return at once. + internalAtOnce = 10 * time.Second + // internalBusyOpens is how many first opens answer busy before an open answers a connection. + internalBusyOpens = 2 + // internalSlowTries is how many first opens a test runs whose every try lasts the whole busy poll. + internalSlowTries = 20 + // internalCallers is how many goroutines call Connect on one connector at once. + internalCallers = 8 + // internalCallsEach is how many times each of internalCallers calls Connect. + internalCallsEach = 50 + // internalOpenEnded starts the error of an open whose context ended before any try answered busy. + internalOpenEnded = "dbkit: open the database: " + // internalFreeTakes is how many times a test takes the free opening slot under an ended context. + internalFreeTakes = 64 ) var ( @@ -27,12 +57,14 @@ var ( errStepFailed = errors.New("connector test: the statement failed on purpose") // errCloseFailed is the error the fake connection's Close answers with. errCloseFailed = errors.New("connector test: the close failed on purpose") + // errEndCause is the cause a test ends a context with. + errEndCause = errors.New("connector test: the context ended with this cause on purpose") ) // internalOptions returns options with every required value set and Create on. func internalOptions() Options { - return Options{BusyTimeout: 1500 * time.Millisecond, CacheSize: 1, MaxConns: 2, Synchronous: SynchronousNormal, - Create: true} + return Options{BusyTimeout: 1500 * time.Millisecond, BusyPoll: internalBusyPoll, CacheSize: 1, MaxConns: 2, + Synchronous: SynchronousNormal, Create: true} } // internalConn is a connection that records its statements and fails the chosen ones. @@ -41,12 +73,8 @@ type internalConn struct { fail map[string]error // closeErr is the error Close answers with. closeErr error - // after runs after each statement with its text. - after func(query string) // ran lists every statement in the order it ran. ran []string - // ended lists every statement that ran on an ended context. - ended []string // closed reports that Close ran. closed bool } @@ -68,20 +96,19 @@ func (c *internalConn) Begin() (driver.Tx, error) { } // ExecContext records query and answers the error chosen for it. -func (c *internalConn) ExecContext(ctx context.Context, query string, _ []driver.NamedValue) (driver.Result, error) { +func (c *internalConn) ExecContext(_ context.Context, query string, _ []driver.NamedValue) (driver.Result, error) { c.ran = append(c.ran, query) - if ctx.Err() != nil { - c.ended = append(c.ended, query) - } - if c.after != nil { - c.after(query) - } if err := c.fail[query]; err != nil { return nil, err } return driver.RowsAffected(0), nil } +// internalNoWait returns conn as it is, in place of the busy wait a connection the driver opened gets. +func internalNoWait(conn driver.Conn) (driver.Conn, error) { + return conn, nil +} + // internalConnector returns a connector that opens conn, or fails with openErr, and counts its opens. func internalConnector(conn driver.Conn, openErr error) (*connector, *int) { c := newConnector(&modernc.Driver{}, "/srv/site/site.db", internalOptions()) @@ -90,6 +117,7 @@ func internalConnector(conn driver.Conn, openErr error) (*connector, *int) { opened++ return conn, openErr } + c.wait = internalNoWait return c, &opened } @@ -101,6 +129,212 @@ func internalMustRun(t *testing.T, conn *internalConn, want ...string) { } } +// internalPlainConn opens a driver connection with no rule to the file at path and closes it when the test ends. +func internalPlainConn(t *testing.T, path string) driver.ExecerContext { + t.Helper() + conn, err := (&modernc.Driver{}).Open(path) + if err != nil { + t.Fatalf("Open(%s) error = %v, want nil", path, err) + } + t.Cleanup(func() { _ = conn.Close() }) + return conn.(driver.ExecerContext) +} + +// internalBusyError returns the error SQLite answers a second writer while a first one holds the write lock. +func internalBusyError(t *testing.T) error { + t.Helper() + if err := ofdOnce(); err != nil { + t.Fatalf("ofdOnce() error = %v, want nil", err) + } + path := filepath.Join(t.TempDir(), "site.db") + if _, err := internalPlainConn(t, path).ExecContext(t.Context(), "BEGIN IMMEDIATE", nil); err != nil { + t.Fatalf("BEGIN IMMEDIATE error = %v, want nil", err) + } + _, err := internalPlainConn(t, path).ExecContext(t.Context(), "BEGIN IMMEDIATE", nil) + if !errors.Is(Classify(err), dbkit.ErrBusy) { + t.Fatalf("a second BEGIN IMMEDIATE error = %v, want busy", err) + } + return err +} + +// internalWaiting returns a connector with the busy timeout and busy poll given that opens through open. +func internalWaiting(timeout, poll time.Duration, open func(name string) (driver.Conn, error)) *connector { + opts := internalOptions() + opts.BusyTimeout, opts.BusyPoll = timeout, poll + c := newConnector(&modernc.Driver{}, "/srv/site/site.db", opts) + c.open = open + c.wait = internalNoWait + return c +} + +// internalBusyFirst returns an open that answers busy to its first n calls and conn after, and its count of calls. +func internalBusyFirst(busy error, n int32, conn driver.Conn) (func(string) (driver.Conn, error), *atomic.Int32) { + opened := &atomic.Int32{} + return func(string) (driver.Conn, error) { + if opened.Add(1) <= n { + return nil, busy + } + return conn, nil + }, opened +} + +// internalCall is a call of Connect running in a goroutine. +type internalCall struct { + // cancel cancels the context of the call. + cancel context.CancelFunc + // ended is closed once the call returned. + ended chan struct{} + // conn is the connection of the call, read once ended is closed. + conn driver.Conn + // err is the error of the call, read once ended is closed. + err error +} + +// internalStart runs Connect on c in a goroutine, which the end of the test cancels and waits for. +func internalStart(t *testing.T, c *connector) *internalCall { + t.Helper() + return internalStartUnder(t, c, context.WithCancel) +} + +// internalShortDeadline returns ctx with a deadline after internalShortBusyTimeout. +func internalShortDeadline(ctx context.Context) (context.Context, context.CancelFunc) { + return context.WithTimeout(ctx, internalShortBusyTimeout) +} + +// internalCauseOf returns ctx with a cancel that ends it with errEndCause as its cause. +func internalCauseOf(ctx context.Context) (context.Context, context.CancelFunc) { + ctx, cancel := context.WithCancelCause(ctx) + return ctx, func() { cancel(errEndCause) } +} + +// internalStartUnder runs Connect on c in a goroutine under the context end makes, which the end of the test cancels. +func internalStartUnder(t *testing.T, c *connector, + end func(context.Context) (context.Context, context.CancelFunc)) *internalCall { + t.Helper() + ctx, cancel := end(t.Context()) + call := &internalCall{cancel: cancel, ended: make(chan struct{})} + go func() { + defer close(call.ended) + call.conn, call.err = c.Connect(ctx) + }() + t.Cleanup(func() { + cancel() + select { + case <-call.ended: + case <-time.After(internalAtOnce): + t.Errorf("Connect() still runs %v after its context ended, want it to end with its context", internalAtOnce) + } + }) + return call +} + +// within returns what the call answered, and fails the test when the call still runs after internalAtOnce. +func (c *internalCall) within(t *testing.T) (driver.Conn, error) { + t.Helper() + select { + case <-c.ended: + return c.conn, c.err + case <-time.After(internalAtOnce): + t.Fatalf("Connect() still runs after %v, want it to return at once", internalAtOnce) + return nil, nil + } +} + +// mustWait fails the test when the call has already returned. +func (c *internalCall) mustWait(t *testing.T) { + t.Helper() + select { + case <-c.ended: + t.Errorf("Connect() returned %v, %v, want it still waiting", c.conn, c.err) + default: + } +} + +// internalMustReceive fails the test when ch receives nothing within internalAtOnce, naming the signal it wants. +func internalMustReceive(t *testing.T, ch <-chan struct{}, want string) { + t.Helper() + select { + case <-ch: + case <-time.After(internalAtOnce): + t.Fatalf("no signal after %v, want %s", internalAtOnce, want) + } +} + +// internalClosing is a connection that reports its Close on a channel. +type internalClosing struct { + *internalConn + // closed is closed once Close ran. + closed chan struct{} +} + +// Close closes the channel closed. +func (c *internalClosing) Close() error { + close(c.closed) + return nil +} + +// internalHeld is an open whose every call waits until the test releases it. +type internalHeld struct { + // conn is what the first call answers once released. + conn driver.Conn + // err is the error the first call answers once released. + err error + // started receives once the first call began. + started chan struct{} + // released is closed once the test lets every call answer. + released chan struct{} + // release closes released once. + release func() + // calls counts the calls that began. + calls atomic.Int32 +} + +// internalHold returns a held open whose first call answers conn and err, which the end of the test releases. +func internalHold(t *testing.T, conn driver.Conn, err error) *internalHeld { + t.Helper() + released := make(chan struct{}) + h := &internalHeld{conn: conn, err: err, started: make(chan struct{}, 1), released: released, + release: sync.OnceFunc(func() { close(released) })} + t.Cleanup(h.release) + return h +} + +// open counts its call, signals the first one and answers once released, a fresh connection after the first call. +func (h *internalHeld) open(string) (driver.Conn, error) { + if h.calls.Add(1) != 1 { + <-h.released + return &internalConn{}, nil + } + h.started <- struct{}{} + <-h.released + return h.conn, h.err +} + +// internalGiveUp starts Connect on c, cancels it once the first try of held began and returns its error. +func internalGiveUp(t *testing.T, c *connector, held *internalHeld) error { + t.Helper() + call := internalStart(t, c) + internalMustReceive(t, held.started, "the first try to begin") + call.cancel() + conn, err := call.within(t) + if conn != nil || !errors.Is(err, context.Canceled) { + t.Errorf("Connect() cancelled during its try = %v, %v, want nil and context.Canceled at once", conn, err) + } + return err +} + +// internalMustOpenAgain fails the test unless a new Connect on c opens through the second call of held. +func internalMustOpenAgain(t *testing.T, c *connector, held *internalHeld) { + t.Helper() + conn, err := internalStart(t, c).within(t) + if err != nil || conn == nil { + t.Errorf("Connect() after the given-up try = %v, %v, want a connection", conn, err) + } + if got := held.calls.Load(); got != 2 { + t.Errorf("opens = %d after the given-up try, want the second open to answer the new Connect", got) + } +} + func TestConnectRefusesAnEndedContext(t *testing.T) { t.Parallel() @@ -110,21 +344,142 @@ func TestConnectRefusesAnEndedContext(t *testing.T) { conn, err := c.Connect(ctx) - if !errors.Is(err, context.Canceled) || conn != nil || *opened != 0 { - t.Errorf("Connect() = %v, %v after %d opens, want nil, context.Canceled and no open", conn, err, *opened) + if err != context.Canceled || conn != nil { + t.Errorf("Connect() = %v, %v, want nil and context.Canceled itself", conn, err) + } + c.opening <- struct{}{} + if *opened != 0 { + t.Errorf("opens = %d once the opening slot is free, want no open", *opened) } } func TestConnectAnswersTheOpenError(t *testing.T) { t.Parallel() - c, _ := internalConnector(nil, errStepFailed) + c, opened := internalConnector(nil, errStepFailed) conn, err := c.Connect(t.Context()) if !errors.Is(err, errStepFailed) || conn != nil { t.Errorf("Connect() = %v, %v, want nil and the open error", conn, err) } + if *opened != 1 { + t.Errorf("opens = %d, want one open for an error that is not busy", *opened) + } +} + +func TestConnectTriesABusyOpenAgain(t *testing.T) { + t.Parallel() + + fake := &internalConn{} + open, opened := internalBusyFirst(internalBusyError(t), internalBusyOpens, fake) + c := internalWaiting(internalLongBusyTimeout, internalBusyPoll, open) + + conn, err := c.Connect(t.Context()) + + if err != nil || conn != fake { + t.Errorf("Connect() = %v, %v, want the connection of the open after the busy ones", conn, err) + } + if got := opened.Load(); got != internalBusyOpens+1 { + t.Errorf("opens = %d, want %d", got, internalBusyOpens+1) + } +} + +func TestABusyOpenStopsAtTheBusyTimeout(t *testing.T) { + t.Parallel() + + busy := internalBusyError(t) + open, _ := internalBusyFirst(busy, math.MaxInt32, nil) + c := internalWaiting(internalShortBusyTimeout, internalShortBusyPoll, open) + began := time.Now() + + conn, err := c.Connect(t.Context()) + + waited := time.Since(began) + if !errors.Is(Classify(err), dbkit.ErrBusy) || conn != nil { + t.Errorf("Connect() = %v, %v, want nil and the busy error", conn, err) + } + if !errors.Is(err, busy) { + t.Errorf("Connect() error = %v, want the last busy error %v", err, busy) + } + if waited < internalShortBusyTimeout { + t.Errorf("Connect() answered after %v, want at least the busy timeout %v", waited, internalShortBusyTimeout) + } +} + +func TestABusyOpenEndsWithItsContext(t *testing.T) { + t.Parallel() + + busy := internalBusyError(t) + ctx, cancel := context.WithCancel(t.Context()) + var opened atomic.Int32 + c := internalWaiting(internalLongBusyTimeout, internalBusyPoll, func(string) (driver.Conn, error) { + if opened.Add(1) == internalBusyOpens { + cancel() + } + return nil, busy + }) + began := time.Now() + + conn, err := c.Connect(ctx) + + waited := time.Since(began) + if !errors.Is(err, context.Canceled) || conn != nil { + t.Errorf("Connect() = %v, %v, want nil and an error matching context.Canceled", conn, err) + } + if !errors.Is(Classify(err), dbkit.ErrBusy) { + t.Errorf("Classify(%v) is not busy, want the busy error kept", err) + } + if waited > internalLongBusyTimeout/10 { + t.Errorf("Connect() answered after %v, want within a tenth of the busy timeout %v", waited, + internalLongBusyTimeout) + } +} + +func TestAWaitingOpenLeavesTheSlotFree(t *testing.T) { + t.Parallel() + + busy, fake := internalBusyError(t), &internalConn{} + waiting := make(chan struct{}, 1) + var opened atomic.Int32 + c := internalWaiting(internalUnreachedWait, internalUnreachedWait, func(string) (driver.Conn, error) { + if opened.Add(1) == 1 { + waiting <- struct{}{} + return nil, busy + } + return fake, nil + }) + first := internalStart(t, c) + internalMustReceive(t, waiting, "the open to signal its call") + + conn, err := internalStart(t, c).within(t) + + if err != nil || conn != fake { + t.Errorf("the second Connect() = %v, %v, want the open connection", conn, err) + } + first.mustWait(t) + first.cancel() + if _, err := first.within(t); !errors.Is(err, context.Canceled) { + t.Errorf("the first Connect() error = %v, want context.Canceled once cancelled", err) + } +} + +func TestEachOpenTryWaitsOneBusyPoll(t *testing.T) { + t.Parallel() + + var name string + c := internalWaiting(internalLongBusyTimeout, internalShortBusyPoll, func(n string) (driver.Conn, error) { + name = n + return &internalConn{}, nil + }) + + if _, err := c.Connect(t.Context()); err != nil { + t.Fatalf("Connect() error = %v, want nil", err) + } + + if want := "&_busy_timeout=5&"; !strings.Contains(name, want) { + t.Errorf("the open's name = %q, want it to hold %q, the busy poll", name, want) + } } func TestTheOptimizeStepRunsInOrder(t *testing.T) { @@ -132,38 +487,62 @@ func TestTheOptimizeStepRunsInOrder(t *testing.T) { fake := &internalConn{} c, _ := internalConnector(fake, nil) + var before [][]string + c.wait = func(conn driver.Conn) (driver.Conn, error) { + before = append(before, slices.Clone(fake.ran)) + return conn, nil + } conn, err := c.Connect(t.Context()) - if err != nil || conn != fake || fake.closed { - t.Errorf("Connect() = %v, %v, closed %v, want the open connection", conn, err, fake.closed) + if err != nil || conn != fake { + t.Errorf("Connect() = %v, %v, want the open connection", conn, err) + } + if fake.closed { + t.Errorf("closed = %v, want the open connection left open", fake.closed) + } + internalMustRun(t, fake, internalLower, internalOptimize) + if want := []string{internalLower, internalOptimize}; len(before) != 1 || !slices.Equal(before[0], want) { + t.Errorf("statements before each busy wait = %q, want one busy wait after %q", before, want) } - internalMustRun(t, fake, internalLower, internalOptimize, internalRestore) +} + +// internalWaitFails answers errStepFailed in place of the busy wait of a connection. +func internalWaitFails(driver.Conn) (driver.Conn, error) { + return nil, errStepFailed } func TestAFailedStepClosesTheConnection(t *testing.T) { t.Parallel() cases := []struct { - name string + // name names the step that fails. + name string + // failed is the statement that fails, empty when no statement does. failed string - ran []string + // wait stands in for the busy wait of the connection. + wait func(driver.Conn) (driver.Conn, error) + // ran lists the statements the connection runs. + ran []string }{ - {"the lowering", internalLower, []string{internalLower, internalRestore}}, - {"the optimize", internalOptimize, []string{internalLower, internalOptimize, internalRestore}}, - {"the restore", internalRestore, []string{internalLower, internalOptimize, internalRestore}}, + {"the lowering", internalLower, internalNoWait, []string{internalLower}}, + {"the optimize", internalOptimize, internalNoWait, []string{internalLower, internalOptimize}}, + {"the busy wait", "", internalWaitFails, []string{internalLower, internalOptimize}}, } for _, c := range cases { t.Run(c.name, func(t *testing.T) { t.Parallel() fake := &internalConn{fail: map[string]error{c.failed: errStepFailed}} connector, _ := internalConnector(fake, nil) + connector.wait = c.wait conn, err := connector.Connect(t.Context()) - if !errors.Is(err, errStepFailed) || conn != nil || !fake.closed { - t.Errorf("Connect() = %v, %v, closed %v, want nil, the step's error and a closed connection", - conn, err, fake.closed) + if !errors.Is(err, errStepFailed) || conn != nil { + t.Errorf("Connect() = %v, %v, want nil and the step's error", conn, err) + } + if !fake.closed { + t.Errorf("closed = %v, want the connection closed", fake.closed) } internalMustRun(t, fake, c.ran...) }) @@ -183,23 +562,271 @@ func TestAFailedCloseJoinsTheStepError(t *testing.T) { } } -func TestTheRestoreOutlivesAnEndedContext(t *testing.T) { +func TestConnectAnswersAnErrorThatOnlyReadsBusy(t *testing.T) { t.Parallel() - ctx, cancel := context.WithCancel(t.Context()) - fake := &internalConn{after: func(query string) { - if query == internalOptimize { + text := errors.New(internalBusyError(t).Error()) + c, opened := internalConnector(nil, text) + + conn, err := c.Connect(t.Context()) + + if !errors.Is(err, text) || conn != nil { + t.Errorf("Connect() = %v, %v, want nil and the open error", conn, err) + } + if *opened != 1 { + t.Errorf("opens = %d, want one open for an error whose text alone reads busy", *opened) + } +} + +func TestABusyOpenTriesNoLongerThanTheBusyTimeout(t *testing.T) { + t.Parallel() + + busy := internalBusyError(t) + open, opened := internalBusyFirst(busy, math.MaxInt32, nil) + c := internalWaiting(internalShortBusyTimeout, internalShortBusyPoll, open) + + _, err := c.Connect(t.Context()) + + if most := int32(internalShortBusyTimeout/internalShortBusyPoll) + 2; opened.Load() > most || err != busy { + t.Errorf("Connect() = %v after %d opens, want the last busy error itself after at most %d", err, + opened.Load(), most) + } +} + +func TestASlowBusyOpenStopsAfterTheTryAtTheBusyTimeout(t *testing.T) { + t.Parallel() + + busy := internalBusyError(t) + for range internalSlowTries { + var opened atomic.Int32 + c := internalWaiting(internalShortBusyTimeout, internalShortBusyTimeout, func(string) (driver.Conn, error) { + opened.Add(1) + time.Sleep(internalShortBusyTimeout) + return nil, busy + }) + + if _, err := c.Connect(t.Context()); err != busy || opened.Load() != 1 { + t.Fatalf("Connect() = %v after %d opens, want the busy error after the one open the busy timeout ended", + err, opened.Load()) + } + } +} + +func TestASlowBusyOpenStopsDuringTheTryItsContextEnded(t *testing.T) { + t.Parallel() + + busy := internalBusyError(t) + for range internalSlowTries { + ctx, cancel := context.WithCancel(t.Context()) + var opened atomic.Int32 + c := internalWaiting(internalUnreachedWait, internalShortBusyPoll, func(string) (driver.Conn, error) { + opened.Add(1) cancel() + time.Sleep(internalShortBusyPoll) + return nil, busy + }) + + if _, err := c.Connect(ctx); !errors.Is(err, context.Canceled) || opened.Load() != 1 { + t.Fatalf("Connect() = %v after %d opens, want context.Canceled after the one open its context ended", + err, opened.Load()) } - }} - c, _ := internalConnector(fake, nil) + } +} - if _, err := c.Connect(ctx); err != nil { - t.Fatalf("Connect() error = %v, want nil", err) +func TestAHandleRunsOneOpenAtATime(t *testing.T) { + t.Parallel() + + var running atomic.Int32 + var overlapped atomic.Bool + c := internalWaiting(internalLongBusyTimeout, internalBusyPoll, func(string) (driver.Conn, error) { + if running.Add(1) > 1 { + overlapped.Store(true) + } + runtime.Gosched() + running.Add(-1) + return &internalConn{}, nil + }) + var wg sync.WaitGroup + for range internalCallers { + wg.Go(func() { + for range internalCallsEach { + _, _ = c.Connect(t.Context()) + } + }) + } + wg.Wait() + + if overlapped.Load() { + t.Error("two opens of one handle ran at once, want one at a time") + } +} + +func TestAnOpenEndsWithItsContextWhileItsTryRuns(t *testing.T) { + t.Parallel() + + fake := &internalClosing{internalConn: &internalConn{}, closed: make(chan struct{})} + held := internalHold(t, fake, nil) + c := internalWaiting(internalUnreachedWait, internalUnreachedWait, held.open) + + err := internalGiveUp(t, c, held) + + if want := internalOpenEnded + context.Canceled.Error(); err == nil || err.Error() != want { + t.Errorf("Connect() error = %v, want %q", err, want) } + held.release() + internalMustReceive(t, fake.closed, "the given-up try's connection closed") + internalMustOpenAgain(t, c, held) +} + +func TestAnOpenWaitingForTheSlotEndsWithItsOwnContext(t *testing.T) { + t.Parallel() + + held := internalHold(t, &internalConn{}, nil) + c := internalWaiting(internalUnreachedWait, internalUnreachedWait, held.open) + _ = internalGiveUp(t, c, held) + + conn, err := internalStartUnder(t, c, internalShortDeadline).within(t) + + if conn != nil || !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("the second Connect() = %v, %v, want nil and context.DeadlineExceeded", conn, err) + } + if want := internalOpenEnded + context.DeadlineExceeded.Error(); err == nil || err.Error() != want { + t.Errorf("the second Connect() error = %v, want %q", err, want) + } + if got := held.calls.Load(); got != 1 { + t.Errorf("opens = %d, want the given-up try alone while it runs", got) + } +} + +func TestOpensEndingDuringOneTryLeaveOnlyThatTryRunning(t *testing.T) { + t.Parallel() + + held := internalHold(t, &internalConn{}, nil) + c := internalWaiting(internalUnreachedWait, internalUnreachedWait, held.open) + _ = internalGiveUp(t, c, held) + calls := make([]*internalCall, internalCallers) + for i := range calls { + calls[i] = internalStartUnder(t, c, internalShortDeadline) + } + + for i, call := range calls { + if _, err := call.within(t); !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("Connect() %d error = %v, want context.DeadlineExceeded", i, err) + } + } + + if got := held.calls.Load(); got != 1 { + t.Errorf("opens = %d after %d Connects ended, want the given-up try alone", got, internalCallers) + } + held.release() + internalMustOpenAgain(t, c, held) +} + +func TestATryThatAnswersBusyAfterItsContextEndedLeavesNothingOpen(t *testing.T) { + t.Parallel() + + held := internalHold(t, nil, internalBusyError(t)) + c := internalWaiting(internalUnreachedWait, internalUnreachedWait, held.open) + + err := internalGiveUp(t, c, held) + + if errors.Is(Classify(err), dbkit.ErrBusy) { + t.Errorf("Connect() error = %v, want the context's alone, since no try answered busy before it ended", err) + } + held.release() + internalMustOpenAgain(t, c, held) +} + +func TestAnOpenEndingDuringItsFirstTryMatchesTheContextCause(t *testing.T) { + t.Parallel() + + held := internalHold(t, &internalConn{}, nil) + c := internalWaiting(internalUnreachedWait, internalUnreachedWait, held.open) + call := internalStartUnder(t, c, internalCauseOf) + internalMustReceive(t, held.started, "the first try to begin") + + call.cancel() + + conn, err := call.within(t) + if conn != nil || !errors.Is(err, errEndCause) { + t.Errorf("Connect() = %v, %v, want nil and an error matching the context's cause %q", conn, err, errEndCause) + } + if !errors.Is(err, context.Canceled) { + t.Errorf("Connect() error = %v, want it to match context.Canceled as the busy shape does", err) + } +} + +func TestAnOpenEndingDuringALaterTryMarksTheLastBusyError(t *testing.T) { + t.Parallel() + + busy := internalBusyError(t) + held := internalHold(t, &internalConn{}, nil) + var opened atomic.Int32 + c := internalWaiting(internalUnreachedWait, internalBusyPoll, func(name string) (driver.Conn, error) { + if opened.Add(1) == 1 { + return nil, busy + } + return held.open(name) + }) + + err := internalGiveUp(t, c, held) + + if want := busy.Error() + ": " + context.Canceled.Error(); !errors.Is(err, busy) || err.Error() != want { + t.Errorf("Connect() error = %v, want %q, the last busy error marked with the context's", err, want) + } +} + +func TestAFreeSlotIsTakenEvenAfterTheContextEnded(t *testing.T) { + t.Parallel() + + c, _ := internalConnector(&internalConn{}, nil) + ctx, cancel := context.WithCancel(t.Context()) + cancel() + + for take := range internalFreeTakes { + if !c.takeSlot(ctx) { + t.Fatalf("takeSlot() %d under an ended context = false, want the free slot taken at once", take) + } + <-c.opening + } +} + +// internalSlowClose is a connection whose Close waits until the test lets it finish. +type internalSlowClose struct { + // internalConn gives the connection the methods of driver.Conn and ExecContext. + *internalConn + // closing receives once Close began. + closing chan struct{} + // finished is closed once the test lets Close return. + finished chan struct{} +} + +// Close signals closing and returns once finished is closed. +func (c *internalSlowClose) Close() error { + c.closing <- struct{}{} + <-c.finished + return nil +} + +func TestAGivenUpTryFreesTheSlotOnlyOnceItsConnectionClosed(t *testing.T) { + t.Parallel() - internalMustRun(t, fake, internalLower, internalOptimize, internalRestore) - if len(fake.ended) != 0 { - t.Errorf("statements on an ended context = %q, want the restore run on a live one", fake.ended) + finished := make(chan struct{}) + finish := sync.OnceFunc(func() { close(finished) }) + t.Cleanup(finish) + fake := &internalSlowClose{internalConn: &internalConn{}, closing: make(chan struct{}, 1), finished: finished} + held := internalHold(t, fake, nil) + c := internalWaiting(internalUnreachedWait, internalUnreachedWait, held.open) + _ = internalGiveUp(t, c, held) + held.release() + internalMustReceive(t, fake.closing, "the given-up try's connection to start closing") + + conn, err := internalStartUnder(t, c, internalShortDeadline).within(t) + + if conn != nil || !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("Connect() while the given-up connection closes = %v, %v, want nil and context.DeadlineExceeded", + conn, err) } + finish() + internalMustOpenAgain(t, c, held) } diff --git a/dbkit/sqlite/connector_test.go b/dbkit/sqlite/connector_test.go index 0875bbb..74a5d63 100644 --- a/dbkit/sqlite/connector_test.go +++ b/dbkit/sqlite/connector_test.go @@ -3,6 +3,7 @@ package sqlite_test import ( + "cmp" "context" "database/sql" "database/sql/driver" @@ -21,6 +22,16 @@ const ( connectorSkipped = "dbkit: PRAGMA optimize skipped on a new connection while the database is busy" // connectorNewFiles is how many new files the test opens a full pool on at once. connectorNewFiles = 100 + // connectorHold is how long a writer holds a new file while a first open must wait for it. + connectorHold = 100 * time.Millisecond + // connectorDeadline is the deadline of a first open behind a writer that never lets go. + connectorDeadline = 100 * time.Millisecond + // connectorWaitDeadline is the deadline of a new connection's wait behind a held writer. + connectorWaitDeadline = 100 * time.Millisecond + // connectorShortBusyTimeout is a busy timeout a first open waits out on purpose. + connectorShortBusyTimeout = 300 * time.Millisecond + // connectorShortBusyPoll is how often a first open that waits out connectorShortBusyTimeout tries again. + connectorShortBusyPoll = 5 * time.Millisecond ) // connectorRecords is a log handler that sends every record it handles to a channel. @@ -66,10 +77,11 @@ func connectorMustSkip(t *testing.T, records <-chan slog.Record, path string) { select { case r := <-records: attrs := connectorAttrs(r) - cause, _ := attrs["error"].Any().(error) - busy := errors.Is(sqlite.Classify(cause), dbkit.ErrBusy) - if r.Message != connectorSkipped || attrs["path"].String() != path || !busy { - t.Errorf("note = %q %v, want %q with the path and the busy error", r.Message, attrs, connectorSkipped) + if r.Message != connectorSkipped || attrs["path"].String() != path { + t.Errorf("note = %q %v, want %q with the path", r.Message, attrs, connectorSkipped) + } + if cause, _ := attrs["error"].Any().(error); !errors.Is(sqlite.Classify(cause), dbkit.ErrBusy) { + t.Errorf("note error = %v, want the busy error", cause) } default: t.Fatal("no note arrived, want the skipped optimize noted") @@ -105,14 +117,29 @@ func TestANewConnectionOpensWhileAWriterHoldsTheLock(t *testing.T) { if took > optionsBusyTimeout/2 { t.Errorf("Conn() took %v, want well inside the busy timeout %v", took, optionsBusyTimeout) } + connectorMustWaitWithItsContext(t, conn) + connectorMustSkip(t, records, path) +} + +// connectorMustWaitWithItsContext fails the test unless conn reads busy_timeout 0 and its wait ends at its deadline. +func connectorMustWaitWithItsContext(t *testing.T, conn *sql.Conn) { + t.Helper() var busy int64 - if err := conn.QueryRowContext(t.Context(), "PRAGMA busy_timeout").Scan(&busy); err != nil { - t.Fatalf("PRAGMA busy_timeout error = %v, want nil", err) + if err := conn.QueryRowContext(t.Context(), "PRAGMA busy_timeout").Scan(&busy); err != nil || busy != 0 { + t.Errorf("PRAGMA busy_timeout = %d, %v, want 0 with dbkit's busy handler in place of SQLite's", busy, err) } - if busy != optionsBusyTimeout.Milliseconds() { - t.Errorf("PRAGMA busy_timeout = %d, want %d set back after the optimize", busy, optionsBusyTimeout.Milliseconds()) + ctx, cancel := context.WithTimeout(t.Context(), connectorWaitDeadline) + defer cancel() + + waited, err := rulesBegin(ctx, conn) + + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("BeginTx() on the new connection error = %v, want one matching context.DeadlineExceeded", err) + } + if waited > optionsBusyTimeout/10 { + t.Errorf("BeginTx() on the new connection answered after %v, want within a tenth of the busy timeout %v", + waited, optionsBusyTimeout) } - connectorMustSkip(t, records, path) } func TestAnOptimizeWithNoWriterLeavesNoNote(t *testing.T) { @@ -187,6 +214,160 @@ func TestTheFirstConnectionsOfANewFileOpenTogether(t *testing.T) { } } +// connectorTwoPoolsAtOnce takes a full pool on each of two handles of one new file at once and returns every error. +func connectorTwoPoolsAtOnce(t *testing.T) error { + t.Helper() + address, _ := testAddress(t) + handles := []*sql.DB{mustOpen(t, address, testOptions()), mustOpen(t, address, testOptions())} + taken := make(chan error, len(handles)) + for _, db := range handles { + go func() { taken <- connectorOpenAtOnce(t, db) }() + } + errs := []error{<-taken, <-taken} + for _, db := range handles { + errs = append(errs, db.Close()) + } + return errors.Join(errs...) +} + +func TestTwoHandlesOpenANewFileTogether(t *testing.T) { + t.Parallel() + + failed := 0 + var first error + for range connectorNewFiles { + if err := connectorTwoPoolsAtOnce(t); err != nil { + failed++ + first = cmp.Or(first, err) + } + } + + if failed != 0 { + t.Errorf("%d of %d new files failed two full pools opened at once, the first with %v, want none", + failed, connectorNewFiles, first) + } +} + +// connectorHoldNewFile returns a handle on a new file and a plain driver connection that ran begin on that file. +func connectorHoldNewFile(t *testing.T, opts sqlite.Options, begin string) (*sql.DB, *sql.Conn) { + t.Helper() + address, path := testAddress(t) + db := mustOpen(t, address, opts) + plain, err := sql.Open("sqlite", path) + if err != nil { + t.Fatalf("sql.Open() error = %v, want nil", err) + } + t.Cleanup(func() { _ = plain.Close() }) + holder, err := plain.Conn(t.Context()) + if err != nil { + t.Fatalf("Conn() error = %v, want nil", err) + } + t.Cleanup(func() { _ = holder.Close() }) + if _, err := holder.ExecContext(t.Context(), begin); err != nil { + t.Fatalf("ExecContext(%q) error = %v, want nil", begin, err) + } + return db, holder +} + +// connectorPing pings db in a goroutine and returns the channel that receives its error. +func connectorPing(t *testing.T, db *sql.DB) <-chan error { + t.Helper() + pinged := make(chan error, 1) + go func() { pinged <- db.PingContext(t.Context()) }() + return pinged +} + +func TestTheFirstOpenOfANewFileWaitsForTheWriter(t *testing.T) { + t.Parallel() + + db, holder := connectorHoldNewFile(t, testOptions(), "BEGIN IMMEDIATE") + began := time.Now() + pinged := connectorPing(t, db) + + select { + case err := <-pinged: + t.Fatalf("PingContext() = %v after %v while the writer holds the file, want it to wait", err, + time.Since(began)) + case <-time.After(connectorHold): + } + if _, err := holder.ExecContext(t.Context(), "ROLLBACK"); err != nil { + t.Fatalf("ROLLBACK error = %v, want nil", err) + } + + if err := <-pinged; err != nil { + t.Fatalf("PingContext() error = %v, want nil once the writer let go", err) + } + var mode string + if err := db.QueryRowContext(t.Context(), "PRAGMA journal_mode").Scan(&mode); err != nil || mode != "wal" { + t.Errorf("PRAGMA journal_mode = %q, %v, want wal", mode, err) + } +} + +func TestTheFirstOpenOfANewFileStopsAtTheBusyTimeout(t *testing.T) { + t.Parallel() + + opts := testOptions() + opts.BusyTimeout, opts.BusyPoll = connectorShortBusyTimeout, connectorShortBusyPoll + db, _ := connectorHoldNewFile(t, opts, "BEGIN IMMEDIATE") + began := time.Now() + + err := db.PingContext(t.Context()) + + waited := time.Since(began) + if !errors.Is(sqlite.Classify(err), dbkit.ErrBusy) { + t.Errorf("PingContext() error = %v, want busy while the writer holds the file", err) + } + if waited < connectorShortBusyTimeout { + t.Errorf("PingContext() answered after %v, want at least the busy timeout %v", waited, connectorShortBusyTimeout) + } +} + +// connectorMustEndAtItsDeadline fails unless a ping ends at its deadline, busy if wantBusy, then opens after holder. +func connectorMustEndAtItsDeadline(t *testing.T, db *sql.DB, holder *sql.Conn, wantBusy bool) { + t.Helper() + ctx, cancel := context.WithTimeout(t.Context(), connectorDeadline) + defer cancel() + began := time.Now() + + err := db.PingContext(ctx) + + waited := time.Since(began) + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("PingContext() error = %v, want one matching context.DeadlineExceeded", err) + } + if busy := errors.Is(sqlite.Classify(err), dbkit.ErrBusy); busy != wantBusy { + t.Errorf("PingContext() error = %v, busy %v, want busy %v", err, busy, wantBusy) + } + if waited > optionsBusyTimeout/10 { + t.Errorf("PingContext() answered after %v, want within a tenth of the busy timeout %v", waited, + optionsBusyTimeout) + } + if _, err := holder.ExecContext(t.Context(), "ROLLBACK"); err != nil { + t.Fatalf("ROLLBACK error = %v, want nil", err) + } + if err := db.PingContext(t.Context()); err != nil { + t.Errorf("PingContext() error = %v once the holder let go, want nil", err) + } +} + +func TestTheFirstOpenEndsWithItsContextBehindAnExclusiveLock(t *testing.T) { + t.Parallel() + + db, holder := connectorHoldNewFile(t, testOptions(), "BEGIN EXCLUSIVE") + + connectorMustEndAtItsDeadline(t, db, holder, true) +} + +func TestTheFirstOpenEndsWithItsContextAtTheLongestBusyPoll(t *testing.T) { + t.Parallel() + + opts := testOptions() + opts.BusyPoll = opts.BusyTimeout + db, holder := connectorHoldNewFile(t, opts, "BEGIN EXCLUSIVE") + + connectorMustEndAtItsDeadline(t, db, holder, false) +} + // connectorWrapped is a connection the test's wrapper returned. type connectorWrapped struct { driver.Conn diff --git a/dbkit/sqlite/connector_unix_test.go b/dbkit/sqlite/connector_unix_test.go new file mode 100644 index 0000000..4bf6dd6 --- /dev/null +++ b/dbkit/sqlite/connector_unix_test.go @@ -0,0 +1,269 @@ +// SPDX-License-Identifier: Apache-2.0 + +//go:build linux || darwin + +package sqlite + +import ( + "context" + "database/sql/driver" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "testing" + "testing/fstest" + "time" + + modernc "modernc.org/sqlite" +) + +const ( + // newFileHolderPath names the switch that makes the test binary hold the write lock of the file at its value. + newFileHolderPath = "DBKIT_TEST_NEW_FILE_HOLDER" + // newFileMigratorPath names the environment switch that makes the test binary migrate the file at its value. + newFileMigratorPath = "DBKIT_TEST_NEW_FILE_MIGRATOR" + // newFileHeld is the line a holder prints once it holds the write lock. + newFileHeld = "dbkit test: the write lock of the new file is held" + // newFileStarted is the line a migrator prints right before it migrates. + newFileStarted = "dbkit test: the migrator starts" + // newFileMigrators is how many processes migrate one new file at once. + newFileMigrators = 8 + // newFileRounds is how many new files the test migrates, one after another. + newFileRounds = 3 + // newFileHold is how long the holder keeps the write lock after the last migrator started. + newFileHold = 300 * time.Millisecond + // newFileBusyTimeout is a busy timeout no migrator waits out. + newFileBusyTimeout = time.Minute + // newFileBusyPoll is how often the first open of a migrator tries again. + newFileBusyPoll = 5 * time.Millisecond + // newFileTable is the version table of the migrators. + newFileTable = "new_file_versions" +) + +// newFileOutput is what a helper process printed, which reports when the process prints its line. +type newFileOutput struct { + // mu guards text. + mu sync.Mutex + // text is everything the process printed. + text strings.Builder + // line is the line the process prints once it is ready. + line string + // printed is closed once text holds line. + printed chan struct{} + // once closes printed once. + once sync.Once +} + +// Write keeps p and closes printed once the output holds the line. +func (o *newFileOutput) Write(p []byte) (int, error) { + o.mu.Lock() + defer o.mu.Unlock() + o.text.Write(p) + if strings.Contains(o.text.String(), o.line+"\n") { + o.once.Do(func() { close(o.printed) }) + } + return len(p), nil +} + +// String returns everything the process printed. +func (o *newFileOutput) String() string { + o.mu.Lock() + defer o.mu.Unlock() + return o.text.String() +} + +// newFileKind is one kind of helper process. +type newFileKind struct { + // test is the test function the process runs. + test string + // env is the environment switch that hands the process the path of the file. + env string + // line is the line the process prints once it is ready. + line string +} + +var ( + // newFileHolder holds the write lock of a new file until its standard input closes. + newFileHolder = newFileKind{test: "TestNewFileHolderProcess", env: newFileHolderPath, line: newFileHeld} + // newFileMigrator migrates a new file. + newFileMigrator = newFileKind{test: "TestNewFileMigratorProcess", env: newFileMigratorPath, line: newFileStarted} +) + +// newFileProcess is a run of the test binary as a helper process. +type newFileProcess struct { + // cmd is the running test binary. + cmd *exec.Cmd + // stdin is the standard input of the process. + stdin io.WriteCloser + // output is what the process printed. + output *newFileOutput + // ended is closed once the process was reaped. + ended chan struct{} + // err is how the process ended, read once ended is closed. + err error +} + +// start runs the helper process on the file at path, which the end of the test kills. +func (k newFileKind) start(t *testing.T, path string) *newFileProcess { + t.Helper() + output := &newFileOutput{line: k.line, printed: make(chan struct{})} + cmd := exec.Command(os.Args[0], "-test.run=^"+k.test+"$") + cmd.Env = append(os.Environ(), k.env+"="+path) + cmd.Stdout, cmd.Stderr = output, output + stdin, err := cmd.StdinPipe() + if err != nil { + t.Fatalf("StdinPipe() error = %v, want nil", err) + } + if err := cmd.Start(); err != nil { + t.Fatalf("Start() error = %v, want nil", err) + } + p := &newFileProcess{cmd: cmd, stdin: stdin, output: output, ended: make(chan struct{})} + go func() { + p.err = cmd.Wait() + close(p.ended) + }() + t.Cleanup(func() { + _ = cmd.Process.Kill() + <-p.ended + }) + return p +} + +// mustPrint fails the test when the process ends before it prints its line. +func (p *newFileProcess) mustPrint(t *testing.T) { + t.Helper() + select { + case <-p.output.printed: + case <-p.ended: + t.Fatalf("the helper process ended with %v before it printed %q\n%s", p.err, p.output.line, p.output) + } +} + +// wait returns how the process ended once it was reaped. +func (p *newFileProcess) wait() error { + <-p.ended + return p.err +} + +// release closes the standard input of the process and returns how the process ended. +func (p *newFileProcess) release() error { + _ = p.stdin.Close() + return p.wait() +} + +// newFileOptions returns the options of a migrator. +func newFileOptions() Options { + opts := internalOptions() + opts.BusyTimeout, opts.BusyPoll = newFileBusyTimeout, newFileBusyPoll + return opts +} + +// newFileMigrations returns the one SQL migration of a migrator. +func newFileMigrations() Migrations { + up := &fstest.MapFile{Data: []byte("-- +goose Up\nCREATE TABLE new_file_rows (a INTEGER);\n")} + return Migrations{Table: newFileTable, FS: fstest.MapFS{"00001_new_file_rows.sql": up}, + LockWait: lockLongWait, LockPoll: lockPoll} +} + +func TestNewFileHolderProcess(t *testing.T) { + path := os.Getenv(newFileHolderPath) + if path == "" { + return + } + if err := ofdOnce(); err != nil { + t.Fatalf("ofdOnce() in the holder error = %v, want nil", err) + } + conn, err := (&modernc.Driver{}).Open(path) + if err != nil { + t.Fatalf("Open() in the holder error = %v, want nil", err) + } + defer func() { _ = conn.Close() }() + plain := conn.(driver.ExecerContext) + if _, err := plain.ExecContext(context.Background(), "BEGIN IMMEDIATE", nil); err != nil { + t.Fatalf("BEGIN IMMEDIATE in the holder error = %v, want nil", err) + } + fmt.Println(newFileHeld) + if _, err := io.Copy(io.Discard, os.Stdin); err != nil { + t.Fatalf("read the standard input of the holder: %v", err) + } + if _, err := plain.ExecContext(context.Background(), "ROLLBACK", nil); err != nil { + t.Fatalf("ROLLBACK in the holder error = %v, want nil", err) + } +} + +func TestNewFileMigratorProcess(t *testing.T) { + path := os.Getenv(newFileMigratorPath) + if path == "" { + return + } + db, err := Open("sqlite:"+path, newFileOptions()) + if err != nil { + t.Fatalf("Open() in a migrator error = %v, want nil", err) + } + defer func() { _ = db.Close() }() + fmt.Println(newFileStarted) + + if err := Migrate(context.Background(), db, newFileMigrations()); err != nil { + t.Fatalf("Migrate() in a migrator error = %v, want nil", err) + } +} + +// newFileStartMigrators starts newFileMigrators migrators on the file at path and returns once each one started. +func newFileStartMigrators(t *testing.T, path string) []*newFileProcess { + t.Helper() + migrators := make([]*newFileProcess, newFileMigrators) + for i := range migrators { + migrators[i] = newFileMigrator.start(t, path) + } + for _, m := range migrators { + m.mustPrint(t) + } + return migrators +} + +// newFileMustApplyOnce fails the test unless the file at path records its one migration as applied once. +func newFileMustApplyOnce(t *testing.T, path string) { + t.Helper() + db, err := Open("sqlite:"+path, internalOptions()) + if err != nil { + t.Fatalf("Open() error = %v, want nil", err) + } + defer func() { _ = db.Close() }() + var n int64 + query := "SELECT count(*) FROM " + newFileTable + " WHERE version_id = 1 AND is_applied" + if err := db.QueryRowContext(t.Context(), query).Scan(&n); err != nil || n != 1 { + t.Errorf("%s records version 1 %d times, %v, want once", newFileTable, n, err) + } +} + +// newFileRound fails the test unless every migrator of a new file ends cleanly while a holder keeps its write lock. +func newFileRound(t *testing.T, round int) { + t.Helper() + path := filepath.Join(t.TempDir(), "site.db") + holder := newFileHolder.start(t, path) + holder.mustPrint(t) + migrators := newFileStartMigrators(t, path) + time.Sleep(newFileHold) + + if err := holder.release(); err != nil { + t.Fatalf("round %d: the holder ended with %v, want a clean end\n%s", round, err, holder.output) + } + for i, m := range migrators { + if err := m.wait(); err != nil { + t.Errorf("round %d: migrator %d ended with %v, want a clean end\n%s", round, i, err, m.output) + } + } + newFileMustApplyOnce(t, path) +} + +func TestMigrateRunsOnANewFileWhileAProcessHoldsTheWriteLock(t *testing.T) { + t.Parallel() + + for round := range newFileRounds { + newFileRound(t, round) + } +} diff --git a/dbkit/sqlite/go.mod b/dbkit/sqlite/go.mod index 3c2e425..42ed200 100644 --- a/dbkit/sqlite/go.mod +++ b/dbkit/sqlite/go.mod @@ -6,6 +6,7 @@ require ( github.com/gopherium/framework/dbkit v0.3.0 github.com/pressly/goose/v3 v3.27.2 golang.org/x/sys v0.48.0 + modernc.org/libc v1.77.1 modernc.org/sqlite v1.60.1 ) @@ -19,7 +20,6 @@ require ( github.com/sethvargo/go-retry v0.3.0 // indirect go.uber.org/multierr v1.11.0 // indirect golang.org/x/sync v0.23.0 // indirect - modernc.org/libc v1.77.1 // indirect modernc.org/mathutil v1.7.1 // indirect modernc.org/memory v1.12.1 // indirect ) diff --git a/dbkit/sqlite/open.go b/dbkit/sqlite/open.go index bdb381c..446a65c 100644 --- a/dbkit/sqlite/open.go +++ b/dbkit/sqlite/open.go @@ -78,7 +78,7 @@ func Open(address string, opts Options) (*sql.DB, error) { func connectionString(path string, opts Options) string { name := fmt.Sprintf("%s?_txlock=immediate&_defensive=1&_busy_timeout=%d&_journal_mode=WAL&_foreign_keys=1"+ "&_synchronous=%s&_time_integer_format=unix_micro&_pragma=temp_store(MEMORY)&_pragma=cache_size(-%d)", - path, opts.BusyTimeout.Milliseconds(), opts.Synchronous.pragma(), opts.CacheSize) + path, opts.BusyPoll.Milliseconds(), opts.Synchronous.pragma(), opts.CacheSize) if opts.JournalSizeLimit != nil { name += fmt.Sprintf("&_pragma=journal_size_limit(%d)", *opts.JournalSizeLimit) } @@ -187,13 +187,17 @@ func checkPlace(path string) error { } return checkFolder(resolved) } - if info, linkErr := os.Lstat(path); errors.Is(err, fs.ErrNotExist) && linkErr == nil && - info.Mode()&fs.ModeSymlink != 0 { + if errors.Is(err, fs.ErrNotExist) && isLink(os.Lstat(path)) { return fmt.Errorf("dbkit: the database path %s is a link to a missing file", path) } return checkFolder(filepath.Dir(path)) } +// isLink reports whether info, which os.Lstat answered with lstatErr, describes a symbolic link. +func isLink(info fs.FileInfo, lstatErr error) bool { + return lstatErr == nil && info.Mode()&fs.ModeSymlink != 0 +} + // checkFile returns the error for a database file that is missing without create, or that cannot be read. func checkFile(path string, create bool) error { _, err := os.Stat(path) diff --git a/dbkit/sqlite/options.go b/dbkit/sqlite/options.go index 9952be4..25f75cd 100644 --- a/dbkit/sqlite/options.go +++ b/dbkit/sqlite/options.go @@ -34,10 +34,12 @@ func (s Synchronous) pragma() string { return "FULL" } -// Options configures one SQLite handle. BusyTimeout, CacheSize, MaxConns and Synchronous are required. +// Options configures one SQLite handle. BusyTimeout, BusyPoll, CacheSize, MaxConns and Synchronous are required. type Options struct { // BusyTimeout is how long a connection waits for a lock, from 1ms to 596h31m23.647s. BusyTimeout time.Duration + // BusyPoll is how often a lock wait tries again and the longest an opening try blocks, from 1ms up to BusyTimeout. + BusyPoll time.Duration // CacheSize is the page cache of each connection in kibibytes, from 1 to 2147483647. CacheSize int // MaxConns caps the open and idle connections, 2 or more, each holding a page cache of CacheSize. @@ -66,11 +68,10 @@ func (o Options) check() error { // checkRequired returns the error for the first required option that is missing or outside its range. func (o Options) checkRequired() error { + if err := o.checkWaits(); err != nil { + return err + } switch { - case o.BusyTimeout < time.Millisecond: - return fmt.Errorf("dbkit: the option BusyTimeout must be 1ms or more, got %v", o.BusyTimeout) - case o.BusyTimeout > maxBusyTimeout: - return fmt.Errorf("dbkit: the option BusyTimeout must be %v or less, got %v", maxBusyTimeout, o.BusyTimeout) case o.CacheSize < 1: return fmt.Errorf("dbkit: the option CacheSize must stand above zero, got %d", o.CacheSize) case o.CacheSize > math.MaxInt32: @@ -84,6 +85,22 @@ func (o Options) checkRequired() error { return nil } +// checkWaits returns the error for a busy timeout or a busy poll that is missing or outside its range. +func (o Options) checkWaits() error { + switch { + case o.BusyTimeout < time.Millisecond: + return fmt.Errorf("dbkit: the option BusyTimeout must be 1ms or more, got %v", o.BusyTimeout) + case o.BusyTimeout > maxBusyTimeout: + return fmt.Errorf("dbkit: the option BusyTimeout must be %v or less, got %v", maxBusyTimeout, o.BusyTimeout) + case o.BusyPoll < time.Millisecond: + return fmt.Errorf("dbkit: the option BusyPoll must be 1ms or more, got %v", o.BusyPoll) + case o.BusyPoll > o.BusyTimeout: + return fmt.Errorf("dbkit: the option BusyPoll must be the BusyTimeout of %v or less, got %v", + o.BusyTimeout, o.BusyPoll) + } + return nil +} + // checkOptional returns the error for the first optional value that is set and refused. func (o Options) checkOptional() error { switch { diff --git a/dbkit/sqlite/options_test.go b/dbkit/sqlite/options_test.go index 889b6ba..47cd6f7 100644 --- a/dbkit/sqlite/options_test.go +++ b/dbkit/sqlite/options_test.go @@ -16,6 +16,8 @@ import ( const ( // optionsBusyTimeout is a busy timeout no uncontended test waits out. optionsBusyTimeout = 20017 * time.Millisecond + // optionsBusyPoll is the busy poll the tests pass, the smallest Open accepts. + optionsBusyPoll = time.Millisecond // optionsCacheSize is the page cache in kibibytes the tests pass. optionsCacheSize = 1536 // optionsMaxConns is the connection cap the tests pass. @@ -30,6 +32,7 @@ const ( func testOptions() sqlite.Options { return sqlite.Options{ BusyTimeout: optionsBusyTimeout, + BusyPoll: optionsBusyPoll, CacheSize: optionsCacheSize, MaxConns: optionsMaxConns, Synchronous: sqlite.SynchronousNormal, @@ -78,6 +81,19 @@ func TestOpenRefusesABadOption(t *testing.T) { "dbkit: the option BusyTimeout must be 596h31m23.647s or less, got 596h31m23.647000001s"}, {"the longest busy timeout", func(o *sqlite.Options) { o.BusyTimeout = math.MaxInt64 }, "dbkit: the option BusyTimeout must be 596h31m23.647s or less, got 2562047h47m16.854775807s"}, + {"a missing busy poll", func(o *sqlite.Options) { o.BusyPoll = 0 }, + "dbkit: the option BusyPoll must be 1ms or more, got 0s"}, + {"a negative busy poll", func(o *sqlite.Options) { o.BusyPoll = -time.Millisecond }, + "dbkit: the option BusyPoll must be 1ms or more, got -1ms"}, + {"a busy poll below a millisecond", func(o *sqlite.Options) { o.BusyPoll = time.Millisecond - 1 }, + "dbkit: the option BusyPoll must be 1ms or more, got 999.999µs"}, + {"a busy poll past the busy timeout", func(o *sqlite.Options) { o.BusyPoll = optionsBusyTimeout + time.Millisecond }, + "dbkit: the option BusyPoll must be the BusyTimeout of 20.017s or less, got 20.018s"}, + {"a busy poll a nanosecond past the busy timeout", + func(o *sqlite.Options) { o.BusyPoll = optionsBusyTimeout + time.Nanosecond }, + "dbkit: the option BusyPoll must be the BusyTimeout of 20.017s or less, got 20.017000001s"}, + {"a missing busy poll and cache size", func(o *sqlite.Options) { o.BusyPoll, o.CacheSize = 0, 0 }, + "dbkit: the option BusyPoll must be 1ms or more, got 0s"}, {"a missing cache size", func(o *sqlite.Options) { o.CacheSize = 0 }, "dbkit: the option CacheSize must stand above zero, got 0"}, {"a negative cache size", func(o *sqlite.Options) { o.CacheSize = -1 }, @@ -112,8 +128,11 @@ func TestOpenRefusesABadOption(t *testing.T) { db, err := sqlite.Open(address, opts) - if err == nil || err.Error() != c.want || db != nil { - t.Errorf("Open() = %v, %v, want nil and %q", db, err, c.want) + if db != nil { + t.Errorf("Open() handle = %v, want nil", db) + } + if err == nil || err.Error() != c.want { + t.Errorf("Open() error = %v, want %q", err, c.want) } mustNotExist(t, path) }) @@ -126,6 +145,7 @@ func TestOpenAcceptsTheSmallestValues(t *testing.T) { address, _ := testAddress(t) opts := testOptions() opts.BusyTimeout = time.Millisecond + opts.BusyPoll = time.Millisecond opts.CacheSize = 1 opts.MaxConns = 2 opts.Synchronous = sqlite.SynchronousFull @@ -143,15 +163,25 @@ func TestOpenAcceptsTheSmallestValues(t *testing.T) { } } +func TestOpenAcceptsABusyPollEqualToTheBusyTimeout(t *testing.T) { + t.Parallel() + + opts := testOptions() + opts.BusyPoll = opts.BusyTimeout + + mustPing(t, rulesOpen(t, opts)) +} + func TestOpenKeepsTheLargestValues(t *testing.T) { t.Parallel() opts := testOptions() opts.BusyTimeout = optionsLargestBusyTimeout + opts.BusyPoll = optionsLargestBusyTimeout opts.CacheSize = math.MaxInt32 busy, cache := rulesOpen(t, opts), rulesOpen(t, opts) - rulesMustAll(t, "PRAGMA busy_timeout", rulesRead[int64](t, busy, "PRAGMA busy_timeout"), math.MaxInt32) + mustPing(t, busy) rulesMustAll(t, "PRAGMA cache_size", rulesRead[int64](t, cache, "PRAGMA cache_size"), -math.MaxInt32) } diff --git a/dbkit/sqlite/rules_test.go b/dbkit/sqlite/rules_test.go index f5aa227..9f68e7e 100644 --- a/dbkit/sqlite/rules_test.go +++ b/dbkit/sqlite/rules_test.go @@ -95,13 +95,51 @@ func TestJournalModeIsWAL(t *testing.T) { rulesMustAll(t, "PRAGMA journal_mode", rulesRead[string](t, db, "PRAGMA journal_mode"), "wal") } -func TestBusyTimeoutIsTheOption(t *testing.T) { +// rulesHeldElsewhere returns a handle with opts on a file whose writer on a second handle holds the lock until cleanup. +func rulesHeldElsewhere(t *testing.T, opts sqlite.Options) *sql.DB { + t.Helper() + address, _ := testAddress(t) + db, holder := mustOpen(t, address, opts), mustOpen(t, address, opts) + rulesExec(t, holder, "CREATE TABLE held_rows (a INTEGER)") + rulesHoldWriter(t, holder, "INSERT INTO held_rows VALUES (1)") + return db +} + +// rulesBegin returns how long BeginTx ran on conn under ctx and its error, rolling back any transaction it got. +func rulesBegin(ctx context.Context, conn *sql.Conn) (time.Duration, error) { + began := time.Now() + tx, err := conn.BeginTx(ctx, nil) + waited := time.Since(began) + if err == nil { + _ = tx.Rollback() + } + return waited, err +} + +func TestEveryConnectionWaitsTheBusyTimeout(t *testing.T) { t.Parallel() - db := rulesOpen(t, testOptions()) + opts := testOptions() + opts.BusyTimeout = rulesShortBusyTimeout + db := rulesHeldElsewhere(t, opts) + + for i, conn := range rulesConns(t, db) { + var timeout int64 + if err := conn.QueryRowContext(t.Context(), "PRAGMA busy_timeout").Scan(&timeout); err != nil || timeout != 0 { + t.Errorf("PRAGMA busy_timeout on connection %d = %d, %v, want 0 with dbkit's busy handler in place of SQLite's", + i, timeout, err) + } + + waited, err := rulesBegin(t.Context(), conn) - rulesMustAll(t, "PRAGMA busy_timeout", rulesRead[int64](t, db, "PRAGMA busy_timeout"), - optionsBusyTimeout.Milliseconds()) + if !errors.Is(sqlite.Classify(err), dbkit.ErrBusy) { + t.Errorf("BeginTx() on connection %d error = %v, want busy while another handle holds the lock", i, err) + } + if waited < rulesShortBusyTimeout { + t.Errorf("BeginTx() on connection %d answered after %v, want at least the busy timeout %v", i, waited, + rulesShortBusyTimeout) + } + } } func TestTempStoreIsMemory(t *testing.T) { diff --git a/dbkit/sqlite/share_test.go b/dbkit/sqlite/share_test.go index 7cbafdd..918e70d 100644 --- a/dbkit/sqlite/share_test.go +++ b/dbkit/sqlite/share_test.go @@ -252,10 +252,17 @@ func TestAStatementPastItsDeadlineStops(t *testing.T) { // shareMustKeepTheRules fails the test unless connection i of a handle from testOptions carries every rule. func shareMustKeepTheRules(t *testing.T, i int, conn *sql.Conn) { + t.Helper() + shareMustReadTheRules(t, i, conn) + shareMustRefuseASchemaEdit(t, i, conn) +} + +// shareMustReadTheRules fails the test unless every pragma of connection i reads the value of its rule. +func shareMustReadTheRules(t *testing.T, i int, conn *sql.Conn) { t.Helper() pragmas := map[string]int64{ "PRAGMA foreign_keys": 1, - "PRAGMA busy_timeout": optionsBusyTimeout.Milliseconds(), + "PRAGMA busy_timeout": 0, "PRAGMA temp_store": 2, "PRAGMA cache_size": -optionsCacheSize, "PRAGMA synchronous": 1, @@ -270,6 +277,11 @@ func shareMustKeepTheRules(t *testing.T, i int, conn *sql.Conn) { if err := conn.QueryRowContext(t.Context(), "PRAGMA journal_mode").Scan(&mode); err != nil || mode != "wal" { t.Errorf("PRAGMA journal_mode on connection %d = %q, %v, want wal", i, mode, err) } +} + +// shareMustRefuseASchemaEdit fails the test unless connection i refuses an edit of the schema table. +func shareMustRefuseASchemaEdit(t *testing.T, i int, conn *sql.Conn) { + t.Helper() if _, err := conn.ExecContext(t.Context(), "PRAGMA writable_schema=ON"); err != nil { t.Fatalf("PRAGMA writable_schema=ON on connection %d error = %v, want nil", i, err) } @@ -279,11 +291,9 @@ func shareMustKeepTheRules(t *testing.T, i int, conn *sql.Conn) { } } -func TestConnectionRulesHoldAfterShareStatements(t *testing.T) { - t.Parallel() - - db := sqlitetest.Open(t, testOptions()) - share := shareOn(t, db, shareOptions("rules-share")) +// shareRulesTable creates the table of the rules test with one row through a transaction of share. +func shareRulesTable(t *testing.T, share *dbkit.Share) { + t.Helper() tx, err := share.Begin(t.Context()) if err != nil { t.Fatalf("Begin() error = %v, want nil", err) @@ -293,8 +303,14 @@ func TestConnectionRulesHoldAfterShareStatements(t *testing.T) { if err := tx.Commit(); err != nil { t.Fatalf("Commit() error = %v, want nil", err) } +} + +// shareQueryOnEvery returns one result set through share on every connection of db. +func shareQueryOnEvery(t *testing.T, db *sql.DB, share *dbkit.Share) []*dbkit.Rows { + t.Helper() held := make([]*dbkit.Rows, optionsMaxConns) for i := range held { + var err error if held[i], err = share.Query(t.Context(), "SELECT a FROM share_rules WHERE a = $1", 1); err != nil { t.Fatalf("Query() error = %v, want nil", err) } @@ -303,11 +319,26 @@ func TestConnectionRulesHoldAfterShareStatements(t *testing.T) { t.Fatalf("Stats().OpenConnections = %d while the share holds %d result sets, want %d", got, len(held), optionsMaxConns) } + return held +} + +// shareCloseAll closes every result set of held and fails the test when one cannot. +func shareCloseAll(t *testing.T, held []*dbkit.Rows) { + t.Helper() for _, rows := range held { if err := rows.Close(); err != nil { t.Fatalf("Close() error = %v, want nil", err) } } +} + +func TestConnectionRulesHoldAfterShareStatements(t *testing.T) { + t.Parallel() + + db := sqlitetest.Open(t, testOptions()) + share := shareOn(t, db, shareOptions("rules-share")) + shareRulesTable(t, share) + shareCloseAll(t, shareQueryOnEvery(t, db, share)) for i, conn := range rulesConns(t, db) { shareMustKeepTheRules(t, i, conn) diff --git a/dbkit/sqlite/snapshot_internal_test.go b/dbkit/sqlite/snapshot_internal_test.go index 0d5400d..318ce93 100644 --- a/dbkit/sqlite/snapshot_internal_test.go +++ b/dbkit/sqlite/snapshot_internal_test.go @@ -6,12 +6,14 @@ import ( "context" "database/sql" "errors" + "fmt" "io/fs" "os" "path/filepath" "slices" "strings" "testing" + "time" modernc "modernc.org/sqlite" sqlite3 "modernc.org/sqlite/lib" @@ -19,8 +21,12 @@ import ( "github.com/gopherium/framework/dbkit" ) -// internalLateTarget is what a file that appears at the target between the copy and the move holds. -const internalLateTarget = "a file written between the copy and the move" +const ( + // internalLateTarget is what a file that appears at the target between the copy and the move holds. + internalLateTarget = "a file written between the copy and the move" + // internalCheckpointDeadline is the deadline of a checkpoint behind a held writer. + internalCheckpointDeadline = 100 * time.Millisecond +) var ( // errSnapshotSyncFailed is the error a chosen sync answers with. @@ -42,7 +48,13 @@ func internalFolder(t *testing.T) string { // internalSnapshotOpen returns a handle on a fresh file holding one row, and an empty folder for the targets. func internalSnapshotOpen(t *testing.T) (*sql.DB, string) { t.Helper() - db, err := Open("sqlite:"+filepath.Join(internalFolder(t), "site.db"), internalOptions()) + return internalSnapshotOpenWith(t, internalOptions()) +} + +// internalSnapshotOpenWith returns a handle with opts on a fresh file holding one row, and an empty target folder. +func internalSnapshotOpenWith(t *testing.T, opts Options) (*sql.DB, string) { + t.Helper() + db, err := Open("sqlite:"+filepath.Join(internalFolder(t), "site.db"), opts) if err != nil { t.Fatalf("Open() error = %v, want nil", err) } @@ -105,6 +117,36 @@ func internalMustFailCheck(t *testing.T, path, want string) { } } +func TestACheckpointEndsWithItsContext(t *testing.T) { + t.Parallel() + + opts := internalOptions() + opts.BusyTimeout = internalLongBusyTimeout + db, _ := internalSnapshotOpenWith(t, opts) + writer, err := db.BeginTx(t.Context(), nil) + if err != nil { + t.Fatalf("BeginTx() error = %v, want nil", err) + } + t.Cleanup(func() { _ = writer.Rollback() }) + if _, err := writer.ExecContext(t.Context(), "INSERT INTO snapshot_rows VALUES (2)"); err != nil { + t.Fatalf("ExecContext() error = %v, want nil", err) + } + ctx, cancel := context.WithTimeout(t.Context(), internalCheckpointDeadline) + defer cancel() + began := time.Now() + + _, err = (&snapshotter{db: db}).checkpoint(ctx) + + waited := time.Since(began) + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("checkpoint() error = %v, want one matching context.DeadlineExceeded", err) + } + if waited > internalLongBusyTimeout/10 { + t.Errorf("checkpoint() answered after %v, want within a tenth of the busy timeout %v", waited, + internalLongBusyTimeout) + } +} + func TestACopyWithNoPagesFailsTheCheck(t *testing.T) { t.Parallel() @@ -151,11 +193,43 @@ func TestACopyThatIsNoDatabaseFailsTheCheck(t *testing.T) { var driverErr *modernc.Error want := "dbkit: check the snapshot copy " + path + ": " - if !errors.As(err, &driverErr) || driverErr.Code() != sqlite3.SQLITE_NOTADB || !strings.HasPrefix(err.Error(), want) { + if !errors.As(err, &driverErr) { + t.Fatalf("checkCopy() error = %v, want a driver error", err) + } + if driverErr.Code() != sqlite3.SQLITE_NOTADB || !strings.HasPrefix(err.Error(), want) { t.Errorf("checkCopy() error = %v, want SQLITE_NOTADB in an error starting %q", err, want) } } +// internalAnswer is what a call of Snapshot answered. +type internalAnswer struct { + // snap is the snapshot of the call. + snap dbkit.Snapshot + // err is the error of the call. + err error +} + +// internalSnapshotTo runs a snapshot of db to target and returns what it answered. +func internalSnapshotTo(t *testing.T, db *sql.DB, target string) internalAnswer { + t.Helper() + snap, err := NewSnapshotter(db).Snapshot(t.Context(), target) + return internalAnswer{snap: snap, err: err} +} + +// internalMustAnswer fails the test unless got holds the snapshot of want and an error matching it that reads text. +func internalMustAnswer(t *testing.T, got, want internalAnswer, text string) { + t.Helper() + if got.snap != want.snap { + t.Errorf("Snapshot() = %+v, want %+v", got.snap, want.snap) + } + if !errors.Is(got.err, want.err) { + t.Errorf("Snapshot() error = %v, want one matching %v", got.err, want.err) + } + if read := fmt.Sprint(got.err); read != text { + t.Errorf("Snapshot() error = %s, want %q", read, text) + } +} + func TestSyncPathFailsOnAMissingFile(t *testing.T) { t.Parallel() @@ -197,12 +271,10 @@ func TestAFailedCopySyncLeavesNoTarget(t *testing.T) { target := filepath.Join(folder, "copy.db") internalFailSync(t, target+partialSuffix) - got, err := NewSnapshotter(db).Snapshot(t.Context(), target) + got := internalSnapshotTo(t, db, target) - want := "dbkit: sync the snapshot copy " + target + partialSuffix + ": " + errSnapshotSyncFailed.Error() - if !errors.Is(err, errSnapshotSyncFailed) || err.Error() != want || got != (dbkit.Snapshot{}) { - t.Errorf("Snapshot() = %+v, %v, want no snapshot and %q", got, err, want) - } + internalMustAnswer(t, got, internalAnswer{err: errSnapshotSyncFailed}, + "dbkit: sync the snapshot copy "+target+partialSuffix+": "+errSnapshotSyncFailed.Error()) internalMustHoldOnly(t, folder, snapshotLockFile) } @@ -213,12 +285,10 @@ func TestAFailedMoveLeavesNoTarget(t *testing.T) { rename = func(string, string) error { return errSnapshotMoveFailed } t.Cleanup(func() { rename = kept }) - got, err := NewSnapshotter(db).Snapshot(t.Context(), target) + got := internalSnapshotTo(t, db, target) - want := "dbkit: move the snapshot copy to " + target + ": " + errSnapshotMoveFailed.Error() - if !errors.Is(err, errSnapshotMoveFailed) || err.Error() != want || got != (dbkit.Snapshot{}) { - t.Errorf("Snapshot() = %+v, %v, want no snapshot and %q", got, err, want) - } + internalMustAnswer(t, got, internalAnswer{err: errSnapshotMoveFailed}, + "dbkit: move the snapshot copy to "+target+": "+errSnapshotMoveFailed.Error()) internalMustHoldOnly(t, folder, snapshotLockFile) } @@ -246,14 +316,11 @@ func TestASecondSnapshotInTheFolderKeepsTheFirstCopy(t *testing.T) { t.Fatalf("the first Snapshot() returned %v before its move", err) } - got, err := NewSnapshotter(db).Snapshot(t.Context(), second) + got := internalSnapshotTo(t, db, second) close(proceed) - want := "dbkit: take the snapshot lock " + filepath.Join(folder, snapshotLockFile) + ": " + - ErrSnapshotRunning.Error() - if !errors.Is(err, ErrSnapshotRunning) || err.Error() != want || got != (dbkit.Snapshot{}) { - t.Errorf("the second Snapshot() = %+v, %v, want no snapshot and %q", got, err, want) - } + internalMustAnswer(t, got, internalAnswer{err: ErrSnapshotRunning}, + "dbkit: take the snapshot lock "+filepath.Join(folder, snapshotLockFile)+": "+ErrSnapshotRunning.Error()) if err := <-result; err != nil { t.Errorf("the first Snapshot() error = %v, want nil", err) } @@ -274,9 +341,11 @@ func TestSnapshotKeepsATargetMadeDuringTheCopy(t *testing.T) { got, err := NewSnapshotter(db).Snapshot(t.Context(), target) - want := "dbkit: the snapshot target " + target + " already exists" - if err == nil || err.Error() != want || got != (dbkit.Snapshot{}) { - t.Errorf("Snapshot() = %+v, %v, want no snapshot and %q", got, err, want) + if got != (dbkit.Snapshot{}) { + t.Errorf("Snapshot() = %+v, want no snapshot", got) + } + if read, want := fmt.Sprint(err), "dbkit: the snapshot target "+target+" already exists"; read != want { + t.Errorf("Snapshot() error = %s, want %q", read, want) } if data, err := os.ReadFile(target); err != nil || string(data) != internalLateTarget { t.Errorf("the target holds %d bytes, %v after the snapshot, want the %d bytes written during the copy", @@ -290,11 +359,9 @@ func TestAFailedFolderSyncKeepsTheCopy(t *testing.T) { target := filepath.Join(folder, "copy.db") internalFailSync(t, folder) - got, err := NewSnapshotter(db).Snapshot(t.Context(), target) + got := internalSnapshotTo(t, db, target) - want := "dbkit: sync the folder of the snapshot " + target + ": " + errSnapshotSyncFailed.Error() - if !errors.Is(err, errSnapshotSyncFailed) || err.Error() != want || got != (dbkit.Snapshot{Path: target}) { - t.Errorf("Snapshot() = %+v, %v, want the path %s with no checkpoint and %q", got, err, target, want) - } + internalMustAnswer(t, got, internalAnswer{snap: dbkit.Snapshot{Path: target}, err: errSnapshotSyncFailed}, + "dbkit: sync the folder of the snapshot "+target+": "+errSnapshotSyncFailed.Error()) internalMustHoldOnly(t, folder, snapshotLockFile, "copy.db") } diff --git a/dbkit/sqlite/sqlitetest/faults_test.go b/dbkit/sqlite/sqlitetest/faults_test.go index 8a53c54..e7f5f50 100644 --- a/dbkit/sqlite/sqlitetest/faults_test.go +++ b/dbkit/sqlite/sqlitetest/faults_test.go @@ -20,6 +20,8 @@ import ( const ( // testBusyTimeout is a busy timeout no uncontended test waits out. testBusyTimeout = 5 * time.Second + // testBusyPoll is the busy poll the tests pass, the smallest sqlite.Open accepts. + testBusyPoll = time.Millisecond // testCacheSize is the page cache in kibibytes the tests pass. testCacheSize = 1024 // testMaxConns is the connection cap the tests pass. @@ -49,6 +51,7 @@ var errInjected = errors.New("sqlitetest test: the fault answered on purpose") func testOptions() sqlite.Options { return sqlite.Options{ BusyTimeout: testBusyTimeout, + BusyPoll: testBusyPoll, CacheSize: testCacheSize, MaxConns: testMaxConns, Synchronous: sqlite.SynchronousNormal, @@ -510,12 +513,35 @@ func mustRead[T any](t *testing.T, conn *sql.Conn, query string, args ...any) T return value } -// mustKeepTheRules fails the test unless connection i of a handle from testOptions carries every rule. -func mustKeepTheRules(t *testing.T, i int, conn *sql.Conn) { +// mustWaitBehind fails the test unless BeginTx on connection i of conns waits testShortBusyTimeout behind the next one. +func mustWaitBehind(t *testing.T, conns []*sql.Conn, i int) { t.Helper() + holder, err := conns[(i+1)%len(conns)].BeginTx(t.Context(), nil) + if err != nil { + t.Fatalf("BeginTx() on the connection after %d error = %v, want nil", i, err) + } + defer func() { _ = holder.Rollback() }() + began := time.Now() + + tx, err := conns[i].BeginTx(t.Context(), nil) + + waited := time.Since(began) + if err == nil { + _ = tx.Rollback() + } + if !errors.Is(sqlite.Classify(err), dbkit.ErrBusy) || waited < testShortBusyTimeout { + t.Errorf("BeginTx() on connection %d behind a held writer = %v after %v, want busy after at least %v", i, err, + waited, testShortBusyTimeout) + } +} + +// mustKeepTheRules fails the test unless connection i of conns, from testOptions, carries every rule. +func mustKeepTheRules(t *testing.T, conns []*sql.Conn, i int) { + t.Helper() + conn := conns[i] pragmas := map[string]int64{ "PRAGMA foreign_keys": 1, - "PRAGMA busy_timeout": testBusyTimeout.Milliseconds(), + "PRAGMA busy_timeout": 0, "PRAGMA temp_store": 2, "PRAGMA cache_size": -testCacheSize, "PRAGMA synchronous": 1, @@ -537,6 +563,7 @@ func mustKeepTheRules(t *testing.T, i int, conn *sql.Conn) { if err == nil || !strings.Contains(err.Error(), "may not be modified") { t.Errorf("an edit of the schema table on connection %d error = %v, want it refused", i, err) } + mustWaitBehind(t, conns, i) } // errMissing is what a probe answers when the raw connection lacks its interface. @@ -900,13 +927,15 @@ func TestTheRulesStayOnThroughAWrappedConnection(t *testing.T) { } opts := testOptions() opts.Functions = functions + opts.BusyTimeout = testShortBusyTimeout faults := &sqlitetest.Faults{} db := sqlitetest.OpenWithFaults(t, opts, faults) mustExec(t, db, faultSchema) faults.FailStatement(faultChosen, errInjected) + conns := mustTakeEvery(t, db) - for i, conn := range mustTakeEvery(t, db) { - mustKeepTheRules(t, i, conn) + for i, conn := range conns { + mustKeepTheRules(t, conns, i) if got := mustRead[string](t, conn, "SELECT casefold('ÑANDÚ')"); got != "ñandú" { t.Errorf("casefold('ÑANDÚ') on connection %d = %q, want ñandú", i, got) } diff --git a/dbkit/sqlite/sqlitetest/open_internal_test.go b/dbkit/sqlite/sqlitetest/open_internal_test.go index d0cebbe..41677cb 100644 --- a/dbkit/sqlite/sqlitetest/open_internal_test.go +++ b/dbkit/sqlite/sqlitetest/open_internal_test.go @@ -51,9 +51,13 @@ func runRecorded(t *testing.T, helper func(testing.TB)) *recorder { return r } +// internalBusyPoll is the busy poll of internalOptions, the smallest sqlite.Open accepts. +const internalBusyPoll = time.Millisecond + // internalOptions returns options with every required value set. func internalOptions() sqlite.Options { - return sqlite.Options{BusyTimeout: 5 * time.Second, CacheSize: 1024, MaxConns: 2, Synchronous: sqlite.SynchronousFull} + return sqlite.Options{BusyTimeout: 5 * time.Second, BusyPoll: internalBusyPoll, CacheSize: 1024, MaxConns: 2, + Synchronous: sqlite.SynchronousFull} } // stubConn offers every interface of the driver's connection and runs nothing. diff --git a/dbkit/sqlite/sqlitetest/template_test.go b/dbkit/sqlite/sqlitetest/template_test.go index 99c0f9d..fdd62bd 100644 --- a/dbkit/sqlite/sqlitetest/template_test.go +++ b/dbkit/sqlite/sqlitetest/template_test.go @@ -6,6 +6,7 @@ import ( "context" "database/sql" "errors" + "fmt" "io/fs" "os" "path/filepath" @@ -34,6 +35,8 @@ const ( templateLockWait = 5 * time.Second // templateLockPoll is how often a run tries the migration lock. templateLockPoll = 10 * time.Millisecond + // templateShortBusyTimeout is a busy timeout a test waits out on purpose. + templateShortBusyTimeout = 30 * time.Millisecond ) // schemaMigration returns a migrate function that creates the template table and counts its runs in runs. @@ -116,35 +119,44 @@ func TestATemplateCopyAnswersFaults(t *testing.T) { } } -func TestTemplateServesEveryTestItsOwnCopy(t *testing.T) { - t.Parallel() +// templateWriteOwnCopy writes one row to a fresh copy of tp, checks the copy holds that row only and returns its path. +func templateWriteOwnCopy(t *testing.T, tp *Template) string { + t.Helper() + db := tp.Open(t) + if _, err := db.ExecContext(t.Context(), templateInsert); err != nil { + t.Fatalf("ExecContext(%q) error = %v, want nil", templateInsert, err) + } + if got := templateRows(t, db); got != 1 { + t.Errorf("rows in the copy = %d, want 1, its own row only", got) + } + return databaseFile(t, db) +} - var runs int - tp := mustTemplate(t, internalOptions(), schemaMigration(&runs)) +// templateCopiesOfEveryTest runs templateTests parallel tests that each write to a copy of tp and returns every path. +func templateCopiesOfEveryTest(t *testing.T, tp *Template) []string { + t.Helper() copies := make(chan string, templateTests) - t.Run("tests", func(t *testing.T) { for i := range templateTests { t.Run(strconv.Itoa(i), func(t *testing.T) { t.Parallel() - db := tp.Open(t) - if _, err := db.ExecContext(t.Context(), templateInsert); err != nil { - t.Fatalf("ExecContext(%q) error = %v, want nil", templateInsert, err) - } - if got := templateRows(t, db); got != 1 { - t.Errorf("rows in the copy = %d, want 1, its own row only", got) - } - copies <- databaseFile(t, db) + copies <- templateWriteOwnCopy(t, tp) }) } }) close(copies) - - if runs != 1 { - t.Errorf("migrate ran %d times, want once for every test", runs) + var paths []string + for path := range copies { + paths = append(paths, path) } + return paths +} + +// templateMustBeDistinct fails the test unless no path sits in the template folder of tp and no two paths are equal. +func templateMustBeDistinct(t *testing.T, tp *Template, paths []string) { + t.Helper() seen := map[string]bool{} - for path := range copies { + for _, path := range paths { if filepath.Dir(path) == tp.folder { t.Errorf("a test opened %s in the template folder, want a copy in its own folder", path) } @@ -153,6 +165,20 @@ func TestTemplateServesEveryTestItsOwnCopy(t *testing.T) { if len(seen) != templateTests { t.Errorf("the tests opened %d distinct files, want %d", len(seen), templateTests) } +} + +func TestTemplateServesEveryTestItsOwnCopy(t *testing.T) { + t.Parallel() + + var runs int + tp := mustTemplate(t, internalOptions(), schemaMigration(&runs)) + + paths := templateCopiesOfEveryTest(t, tp) + + if runs != 1 { + t.Errorf("migrate ran %d times, want once for every test", runs) + } + templateMustBeDistinct(t, tp, paths) if got := templateRows(t, tp.Open(t)); got != 0 { t.Errorf("rows in a copy made after every test wrote = %d, want 0 with the template untouched", got) } @@ -217,12 +243,42 @@ func templateRead[T any](t *testing.T, conn *sql.Conn, query string, args ...any // templateSynchronous maps each Synchronous option to the value PRAGMA synchronous reads back. var templateSynchronous = map[sqlite.Synchronous]int64{sqlite.SynchronousNormal: 1, sqlite.SynchronousFull: 2} -// templateMustKeepTheRules fails the test unless connection i of a copy opened with opts carries every rule. -func templateMustKeepTheRules(t *testing.T, i int, conn *sql.Conn, opts sqlite.Options) { +// templateMustWaitBehind fails the test unless BeginTx on connection i of conns waits timeout behind the next one. +func templateMustWaitBehind(t *testing.T, conns []*sql.Conn, i int, timeout time.Duration) { + t.Helper() + holder, err := conns[(i+1)%len(conns)].BeginTx(t.Context(), nil) + if err != nil { + t.Fatalf("BeginTx() on the connection after %d error = %v, want nil", i, err) + } + defer func() { _ = holder.Rollback() }() + began := time.Now() + + tx, err := conns[i].BeginTx(t.Context(), nil) + + waited := time.Since(began) + if err == nil { + _ = tx.Rollback() + } + if !errors.Is(sqlite.Classify(err), dbkit.ErrBusy) || waited < timeout { + t.Errorf("BeginTx() on connection %d behind a held writer = %v after %v, want busy after at least %v", i, err, + waited, timeout) + } +} + +// templateMustKeepTheRules fails the test unless connection i of conns, from a copy opened with opts, keeps every rule. +func templateMustKeepTheRules(t *testing.T, conns []*sql.Conn, i int, opts sqlite.Options) { + t.Helper() + templateMustReadTheRules(t, i, conns[i], opts) + templateMustRefuseASchemaEdit(t, i, conns[i]) + templateMustWaitBehind(t, conns, i, opts.BusyTimeout) +} + +// templateMustReadTheRules fails the test unless every pragma and function of connection i reads the rule of opts. +func templateMustReadTheRules(t *testing.T, i int, conn *sql.Conn, opts sqlite.Options) { t.Helper() pragmas := map[string]int64{ "PRAGMA foreign_keys": 1, - "PRAGMA busy_timeout": opts.BusyTimeout.Milliseconds(), + "PRAGMA busy_timeout": 0, "PRAGMA temp_store": 2, "PRAGMA cache_size": -int64(opts.CacheSize), "PRAGMA synchronous": templateSynchronous[opts.Synchronous], @@ -242,6 +298,11 @@ func templateMustKeepTheRules(t *testing.T, i int, conn *sql.Conn, opts sqlite.O if got := templateRead[string](t, conn, "SELECT casefold('ÑANDÚ')"); got != "ñandú" { t.Errorf("casefold('ÑANDÚ') on connection %d = %q, want ñandú", i, got) } +} + +// templateMustRefuseASchemaEdit fails the test unless connection i refuses an edit of the schema table. +func templateMustRefuseASchemaEdit(t *testing.T, i int, conn *sql.Conn) { + t.Helper() if _, err := conn.ExecContext(t.Context(), "PRAGMA writable_schema=ON"); err != nil { t.Fatalf("PRAGMA writable_schema=ON on connection %d error = %v, want nil", i, err) } @@ -260,13 +321,15 @@ func TestTemplateCopyKeepsEveryRule(t *testing.T) { } opts := internalOptions() opts.Functions = functions + opts.BusyTimeout = templateShortBusyTimeout var runs int tp := mustTemplate(t, opts, schemaMigration(&runs)) db := tp.Open(t) - for i, conn := range templateConns(t, db, opts.MaxConns) { - templateMustKeepTheRules(t, i, conn, opts) + conns := templateConns(t, db, opts.MaxConns) + for i := range conns { + templateMustKeepTheRules(t, conns, i, opts) } } @@ -301,6 +364,17 @@ func templateLinkRoot(t *testing.T) string { return target } +// mustRefuseTemplate fails the test unless NewTemplate answered no template and an error that reads want. +func mustRefuseTemplate(t *testing.T, tp *Template, err error, want string) { + t.Helper() + if tp != nil { + t.Errorf("NewTemplate() template = %v, want nil", tp) + } + if got := fmt.Sprint(err); got != want { + t.Errorf("NewTemplate() error = %s, want %q", got, want) + } +} + // mustHoldNothing fails the test unless folder is empty. func mustHoldNothing(t *testing.T, folder string) { t.Helper() @@ -334,9 +408,7 @@ func TestTemplateThatFailsLeavesNoFolderBehind(t *testing.T) { tp, err := NewTemplate(t.Context(), c.opts, c.migrate) - if tp != nil || err == nil || err.Error() != c.want { - t.Errorf("NewTemplate() = %v, %v, want nil and %q", tp, err, c.want) - } + mustRefuseTemplate(t, tp, err, c.want) mustHoldNothing(t, root) }) } @@ -428,9 +500,11 @@ func TestTemplateNeedsATempFolder(t *testing.T) { tp, err := NewTemplate(t.Context(), internalOptions(), schemaMigration(&runs)) - if tp != nil || !errors.Is(err, c.marked) || !strings.HasPrefix(err.Error(), c.want) { - t.Errorf("NewTemplate() = %v, %v, want nil and an error marked %v starting %q", - tp, err, c.marked, c.want) + if tp != nil { + t.Errorf("NewTemplate() template = %v, want nil", tp) + } + if !errors.Is(err, c.marked) || !strings.HasPrefix(err.Error(), c.want) { + t.Errorf("NewTemplate() error = %v, want an error marked %v starting %q", err, c.marked, c.want) } }) } @@ -457,10 +531,7 @@ func TestTemplateRefusesANilMigrate(t *testing.T) { tp, err := NewTemplate(t.Context(), internalOptions(), nil) - want := "dbkit: NewTemplate needs a migrate function, got nil" - if tp != nil || err == nil || err.Error() != want { - t.Errorf("NewTemplate() = %v, %v, want nil and %q", tp, err, want) - } + mustRefuseTemplate(t, tp, err, "dbkit: NewTemplate needs a migrate function, got nil") mustHoldNothing(t, root) } @@ -496,11 +567,8 @@ func TestTemplateRefusesAMigrateThatLeavesAConnectionOpen(t *testing.T) { return err }) - want := "dbkit: the template keeps its write-ahead log " + file + "-wal after its handle closed, " + - "so migrate left a connection open" - if tp != nil || err == nil || err.Error() != want { - t.Errorf("NewTemplate() = %v, %v, want nil and %q", tp, err, want) - } + mustRefuseTemplate(t, tp, err, "dbkit: the template keeps its write-ahead log "+file+"-wal after its handle closed, "+ + "so migrate left a connection open") mustHoldNothing(t, root) }