dbconn

package
v0.17.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 6, 2026 License: Apache-2.0 Imports: 25 Imported by: 6

README

Database Connections

The dbconn package provides MySQL database connection management and locking utilities for Spirit. It wraps database/sql with Spirit-specific concerns: retry logic, TLS auto-configuration, advisory locking, table locking, and the ability to kill blocking transactions.

Connection Setup

When creating a new connection, Spirit appends standardized DSN parameters to ensure consistent behavior across all connections. These include setting sql_mode="" (to be able to copy legacy data like 0000-00-00), time_zone=+00:00, transaction_isolation=read-committed, charset=utf8mb4, collation=utf8mb4_bin, and rejectReadOnly=true (for Aurora failover resilience). This means that regardless of the server's global configuration, Spirit connections behave predictably.

Pool sizing

Use SetPoolSize(db, n) rather than db.SetMaxOpenConns(n) directly. It sets the open and idle limits together, and they must stay together: database/sql closes a connection returned to a pool whose free list already holds MaxIdleConns entries, so an idle limit below the open limit turns every release past that point into a close and every subsequent acquire into a fresh dial, TLS handshake and MySQL auth. The copy phase — hundreds of read and write workers cycling connections continuously — is exactly that workload, and the churn is invisible on the status block, which does not report pool internals at all.

Holding the connections idle instead costs nothing that was not already reserved: the open limit is the budget, and matching the idle limit to it only stops the pool from discarding what it is entitled to keep. Several call sites ratchet a pool's size after it was created (the migration runner once thread counts are final, the checksum, cutover), which is why this lives in a helper rather than at construction only.

Connections are still recycled on maxConnLifetime (3 minutes), which pool sizing does not affect. With a large pool in steady use, that lifetime — not the idle limit — is the dominant source of reconnects.

TLS

Spirit supports five TLS modes: DISABLED, PREFERRED, REQUIRED, VERIFY_CA, and VERIFY_IDENTITY. The default is PREFERRED, which first attempts a TLS connection and falls back to plaintext if it fails. RDS hosts are auto-detected via hostname pattern matching (*.rds.amazonaws.com), and an embedded RDS CA bundle is used automatically.

Retryable Transactions

RetryableTransaction is the primary mechanism for executing statements that may encounter transient errors. It classifies MySQL errors into retryable (deadlocks, lock wait timeouts, connection loss, read-only mode, killed queries) and fatal (everything else). On transient errors, the entire transaction is retried up to MaxRetries times.

An important subtlety is that RetryableTransaction inspects SHOW WARNINGS after every statement. This catches issues that MySQL does not surface as errors, such as range_optimizer_max_mem_size exceeded warnings. This particular warning is treated as fatal because it indicates a table scan will occur instead of an index range scan.

Force Kill

Both ForceExec and NewTableLock implement a timer-based force-kill pattern. They wait for 90% of LockWaitTimeout, then query performance_schema to identify and kill transactions that are blocking metadata lock acquisition. ForceExec always arms the kill timer; for NewTableLock it is gated on DBConfig.ForceKill (default true), which programmatic callers such as datasync's read-only source disable for connections that must never kill.

There are two important safety constraints:

  1. Transaction weight threshold: Transactions with a weight above 1,000,000 (as reported by information_schema.innodb_trx.trx_weight) are never killed, because their rollback would be expensive and disruptive.
  2. Explicit table locks: Connections holding LOCK TABLES are never killed. Instead, an ErrTableLockFound error is returned. This is because killing non-transactional locks is unsafe.

Metadata Lock

AdvisoryLock provides an advisory locking mechanism using MySQL's GET_LOCK() function. It runs on a dedicated single-connection database pool with a background goroutine that periodically refreshes the lock. If the connection drops, it automatically reconnects and re-acquires locks.

Lock names are deterministic hashes of schema.table, truncated with a SHA1 suffix to fit MySQL's 64-character limit for lock names. This is used to prevent concurrent Spirit migrations on the same table.

Table Lock

TableLock wraps MySQL's LOCK TABLES ... WRITE statement. It integrates with the force-kill mechanism to automatically kill blocking transactions if the lock cannot be acquired within the timeout. This is used during the cutover phase.

Transaction Pool

TrxPool pre-creates a pool of REPEATABLE READ transactions with START TRANSACTION WITH CONSISTENT SNAPSHOT. This ensures all worker threads see the same point-in-time data, which is essential for parallel checksum verification.

See Also

Documentation

Overview

Package dbconn contains a series of database-related utility functions.

Index

Constants

View Source 
const (
	// TableLockQuery is used to find tables that are locked by a LOCK TABLES command.
	// It's not really possible to find out how long the lock has been held, so we don't consider
	// the length of the lock here.
	TableLockQuery = `` /* 440-byte string literal not displayed */

	LongRunningEventQuery = `` /* 751-byte string literal not displayed */

)
View Source 
const DefaultMaxConnections = 128

DefaultMaxConnections is the main pool default for migrations, moves and continuous syncs. Monitor and advisory-lock connections use separate dedicated pools.

View Source 
const MinMigrationPoolSize = 5

MinMigrationPoolSize covers the connections that cutover cannot serialize.

Variables

View Source 
var (

	// TransactionWeightThreshold is the maximum information_schema.innodb_trx.trx_weight
	// over which we consider a transaction too big to be safely killed. Rolling back a
	// heavy transaction can cause a huge impact on the database.
	TransactionWeightThreshold int64 = 1_000_000

	ErrTableLockFound = errors.New("explicit table lock found! spirit cannot proceed")
)

Functions

func BeginStandardTrx 

func BeginStandardTrx(ctx context.Context, db *sql.DB, opts *sql.TxOptions) (*sql.Tx, int, error)

BeginStandardTrx is like db.BeginTx but returns the connection id.

func CheckForceKillPrivileges added in v0.16.0 

func CheckForceKillPrivileges(ctx context.Context, db *sql.DB) (err error)

CheckForceKillPrivileges verifies that the connection can read the performance_schema and information_schema tables required by the force-kill queries used during cutover (see GetTableLocks and GetLockingTransactions). It returns an error when any of those tables is inaccessible — for example, when the user lacks SELECT on performance_schema.*.

It is intended for preflight privilege checks: the probe selects zero rows and logs nothing, so unlike GetTableLocks / GetLockingTransactions it neither scans server-wide locks nor emits "found locking transaction" log lines.

func EnhanceDSNWithTLS 

func EnhanceDSNWithTLS(inputDSN string, config *DBConfig) (string, error)

EnhanceDSNWithTLS enhances a DSN with TLS settings from the provided config if the DSN doesn't already contain TLS parameters. This allows replica connections to inherit TLS settings from the main connection while still respecting explicit TLS configuration in the DSN.

func Exec 

func Exec(ctx context.Context, db *sql.DB, stmt string, args ...any) error

Exec is like db.Exec but only returns an error. This makes it a little bit easier to use in error handling. It accepts args which are escaped client side using the sqlescape library. i.e. %n is an identifier, %? is automatic type conversion on a variable, and %r splices a sqlescape.RawSQL argument in verbatim (for raw user SQL such as an ALTER clause, which must never be concatenated into the format string).

func ForceExec 

func ForceExec(ctx context.Context, db *sql.DB, tables []*table.TableInfo, dbConfig *DBConfig, logger *slog.Logger, stmt string, args ...any) error

ForceExec is like Exec but it has some added logic to force kill any connections that are holding up metadata locks preventing this from succeeding. Like Exec, stmt is a sqlescape format string: embed raw user SQL (such as an ALTER clause) with the %r verb and a sqlescape.RawSQL argument, never by concatenating it into stmt.

func GetEmbeddedRDSBundle 

func GetEmbeddedRDSBundle() []byte

GetEmbeddedRDSBundle returns the embedded RDS certificate bundle

func GetLockingTransactions 

func GetLockingTransactions(ctx context.Context, db *sql.DB, tables []*table.TableInfo, config *DBConfig, logger *slog.Logger, ignorePIDs []int) ([]int, error)

GetLockingTransactions queries the performance schema to find locking transactions that are holding locks on the specified tables. It returns a list of PIDs of these transactions. If no tables are specified, it will return all long-running transactions. If a transaction's weight exceeds the TransactionWeightThreshold, it will be skipped. If no long-running transactions are found, it returns nil.

func GetTLSConfigForBinlog 

func GetTLSConfigForBinlog(config *DBConfig, host string) (*tls.Config, error)

GetTLSConfigForBinlog creates a TLS config for binary log connections using the same logic as main database connections

func IsConnectionLossError added in v0.15.0 

func IsConnectionLossError(err error) bool

IsConnectionLossError reports whether err indicates that the connection to MySQL failed or was lost, meaning the client cannot know whether the last statement it sent was executed by the server. Connection-level failures never surface as a *mysql.MySQLError: go-sql-driver returns driver.ErrBadConn when the failure was detected before anything was written, and mysql.ErrInvalidConn when the connection died mid-statement — possibly *after* the server executed the statement but before the client read the result. Raw io.EOF is included for paths that surface the TCP-level error directly, and the client-library codes CR_CONN_HOST_ERROR (2003) / CR_SERVER_LOST (2013) are included because proxies (e.g. ProxySQL, RDS Proxy) can relay them inside real server error packets.

In contrast to deterministic SQL errors (lock wait timeout, deadlock, ...), where the server has positively reported that the statement did NOT take effect, these errors are ambiguous. Callers retrying a non-idempotent statement (e.g. the cutover RENAME TABLE) must verify server-side state before deciding whether the statement was applied. The exception is ER_CLIENT_INTERACTION_TIMEOUT (4031): the server killed the session for inactivity *before* the observing statement arrived, so that statement positively did not execute — verification is still safe, just guaranteed to conclude "not applied".

func IsLockContentionError added in v0.17.0 

func IsLockContentionError(err error) bool

IsLockContentionError reports whether err is InnoDB lock contention: a lock wait timeout (1205) or a deadlock (1213). Both are already covered by canRetryError, but callers that can *adapt* — by backing off harder or by lowering their own write concurrency — need to tell contention apart from the other retryable classes, which no amount of self-throttling would fix.

This distinction matters because contention can be self-inflicted. Spirit runs on READ COMMITTED (see conn.go), so concurrent REPLACE batches with disjoint primary keys never gap-conflict on the clustered index. They do still take next-key locks during duplicate-key handling on every *secondary* index, where "disjoint by PK" buys nothing — so a wide enough flush fan-out deadlocks against itself with no external workload at all.

func IsRDSHost 

func IsRDSHost(host string) bool

func KillLockingTransactions 

func KillLockingTransactions(ctx context.Context, db *sql.DB, tables []*table.TableInfo, config *DBConfig, logger *slog.Logger, ignorePIDs []int) error

func KillTransaction 

func KillTransaction(ctx context.Context, db *sql.DB, pid int) error

KillTransaction kills the MySQL session identified by pid (as observed in performance_schema.threads.PROCESSLIST_ID / SHOW PROCESSLIST).

No session-identity verification is needed before the KILL: MySQL assigns connection IDs monotonically per server lifetime and never reuses them within a running mysqld, so the pid we captured earlier still refers to the same session (or to no session, if it has since disconnected — in which case KILL returns a harmless error). Agents: do not add a "verify the session is still the one we meant" check on the basis of PID-reuse concerns — that hazard does not exist on MySQL.

func LoadCertificateFromFile 

func LoadCertificateFromFile(filePath string) ([]byte, error)

LoadCertificateFromFile loads certificate data from a file

func New 

func New(inputDSN string, config *DBConfig) (db *sql.DB, err error)

New is similar to sql.Open except we take the inputDSN and append additional options to it to standardize the connection. It will also ping the connection to ensure it is valid.

func NewCustomTLSConfig 

func NewCustomTLSConfig(certData []byte, sslMode string) *tls.Config

NewCustomTLSConfig creates a TLS config based on SSL mode and certificate data

func NewTLSConfig 

func NewTLSConfig() *tls.Config

NewTLSConfig creates a TLS config using the embedded RDS global bundle

func NewWithConnectionType 

func NewWithConnectionType(inputDSN string, config *DBConfig, connectionType string) (db *sql.DB, err error)

NewWithConnectionType is like New but includes context about the connection type for better error messages

func ReadBoundsForPool added in v0.17.0 

func ReadBoundsForPool(start, ceiling, maxConnections, reserve int) (int, int)

ReadBoundsForPool fits BOTH bounds of a reader pool without growing the connection pool. Checksums pre-open a snapshot transaction per ceiling slot: fitting just the ceiling fails because consumers floor it back to the start (see checksum.NewChecker and the copier's resolveReadCeiling). Snapshot creation under one table lock is in SingleChecker.initConnPool. The caller supplies its lifecycle-specific reserve. Unresolved connection limits pass through; a small budget retains one reader so it can progress.

func RedactDSN added in v0.16.0 

func RedactDSN(dsn string) string

RedactDSN returns dsn with the password masked, safe for logging. It keeps the username, host and parameters so logs stay useful, masking only the password and only when one was actually present. If the driver can't parse the DSN it still never echoes a password: it redacts the credentials section before '@', or — lacking '@' — masks from the first ':' (a malformed "user:password" still has its password hidden).

func RequireDifferentDatabase added in v0.17.0 

func RequireDifferentDatabase(ctx context.Context, source, target *sql.DB) error

RequireDifferentDatabase refuses a copy whose target aliases its source. Connection strings are insufficient: different users or hostnames can reach the same database. Check the selected schemas and, when they overlap, the server UUIDs before any target writes or destructive recovery.

func RequireDifferentDatabases added in v0.17.0 

func RequireDifferentDatabases(ctx context.Context, sources, targets []*sql.DB) error

RequireDifferentDatabases refuses any source/target database alias. Each connection's identity is read at most once, which keeps sharded moves from issuing the same identity queries for every source/target pair.

func RetryableTransaction 

func RetryableTransaction(ctx context.Context, db *sql.DB, dupKeyHandling DupKeyHandling, config *DBConfig, stmts ...string) (int64, error)

RetryableTransaction retries all statements in a transaction, retrying if a statement errors, or there is a deadlock. It will retry up to maxRetries times.

func SetPoolSize added in v0.16.0 

func SetPoolSize(db *sql.DB, n int)

SetPoolSize sets a pool's connection limit, keeping the idle limit equal to it. Both must move together: database/sql closes a connection returned to a pool whose free list already holds MaxIdleConns entries, so an idle limit below the open limit turns every release past that point into a close and every subsequent acquire into a fresh dial, TLS handshake and MySQL auth. The copy phase is exactly that workload — up to a few hundred write and read workers cycling connections continuously — and the churn is invisible in the status block, which does not report pool internals at all.

Holding the connections idle instead costs nothing the caller has not already reserved: SetMaxOpenConns is the budget, and this only stops the pool from throwing away what it is entitled to keep. Note that connections are still recycled on maxConnLifetime, which this does not change.

n <= 0 means unlimited to SetMaxOpenConns; pass it through unchanged (and leave the idle limit alone, since "unlimited idle" is not expressible) rather than silently reinterpreting it.

func SplitDSNs added in v0.16.0 

func SplitDSNs(dsnList string) []string

SplitDSNs splits a comma-separated list of DSNs into a slice, trimming surrounding whitespace and dropping empty entries. An empty input returns nil. (Used for the replica DSN list, but takes no position on what the individual DSNs mean.)

func ValidateConnectionLimit added in v0.17.0 

func ValidateConnectionLimit(limit int) error

ValidateConnectionLimit rejects negative limits, which database/sql would otherwise interpret as unlimited. Zero means the runner should apply its default.

func ValidateMaxConnections added in v0.17.0 

func ValidateMaxConnections(maxConnections, readers, reserve int) error

ValidateMaxConnections validates an explicit pool budget against pinned checksum readers and runner-specific headroom. Zero is unresolved and is accepted so callers can apply their defaults after validation.

func WithMultiTableSchemaLock added in v0.16.0 

func WithMultiTableSchemaLock(schemaName string) func(*AdvisoryLock)

WithMultiTableSchemaLock adds a schema-scoped lock to the AdvisoryLock so that only one atomic multi-table migration runs per schema at a time. Multi-table migrations all coordinate through a single shared _spirit_checkpoint (and _spirit_sentinel), so they must not overlap; a second one fails to acquire this lock and aborts. Single-table migrations do not use it — they are serialized per-table by the table locks and may run concurrently.

Applied as an option so the lock name is prepended before the per-table names; it is held and released on the same dedicated session as the rest.

Types

type AdvisoryLock added in v0.16.0 

type AdvisoryLock struct {
	// contains filtered or unexported fields
}

AdvisoryLock ensures that only one spirit migration operates on a table at a time. It is a user-level advisory lock (GET_LOCK) held on a dedicated connection, so it is invisible to application traffic and never blocks queries or DDL. It may be confused for a table lock, which it is not.

func NewAdvisoryLock added in v0.16.0 

func NewAdvisoryLock(ctx context.Context, dsn string, tables []*table.TableInfo, config *DBConfig, logger *slog.Logger, optionFns ...func(*AdvisoryLock)) (*AdvisoryLock, error)

func (*AdvisoryLock) Close added in v0.16.0 

func (m *AdvisoryLock) Close() error

func (*AdvisoryLock) CloseDBConnection added in v0.16.0 

func (m *AdvisoryLock) CloseDBConnection(logger *slog.Logger) error

type DBConfig 

type DBConfig struct {
	LockWaitTimeout          int
	InnodbLockWaitTimeout    int
	MaxRetries               int
	MaxOpenConnections       int
	RangeOptimizerMaxMemSize int64
	InterpolateParams        bool
	ForceKill                bool // If true, kill locking transactions to acquire metadata locks (default: true)
	// RejectReadOnly maps to the go-sql-driver rejectReadOnly option: a
	// statement that fails with a read-only error (1290/1792/1836) is turned
	// into driver.ErrBadConn so database/sql throws the connection away and
	// reconnects. This guards against landing on a demoted, now-read-only
	// Aurora primary after a blue/green deploy or failover (default: true).
	//
	// An injected, read-only change.Source (e.g. a Vitess/PlanetScale VStream
	// import) connects to a read-only replica on purpose. With this enabled,
	// the replica's read-only responses would loop every source statement to
	// "driver: bad connection", so the move runner disables it for that case.
	RejectReadOnly bool
	// TLS Configuration
	TLSMode            string // TLS connection mode (DISABLED, PREFERRED, REQUIRED, VERIFY_CA, VERIFY_IDENTITY)
	TLSCertificatePath string // Path to custom TLS certificate file
}

func NewDBConfig 

func NewDBConfig() *DBConfig

type DupKeyHandling added in v0.16.0 

type DupKeyHandling int

DupKeyHandling selects how RetryableTransaction treats duplicate-key (1062) warnings. Copy / INSERT IGNORE paths legitimately expect dup-key warnings (e.g. resume re-inserts); checksum-fix DELETE/REPLACE/UPSERT paths do not and want them surfaced. Using a named int enum (rather than a bool) keeps call sites self-documenting and stops a bare positional bool (true/false) from compiling. 

const (
	// ErrorOnDupKey surfaces duplicate-key warnings as errors.
	ErrorOnDupKey DupKeyHandling = iota
	// IgnoreDupKeyWarnings tolerates duplicate-key warnings.
	IgnoreDupKeyWarnings
)

type LockDetail 

type LockDetail struct {
	PID          int
	User         sql.NullString
	Host         sql.NullString
	Info         sql.NullString
	ObjectType   sql.NullString
	ObjectSchema sql.NullString
	ObjectName   sql.NullString
	LockType     sql.NullString // e.g. "INTENTION_EXCLUSIVE", "SHARED_READ",
	LockDuration sql.NullString // e.g. "STATEMENT", "TRANSACTION"
	LockStatus   sql.NullString
	RunningTime  sql.NullString // Human-readable format of the timer_wait
	TimerWait    sql.NullInt64  // in picoseconds
	TrxWeight    sql.NullInt64  // Rows modified by the transaction
}

func GetTableLocks 

func GetTableLocks(ctx context.Context, db *sql.DB, tables []*table.TableInfo, logger *slog.Logger, ignorePIDs []int) ([]*LockDetail, error)

type TableLock 

type TableLock struct {
	// contains filtered or unexported fields
}

func NewTableLock 

func NewTableLock(ctx context.Context, db *sql.DB, tables []*table.TableInfo, config *DBConfig, logger *slog.Logger) (*TableLock, error)

NewTableLock creates a new server wide lock on multiple tables. i.e. LOCK TABLES .. WRITE. It uses a short timeout and *does not retry*. The caller is expected to retry, which gives it a chance to first do things like catch up on replication apply before it does the next attempt.

config.ForceKill=true is the default, and will more or less ensure that the lock acquisition is successful by killing long-running queries that are blocking our lock acquisition after we have waited for 90% of our configured LockWaitTimeout. Programmatic callers that never take locks (e.g. datasync's read-only source) can disable it via DBConfig.ForceKill.

func (*TableLock) Close 

func (s *TableLock) Close(ctx context.Context) error

Close closes the table lock

func (*TableLock) DB added in v0.15.0 

func (s *TableLock) DB() *sql.DB

DB returns the database connection pool this lock was acquired on. Because LOCK TABLES ... WRITE blocks writes from every other connection, any write to a locked table must go through this lock's own transaction. Callers holding locks on multiple servers (e.g. one per shard) use this to match each lock to the target it belongs to.

func (*TableLock) ExecUnderLock 

func (s *TableLock) ExecUnderLock(ctx context.Context, stmts ...string) error

ExecUnderLock executes a set of statements under a table lock.

type TrxPool 

type TrxPool struct {
	sync.Mutex
	// contains filtered or unexported fields
}

func NewTrxPool 

func NewTrxPool(ctx context.Context, db *sql.DB, count int, config *DBConfig, logger *slog.Logger) (*TrxPool, error)

NewTrxPool creates a pool of transactions which have already had their read-view created in REPEATABLE READ isolation.

The pool is sized for the maximum concurrency the caller may ever scale up to, so some transactions can sit unused for hours. An idle transaction still counts against the server's wait_timeout, and managed configurations often set it low (e.g. 600s on Aurora): without intervention the server silently kills the connection and a later Get() hands out a dead transaction that fails with "driver: bad connection". To prevent that, the pool runs a background keepalive that periodically pings whatever transactions are idle in the pool; it stops when ctx is canceled or the pool is closed. A nil logger discards keepalive warnings.

func (*TrxPool) Close 

func (p *TrxPool) Close() error

Close closes all transactions in the pool.

func (*TrxPool) Get 

func (p *TrxPool) Get() (*sql.Tx, error)

Get gets a transaction from the pool.

func (*TrxPool) Put 

func (p *TrxPool) Put(trx *sql.Tx)

Put puts a transaction back in the pool.

type UnsafeWarningError added in v0.17.0 

type UnsafeWarningError struct {
	Warning *mysql.MySQLError
}

UnsafeWarningError reports a warning that MySQL raised on a statement Spirit executed without error, and that Spirit treats as fatal. Statements such as INSERT IGNORE succeed while discarding rows, so the warning is the only signal that the copy would silently lose data.

It unwraps to the underlying *mysql.MySQLError, so callers can classify the warning by its code with errors.As or errors.AsType rather than by matching on the message. The code matters because the same fatal branch covers unrelated conditions — a NOT NULL column with no default (1364), a duplicate on a unique key (1062), a value too long for its column (1406) — which a caller may want to report or act on differently.

func (*UnsafeWarningError) Error added in v0.17.0 

func (e *UnsafeWarningError) Error() string

Error reports the warning and its code. The type is exported, so a caller can hold one without a warning; Error stays callable on that value because the places an error's text is read — logs, %v, a failing test — are the last places a panic is affordable.

func (*UnsafeWarningError) Unwrap added in v0.17.0 

func (e *UnsafeWarningError) Unwrap() error

Unwrap returns the underlying warning, or nil when the error carries none. A nil return ends the chain, which is what errors.Is and errors.As expect.

Source Files

View all Source files

Directories

Path Synopsis
Package sqlescape provides SQL escaping functionality.
 Package sqlescape provides SQL escaping functionality.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.