15624673805

Committed 13 Jun 2025 01:56AM UTC coverage: 82.906% (-0.004%) from 82.91%

Build # 15624673805

Build Type

push

github

Committed by

web-flow

Commit Message

[1.9] fix PING on compressed connections (#1723)

Add missing mc.syncSequence()

Fix #1718

Run Details

15 of 18 new or added lines in 2 files covered. (83.33%)

7 existing lines in 3 files now uncovered.

3264 of 3937 relevant lines covered (82.91%)

2489344.05 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

74.0

/connection.go

// Go MySQL Driver - A MySQL-Driver for Go's database/sql package
//
// Copyright 2012 The Go-MySQL-Driver Authors. All rights reserved.
//
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this file,
// You can obtain one at http://mozilla.org/MPL/2.0/.

package mysql

import (
        "context"
        "database/sql"
        "database/sql/driver"
        "encoding/json"
        "fmt"
        "io"
        "net"
        "runtime"
        "strconv"
        "strings"
        "sync/atomic"
        "time"
)

type mysqlConn struct {
        buf              buffer
        netConn          net.Conn
        rawConn          net.Conn    // underlying connection when netConn is TLS connection.
        result           mysqlResult // managed by clearResult() and handleOkPacket().
        compIO           *compIO
        cfg              *Config
        connector        *connector
        maxAllowedPacket int
        maxWriteSize     int
        flags            clientFlag
        status           statusFlag
        sequence         uint8
        compressSequence uint8
        parseTime        bool
        compress         bool

        // for context support (Go 1.8+)
        watching bool
        watcher  chan<- context.Context
        closech  chan struct{}
        finished chan<- struct{}
        canceled atomicError // set non-nil if conn is canceled
        closed   atomic.Bool // set when conn is closed, before closech is closed
}

// Helper function to call per-connection logger.
func (mc *mysqlConn) log(v ...any) {
        _, filename, lineno, ok := runtime.Caller(1)
        if ok {
                pos := strings.LastIndexByte(filename, '/')
                if pos != -1 {
                        filename = filename[pos+1:]
                }
                prefix := fmt.Sprintf("%s:%d ", filename, lineno)
                v = append([]any{prefix}, v...)
        }

        mc.cfg.Logger.Print(v...)
}

func (mc *mysqlConn) readWithTimeout(b []byte) (int, error) {
        to := mc.cfg.ReadTimeout
        if to > 0 {
                if err := mc.netConn.SetReadDeadline(time.Now().Add(to)); err != nil {
                        return 0, err
                }
        }
        return mc.netConn.Read(b)
}

func (mc *mysqlConn) writeWithTimeout(b []byte) (int, error) {
        to := mc.cfg.WriteTimeout
        if to > 0 {
                if err := mc.netConn.SetWriteDeadline(time.Now().Add(to)); err != nil {
                        return 0, err
                }
        }
        return mc.netConn.Write(b)
}

func (mc *mysqlConn) resetSequence() {
        mc.sequence = 0
        mc.compressSequence = 0
}

// syncSequence must be called when finished writing some packet and before start reading.
func (mc *mysqlConn) syncSequence() {
        // Syncs compressionSequence to sequence.
        // This is not documented but done in `net_flush()` in MySQL and MariaDB.
        // https://github.com/mariadb-corporation/mariadb-connector-c/blob/8228164f850b12353da24df1b93a1e53cc5e85e9/libmariadb/ma_net.c#L170-L171
        // https://github.com/mysql/mysql-server/blob/824e2b4064053f7daf17d7f3f84b7a3ed92e5fb4/sql-common/net_serv.cc#L293
        if mc.compress {
                mc.sequence = mc.compressSequence
                mc.compIO.reset()
        }
}

// Handles parameters set in DSN after the connection is established
func (mc *mysqlConn) handleParams() (err error) {
        var cmdSet strings.Builder

        for param, val := range mc.cfg.Params {
                if cmdSet.Len() == 0 {
                        // Heuristic: 29 chars for each other key=value to reduce reallocations
                        cmdSet.Grow(4 + len(param) + 3 + len(val) + 30*(len(mc.cfg.Params)-1))
                        cmdSet.WriteString("SET ")
                } else {
                        cmdSet.WriteString(", ")
                }
                cmdSet.WriteString(param)
                cmdSet.WriteString(" = ")
                cmdSet.WriteString(val)
        }

        if cmdSet.Len() > 0 {
                err = mc.exec(cmdSet.String())
        }

        return
}

// markBadConn replaces errBadConnNoWrite with driver.ErrBadConn.
// This function is used to return driver.ErrBadConn only when safe to retry.
func (mc *mysqlConn) markBadConn(err error) error {
        if err == errBadConnNoWrite {
                return driver.ErrBadConn
        }
        return err
}

func (mc *mysqlConn) Begin() (driver.Tx, error) {
        return mc.begin(false)
}

func (mc *mysqlConn) begin(readOnly bool) (driver.Tx, error) {
        if mc.closed.Load() {
                return nil, driver.ErrBadConn
        }
        var q string
        if readOnly {
                q = "START TRANSACTION READ ONLY"
        } else {
                q = "START TRANSACTION"
        }
        err := mc.exec(q)
        if err == nil {
                return &mysqlTx{mc}, err
        }
        return nil, mc.markBadConn(err)
}

func (mc *mysqlConn) Close() (err error) {
        // Makes Close idempotent
        if !mc.closed.Load() {
                err = mc.writeCommandPacket(comQuit)
        }
        mc.close()
        return
}

// close closes the network connection and clear results without sending COM_QUIT.
func (mc *mysqlConn) close() {
        mc.cleanup()
        mc.clearResult()
}

// Closes the network connection and unsets internal variables. Do not call this
// function after successfully authentication, call Close instead. This function
// is called before auth or on auth failure because MySQL will have already
// closed the network connection.
func (mc *mysqlConn) cleanup() {
        if mc.closed.Swap(true) {
                return
        }

        // Makes cleanup idempotent
        close(mc.closech)
        conn := mc.rawConn
        if conn == nil {
                return
        }
        if err := conn.Close(); err != nil {
                mc.log("closing connection:", err)
        }
        // This function can be called from multiple goroutines.
        // So we can not mc.clearResult() here.
        // Caller should do it if they are in safe goroutine.
}

func (mc *mysqlConn) error() error {
        if mc.closed.Load() {
                if err := mc.canceled.Value(); err != nil {
                        return err
                }
                return ErrInvalidConn
        }
        return nil
}

func (mc *mysqlConn) Prepare(query string) (driver.Stmt, error) {
        if mc.closed.Load() {
                return nil, driver.ErrBadConn
        }
        // Send command
        err := mc.writeCommandPacketStr(comStmtPrepare, query)
        if err != nil {
                // STMT_PREPARE is safe to retry.  So we can return ErrBadConn here.
                mc.log(err)
                return nil, driver.ErrBadConn
        }

        stmt := &mysqlStmt{
                mc: mc,
        }

        // Read Result
        columnCount, err := stmt.readPrepareResultPacket()
        if err == nil {
                if stmt.paramCount > 0 {
                        if err = mc.readUntilEOF(); err != nil {
                                return nil, err
                        }
                }

                if columnCount > 0 {
                        err = mc.readUntilEOF()
                }
        }

        return stmt, err
}

func (mc *mysqlConn) interpolateParams(query string, args []driver.Value) (string, error) {
        // Number of ? should be same to len(args)
        if strings.Count(query, "?") != len(args) {
                return "", driver.ErrSkip
        }

        buf, err := mc.buf.takeCompleteBuffer()
        if err != nil {
                // can not take the buffer. Something must be wrong with the connection
                mc.cleanup()
                // interpolateParams would be called before sending any query.
                // So its safe to retry.
                return "", driver.ErrBadConn
        }
        buf = buf[:0]
        argPos := 0

        for i := 0; i < len(query); i++ {
                q := strings.IndexByte(query[i:], '?')
                if q == -1 {
                        buf = append(buf, query[i:]...)
                        break
                }
                buf = append(buf, query[i:i+q]...)
                i += q

                arg := args[argPos]
                argPos++

                if arg == nil {
                        buf = append(buf, "NULL"...)
                        continue
                }

                switch v := arg.(type) {
                case int64:
                        buf = strconv.AppendInt(buf, v, 10)
                case uint64:
                        // Handle uint64 explicitly because our custom ConvertValue emits unsigned values
                        buf = strconv.AppendUint(buf, v, 10)
                case float64:
                        buf = strconv.AppendFloat(buf, v, 'g', -1, 64)
                case bool:
                        if v {
                                buf = append(buf, '1')
                        } else {
                                buf = append(buf, '0')
                        }
                case time.Time:
                        if v.IsZero() {
                                buf = append(buf, "'0000-00-00'"...)
                        } else {
                                buf = append(buf, '\'')
                                buf, err = appendDateTime(buf, v.In(mc.cfg.Loc), mc.cfg.timeTruncate)
                                if err != nil {
                                        return "", err
                                }
                                buf = append(buf, '\'')
                        }
                case json.RawMessage:
                        buf = append(buf, '\'')
                        if mc.status&statusNoBackslashEscapes == 0 {
                                buf = escapeBytesBackslash(buf, v)
                        } else {
                                buf = escapeBytesQuotes(buf, v)
                        }
                        buf = append(buf, '\'')
                case []byte:
                        if v == nil {
                                buf = append(buf, "NULL"...)
                        } else {
                                buf = append(buf, "_binary'"...)
                                if mc.status&statusNoBackslashEscapes == 0 {
                                        buf = escapeBytesBackslash(buf, v)
                                } else {
                                        buf = escapeBytesQuotes(buf, v)
                                }
                                buf = append(buf, '\'')
                        }
                case string:
                        buf = append(buf, '\'')
                        if mc.status&statusNoBackslashEscapes == 0 {
                                buf = escapeStringBackslash(buf, v)
                        } else {
                                buf = escapeStringQuotes(buf, v)
                        }
                        buf = append(buf, '\'')
                default:
                        return "", driver.ErrSkip
                }

                if len(buf)+4 > mc.maxAllowedPacket {
                        return "", driver.ErrSkip
                }
        }
        if argPos != len(args) {
                return "", driver.ErrSkip
        }
        return string(buf), nil
}

func (mc *mysqlConn) Exec(query string, args []driver.Value) (driver.Result, error) {
        if mc.closed.Load() {
                return nil, driver.ErrBadConn
        }
        if len(args) != 0 {
                if !mc.cfg.InterpolateParams {
                        return nil, driver.ErrSkip
                }
                // try to interpolate the parameters to save extra roundtrips for preparing and closing a statement
                prepared, err := mc.interpolateParams(query, args)
                if err != nil {
                        return nil, err
                }
                query = prepared
        }

        err := mc.exec(query)
        if err == nil {
                copied := mc.result
                return &copied, err
        }
        return nil, mc.markBadConn(err)
}

// Internal function to execute commands
func (mc *mysqlConn) exec(query string) error {
        handleOk := mc.clearResult()
        // Send command
        if err := mc.writeCommandPacketStr(comQuery, query); err != nil {
                return mc.markBadConn(err)
        }

        // Read Result
        resLen, err := handleOk.readResultSetHeaderPacket()
        if err != nil {
                return err
        }

        if resLen > 0 {
                // columns
                if err := mc.readUntilEOF(); err != nil {
                        return err
                }

                // rows
                if err := mc.readUntilEOF(); err != nil {
                        return err
                }
        }

        return handleOk.discardResults()
}

func (mc *mysqlConn) Query(query string, args []driver.Value) (driver.Rows, error) {
        return mc.query(query, args)
}

func (mc *mysqlConn) query(query string, args []driver.Value) (*textRows, error) {
        handleOk := mc.clearResult()

        if mc.closed.Load() {
                return nil, driver.ErrBadConn
        }
        if len(args) != 0 {
                if !mc.cfg.InterpolateParams {
                        return nil, driver.ErrSkip
                }
                // try client-side prepare to reduce roundtrip
                prepared, err := mc.interpolateParams(query, args)
                if err != nil {
                        return nil, err
                }
                query = prepared
        }
        // Send command
        err := mc.writeCommandPacketStr(comQuery, query)
        if err != nil {
                return nil, mc.markBadConn(err)
        }

        // Read Result
        var resLen int
        resLen, err = handleOk.readResultSetHeaderPacket()
        if err != nil {
                return nil, err
        }

        rows := new(textRows)
        rows.mc = mc

        if resLen == 0 {
                rows.rs.done = true

                switch err := rows.NextResultSet(); err {
                case nil, io.EOF:
                        return rows, nil
                default:
                        return nil, err
                }
        }

        // Columns
        rows.rs.columns, err = mc.readColumns(resLen)
        return rows, err
}

// Gets the value of the given MySQL System Variable
// The returned byte slice is only valid until the next read
func (mc *mysqlConn) getSystemVar(name string) ([]byte, error) {
        // Send command
        handleOk := mc.clearResult()
        if err := mc.writeCommandPacketStr(comQuery, "SELECT @@"+name); err != nil {
                return nil, err
        }

        // Read Result
        resLen, err := handleOk.readResultSetHeaderPacket()
        if err == nil {
                rows := new(textRows)
                rows.mc = mc
                rows.rs.columns = []mysqlField{{fieldType: fieldTypeVarChar}}

                if resLen > 0 {
                        // Columns
                        if err := mc.readUntilEOF(); err != nil {
                                return nil, err
                        }
                }

                dest := make([]driver.Value, resLen)
                if err = rows.readRow(dest); err == nil {
                        return dest[0].([]byte), mc.readUntilEOF()
                }
        }
        return nil, err
}

// cancel is called when the query has canceled.
func (mc *mysqlConn) cancel(err error) {
        mc.canceled.Set(err)
        mc.cleanup()
}

// finish is called when the query has succeeded.
func (mc *mysqlConn) finish() {
        if !mc.watching || mc.finished == nil {
                return
        }
        select {
        case mc.finished <- struct{}{}:
                mc.watching = false
        case <-mc.closech:
        }
}

// Ping implements driver.Pinger interface
func (mc *mysqlConn) Ping(ctx context.Context) (err error) {
        if mc.closed.Load() {
                return driver.ErrBadConn
        }

        if err = mc.watchCancel(ctx); err != nil {
                return
        }
        defer mc.finish()

        handleOk := mc.clearResult()
        if err = mc.writeCommandPacket(comPing); err != nil {
                return mc.markBadConn(err)
        }

        return handleOk.readResultOK()
}

// BeginTx implements driver.ConnBeginTx interface
func (mc *mysqlConn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, error) {
        if mc.closed.Load() {
                return nil, driver.ErrBadConn
        }

        if err := mc.watchCancel(ctx); err != nil {
                return nil, err
        }
        defer mc.finish()

        if sql.IsolationLevel(opts.Isolation) != sql.LevelDefault {
                level, err := mapIsolationLevel(opts.Isolation)
                if err != nil {
                        return nil, err
                }
                err = mc.exec("SET TRANSACTION ISOLATION LEVEL " + level)
                if err != nil {
                        return nil, err
                }
        }

        return mc.begin(opts.ReadOnly)
}

func (mc *mysqlConn) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {
        dargs, err := namedValueToValue(args)
        if err != nil {
                return nil, err
        }

        if err := mc.watchCancel(ctx); err != nil {
                return nil, err
        }

        rows, err := mc.query(query, dargs)
        if err != nil {
                mc.finish()
                return nil, err
        }
        rows.finish = mc.finish
        return rows, err
}

func (mc *mysqlConn) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) {
        dargs, err := namedValueToValue(args)
        if err != nil {
                return nil, err
        }

        if err := mc.watchCancel(ctx); err != nil {
                return nil, err
        }
        defer mc.finish()

        return mc.Exec(query, dargs)
}

func (mc *mysqlConn) PrepareContext(ctx context.Context, query string) (driver.Stmt, error) {
        if err := mc.watchCancel(ctx); err != nil {
                return nil, err
        }

        stmt, err := mc.Prepare(query)
        mc.finish()
        if err != nil {
                return nil, err
        }

        select {
        default:
        case <-ctx.Done():
                stmt.Close()
                return nil, ctx.Err()
        }
        return stmt, nil
}

func (stmt *mysqlStmt) QueryContext(ctx context.Context, args []driver.NamedValue) (driver.Rows, error) {
        dargs, err := namedValueToValue(args)
        if err != nil {
                return nil, err
        }

        if err := stmt.mc.watchCancel(ctx); err != nil {
                return nil, err
        }

        rows, err := stmt.query(dargs)
        if err != nil {
                stmt.mc.finish()
                return nil, err
        }
        rows.finish = stmt.mc.finish
        return rows, err
}

func (stmt *mysqlStmt) ExecContext(ctx context.Context, args []driver.NamedValue) (driver.Result, error) {
        dargs, err := namedValueToValue(args)
        if err != nil {
                return nil, err
        }

        if err := stmt.mc.watchCancel(ctx); err != nil {
                return nil, err
        }
        defer stmt.mc.finish()

        return stmt.Exec(dargs)
}

func (mc *mysqlConn) watchCancel(ctx context.Context) error {
        if mc.watching {
                // Reach here if canceled,
                // so the connection is already invalid
                mc.cleanup()
                return nil
        }
        // When ctx is already cancelled, don't watch it.
        if err := ctx.Err(); err != nil {
                return err
        }
        // When ctx is not cancellable, don't watch it.
        if ctx.Done() == nil {
                return nil
        }
        // When watcher is not alive, can't watch it.
        if mc.watcher == nil {
                return nil
        }

        mc.watching = true
        mc.watcher <- ctx
        return nil
}

func (mc *mysqlConn) startWatcher() {
        watcher := make(chan context.Context, 1)
        mc.watcher = watcher
        finished := make(chan struct{})
        mc.finished = finished
        go func() {
                for {
                        var ctx context.Context
                        select {
                        case ctx = <-watcher:
                        case <-mc.closech:
                                return
                        }

                        select {
                        case <-ctx.Done():
                                mc.cancel(ctx.Err())
                        case <-finished:
                        case <-mc.closech:
                                return
                        }
                }
        }()
}

func (mc *mysqlConn) CheckNamedValue(nv *driver.NamedValue) (err error) {
        nv.Value, err = converter{}.ConvertValue(nv.Value)
        return
}

// ResetSession implements driver.SessionResetter.
// (From Go 1.10)
func (mc *mysqlConn) ResetSession(ctx context.Context) error {
        if mc.closed.Load() || mc.buf.busy() {
                return driver.ErrBadConn
        }

        // Perform a stale connection check. We only perform this check for
        // the first query on a connection that has been checked out of the
        // connection pool: a fresh connection from the pool is more likely
        // to be stale, and it has not performed any previous writes that
        // could cause data corruption, so it's safe to return ErrBadConn
        // if the check fails.
        if mc.cfg.CheckConnLiveness {
                conn := mc.netConn
                if mc.rawConn != nil {
                        conn = mc.rawConn
                }
                var err error
                if mc.cfg.ReadTimeout != 0 {
                        err = conn.SetReadDeadline(time.Now().Add(mc.cfg.ReadTimeout))
                }
                if err == nil {
                        err = connCheck(conn)
                }
                if err != nil {
                        mc.log("closing bad idle connection: ", err)
                        return driver.ErrBadConn
                }
        }

        return nil
}

// IsValid implements driver.Validator interface
// (From Go 1.15)
func (mc *mysqlConn) IsValid() bool {
        return !mc.closed.Load() && !mc.buf.busy()
}

var _ driver.SessionResetter = &mysqlConn{}
var _ driver.Validator = &mysqlConn{}

1	// Go MySQL Driver - A MySQL-Driver for Go's database/sql package
2	//
3	// Copyright 2012 The Go-MySQL-Driver Authors. All rights reserved.
4	//
5	// This Source Code Form is subject to the terms of the Mozilla Public
6	// License, v. 2.0. If a copy of the MPL was not distributed with this file,
7	// You can obtain one at http://mozilla.org/MPL/2.0/.
8
9	package mysql
10
11	import (
12	"context"
13	"database/sql"
14	"database/sql/driver"
15	"encoding/json"
16	"fmt"
17	"io"
18	"net"
19	"runtime"
20	"strconv"
21	"strings"
22	"sync/atomic"
23	"time"
24	)
25
26	type mysqlConn struct {
27	buf buffer
28	netConn net.Conn
29	rawConn net.Conn // underlying connection when netConn is TLS connection.
30	result mysqlResult // managed by clearResult() and handleOkPacket().
31	compIO *compIO
32	cfg *Config
33	connector *connector
34	maxAllowedPacket int
35	maxWriteSize int
36	flags clientFlag
37	status statusFlag
38	sequence uint8
39	compressSequence uint8
40	parseTime bool
41	compress bool
42
43	// for context support (Go 1.8+)
44	watching bool
45	watcher chan<- context.Context
46	closech chan struct{}
47	finished chan<- struct{}
48	canceled atomicError // set non-nil if conn is canceled
49	closed atomic.Bool // set when conn is closed, before closech is closed
50	}
51
52	// Helper function to call per-connection logger.
53	func (mc *mysqlConn) log(v ...any) {	226✔
54	_, filename, lineno, ok := runtime.Caller(1)	226✔
55	if ok {	452✔
56	pos := strings.LastIndexByte(filename, '/')	226✔
57	if pos != -1 {	452✔
58	filename = filename[pos+1:]	226✔
59	}	226✔
60	prefix := fmt.Sprintf("%s:%d ", filename, lineno)	226✔
61	v = append([]any{prefix}, v...)	226✔
62	}
63
64	mc.cfg.Logger.Print(v...)	226✔
65	}
66
67	func (mc *mysqlConn) readWithTimeout(b []byte) (int, error) {	104,615✔
68	to := mc.cfg.ReadTimeout	104,615✔
69	if to > 0 {	104,615✔
70	if err := mc.netConn.SetReadDeadline(time.Now().Add(to)); err != nil {	×
71	return 0, err	×
72	}	×
73	}
74	return mc.netConn.Read(b)	104,615✔
75	}
76
77	func (mc *mysqlConn) writeWithTimeout(b []byte) (int, error) {	102,491✔
78	to := mc.cfg.WriteTimeout	102,491✔
79	if to > 0 {	102,491✔
80	if err := mc.netConn.SetWriteDeadline(time.Now().Add(to)); err != nil {	×
81	return 0, err	×
82	}	×
83	}
84	return mc.netConn.Write(b)	102,491✔
85	}
86
87	func (mc *mysqlConn) resetSequence() {	80,684✔
88	mc.sequence = 0	80,684✔
89	mc.compressSequence = 0	80,684✔
90	}	80,684✔
91
92	// syncSequence must be called when finished writing some packet and before start reading.
93	func (mc *mysqlConn) syncSequence() {	80,524✔
94	// Syncs compressionSequence to sequence.	80,524✔
95	// This is not documented but done in `net_flush()` in MySQL and MariaDB.	80,524✔
96	// https://github.com/mariadb-corporation/mariadb-connector-c/blob/8228164f850b12353da24df1b93a1e53cc5e85e9/libmariadb/ma_net.c#L170-L171	80,524✔
97	// https://github.com/mysql/mysql-server/blob/824e2b4064053f7daf17d7f3f84b7a3ed92e5fb4/sql-common/net_serv.cc#L293	80,524✔
98	if mc.compress {	104,714✔
99	mc.sequence = mc.compressSequence	24,190✔
100	mc.compIO.reset()	24,190✔
101	}	24,190✔
102	}
103
104	// Handles parameters set in DSN after the connection is established
105	func (mc *mysqlConn) handleParams() (err error) {	12,522✔
106	var cmdSet strings.Builder	12,522✔
107		12,522✔
108	for param, val := range mc.cfg.Params {	12,906✔
109	if cmdSet.Len() == 0 {	768✔
110	// Heuristic: 29 chars for each other key=value to reduce reallocations	384✔
111	cmdSet.Grow(4 + len(param) + 3 + len(val) + 30*(len(mc.cfg.Params)-1))	384✔
112	cmdSet.WriteString("SET ")	384✔
113	} else {	384✔
114	cmdSet.WriteString(", ")	×
115	}	×
116	cmdSet.WriteString(param)	384✔
117	cmdSet.WriteString(" = ")	384✔
118	cmdSet.WriteString(val)	384✔
119	}
120
121	if cmdSet.Len() > 0 {	12,906✔
122	err = mc.exec(cmdSet.String())	384✔
123	}	384✔
124
125	return	12,522✔
126	}
127
128	// markBadConn replaces errBadConnNoWrite with driver.ErrBadConn.
129	// This function is used to return driver.ErrBadConn only when safe to retry.
130	func (mc *mysqlConn) markBadConn(err error) error {	525✔
131	if err == errBadConnNoWrite {	557✔
132	return driver.ErrBadConn	32✔
133	}	32✔
134	return err	493✔
135	}
136
137	func (mc *mysqlConn) Begin() (driver.Tx, error) {	×
138	return mc.begin(false)	×
139	}	×
140
141	func (mc *mysqlConn) begin(readOnly bool) (driver.Tx, error) {	4,842✔
142	if mc.closed.Load() {	4,842✔
143	return nil, driver.ErrBadConn	×
144	}	×
145	var q string	4,842✔
146	if readOnly {	4,874✔
147	q = "START TRANSACTION READ ONLY"	32✔
148	} else {	4,842✔
149	q = "START TRANSACTION"	4,810✔
150	}	4,810✔
151	err := mc.exec(q)	4,842✔
152	if err == nil {	9,684✔
153	return &mysqlTx{mc}, err	4,842✔
154	}	4,842✔
155	return nil, mc.markBadConn(err)	×
156	}
157
158	func (mc *mysqlConn) Close() (err error) {	12,490✔
159	// Makes Close idempotent	12,490✔
160	if !mc.closed.Load() {	22,620✔
161	err = mc.writeCommandPacket(comQuit)	10,130✔
162	}	10,130✔
163	mc.close()	12,490✔
164	return	12,490✔
165	}
166
167	// close closes the network connection and clear results without sending COM_QUIT.
168	func (mc *mysqlConn) close() {	14,841✔
169	mc.cleanup()	14,841✔
170	mc.clearResult()	14,841✔
171	}	14,841✔
172
173	// Closes the network connection and unsets internal variables. Do not call this
174	// function after successfully authentication, call Close instead. This function
175	// is called before auth or on auth failure because MySQL will have already
176	// closed the network connection.
177	func (mc *mysqlConn) cleanup() {	25,978✔
178	if mc.closed.Swap(true) {	34,868✔
179	return	8,890✔
180	}	8,890✔
181
182	// Makes cleanup idempotent
183	close(mc.closech)	17,088✔
184	conn := mc.rawConn	17,088✔
185	if conn == nil {	17,280✔
186	return	192✔
187	}	192✔
188	if err := conn.Close(); err != nil {	16,896✔
189	mc.log("closing connection:", err)	×
190	}	×
191	// This function can be called from multiple goroutines.
192	// So we can not mc.clearResult() here.
193	// Caller should do it if they are in safe goroutine.
194	}
195
196	func (mc *mysqlConn) error() error {	45,072✔
197	if mc.closed.Load() {	47,036✔
198	if err := mc.canceled.Value(); err != nil {	3,928✔
199	return err	1,964✔
200	}	1,964✔
201	return ErrInvalidConn	×
202	}
203	return nil	43,108✔
204	}
205
206	func (mc *mysqlConn) Prepare(query string) (driver.Stmt, error) {	8,464✔
207	if mc.closed.Load() {	8,464✔
208	return nil, driver.ErrBadConn	×
209	}	×
210	// Send command
211	err := mc.writeCommandPacketStr(comStmtPrepare, query)	8,464✔
212	if err != nil {	8,464✔
213	// STMT_PREPARE is safe to retry. So we can return ErrBadConn here.	×
214	mc.log(err)	×
215	return nil, driver.ErrBadConn	×
216	}	×
217
218	stmt := &mysqlStmt{	8,464✔
219	mc: mc,	8,464✔
220	}	8,464✔
221		8,464✔
222	// Read Result	8,464✔
223	columnCount, err := stmt.readPrepareResultPacket()	8,464✔
224	if err == nil {	16,928✔
225	if stmt.paramCount > 0 {	16,608✔
226	if err = mc.readUntilEOF(); err != nil {	8,144✔
227	return nil, err	×
228	}	×
229	}
230
231	if columnCount > 0 {	14,944✔
232	err = mc.readUntilEOF()	6,480✔
233	}	6,480✔
234	}
235
236	return stmt, err	8,464✔
237	}
238
239	func (mc *mysqlConn) interpolateParams(query string, args []driver.Value) (string, error) {	3,592✔
240	// Number of ? should be same to len(args)	3,592✔
241	if strings.Count(query, "?") != len(args) {	3,656✔
242	return "", driver.ErrSkip	64✔
243	}	64✔
244
245	buf, err := mc.buf.takeCompleteBuffer()	3,528✔
246	if err != nil {	3,528✔
247	// can not take the buffer. Something must be wrong with the connection	×
248	mc.cleanup()	×
249	// interpolateParams would be called before sending any query.	×
250	// So its safe to retry.	×
251	return "", driver.ErrBadConn	×
252	}	×
253	buf = buf[:0]	3,528✔
254	argPos := 0	3,528✔
255		3,528✔
256	for i := 0; i < len(query); i++ {	10,488✔
257	q := strings.IndexByte(query[i:], '?')	6,960✔
258	if q == -1 {	10,232✔
259	buf = append(buf, query[i:]...)	3,272✔
260	break	3,272✔
261	}
262	buf = append(buf, query[i:i+q]...)	3,688✔
263	i += q	3,688✔
264		3,688✔
265	arg := args[argPos]	3,688✔
266	argPos++	3,688✔
267		3,688✔
268	if arg == nil {	3,688✔
269	buf = append(buf, "NULL"...)	×
270	continue	×
271	}
272
273	switch v := arg.(type) {	3,688✔
274	case int64:	320✔
275	buf = strconv.AppendInt(buf, v, 10)	320✔
276	case uint64:	32✔
277	// Handle uint64 explicitly because our custom ConvertValue emits unsigned values	32✔
278	buf = strconv.AppendUint(buf, v, 10)	32✔
279	case float64:	×
280	buf = strconv.AppendFloat(buf, v, 'g', -1, 64)	×
281	case bool:	×
282	if v {	×
283	buf = append(buf, '1')	×
284	} else {	×
285	buf = append(buf, '0')	×
286	}	×
287	case time.Time:	1,140✔
288	if v.IsZero() {	1,576✔
289	buf = append(buf, "'0000-00-00'"...)	436✔
290	} else {	1,140✔
291	buf = append(buf, '\'')	704✔
292	buf, err = appendDateTime(buf, v.In(mc.cfg.Loc), mc.cfg.timeTruncate)	704✔
293	if err != nil {	704✔
294	return "", err	×
295	}	×
296	buf = append(buf, '\'')	704✔
297	}
298	case json.RawMessage:	32✔
299	buf = append(buf, '\'')	32✔
300	if mc.status&statusNoBackslashEscapes == 0 {	64✔
301	buf = escapeBytesBackslash(buf, v)	32✔
302	} else {	32✔
303	buf = escapeBytesQuotes(buf, v)	×
304	}	×
305	buf = append(buf, '\'')	32✔
306	case []byte:	×
307	if v == nil {	×
308	buf = append(buf, "NULL"...)	×
309	} else {	×
310	buf = append(buf, "_binary'"...)	×
311	if mc.status&statusNoBackslashEscapes == 0 {	×
312	buf = escapeBytesBackslash(buf, v)	×
313	} else {	×
314	buf = escapeBytesQuotes(buf, v)	×
315	}	×
316	buf = append(buf, '\'')	×
317	}
318	case string:	2,164✔
319	buf = append(buf, '\'')	2,164✔
320	if mc.status&statusNoBackslashEscapes == 0 {	4,232✔
321	buf = escapeStringBackslash(buf, v)	2,068✔
322	} else {	2,164✔
323	buf = escapeStringQuotes(buf, v)	96✔
324	}	96✔
325	buf = append(buf, '\'')	2,164✔
326	default:	×
327	return "", driver.ErrSkip	×
328	}
329
330	if len(buf)+4 > mc.maxAllowedPacket {	3,688✔
331	return "", driver.ErrSkip	×
332	}	×
333	}
334	if argPos != len(args) {	3,528✔
335	return "", driver.ErrSkip	×
336	}	×
337	return string(buf), nil	3,528✔
338	}
339
340	func (mc *mysqlConn) Exec(query string, args []driver.Value) (driver.Result, error) {	18,160✔
341	if mc.closed.Load() {	18,160✔
342	return nil, driver.ErrBadConn	×
343	}	×
344	if len(args) != 0 {	20,432✔
345	if !mc.cfg.InterpolateParams {	4,064✔
346	return nil, driver.ErrSkip	1,792✔
347	}	1,792✔
348	// try to interpolate the parameters to save extra roundtrips for preparing and closing a statement
349	prepared, err := mc.interpolateParams(query, args)	480✔
350	if err != nil {	480✔
351	return nil, err	×
352	}	×
353	query = prepared	480✔
354	}
355
356	err := mc.exec(query)	16,368✔
357	if err == nil {	32,276✔
358	copied := mc.result	15,908✔
359	return &copied, err	15,908✔
360	}	15,908✔
361	return nil, mc.markBadConn(err)	460✔
362	}
363
364	// Internal function to execute commands
365	func (mc *mysqlConn) exec(query string) error {	27,032✔
366	handleOk := mc.clearResult()	27,032✔
367	// Send command	27,032✔
368	if err := mc.writeCommandPacketStr(comQuery, query); err != nil {	27,032✔
369	return mc.markBadConn(err)	×
370	}	×
371
372	// Read Result
373	resLen, err := handleOk.readResultSetHeaderPacket()	27,032✔
374	if err != nil {	27,652✔
375	return err	620✔
376	}	620✔
377
378	if resLen > 0 {	26,412✔
UNCOV 379	// columns	×
UNCOV 380	if err := mc.readUntilEOF(); err != nil {	×
381	return err	×
382	}	×
383
384	// rows
UNCOV 385	if err := mc.readUntilEOF(); err != nil {	×
386	return err	×
387	}	×
388	}
389
390	return handleOk.discardResults()	26,412✔
391	}
392
393	func (mc *mysqlConn) Query(query string, args []driver.Value) (driver.Rows, error) {	160✔
394	return mc.query(query, args)	160✔
395	}	160✔
396
397	func (mc mysqlConn) query(query string, args []driver.Value) (textRows, error) {	21,044✔
398	handleOk := mc.clearResult()	21,044✔
399		21,044✔
400	if mc.closed.Load() {	21,044✔
401	return nil, driver.ErrBadConn	×
402	}	×
403	if len(args) != 0 {	30,220✔
404	if !mc.cfg.InterpolateParams {	15,400✔
405	return nil, driver.ErrSkip	6,224✔
406	}	6,224✔
407	// try client-side prepare to reduce roundtrip
408	prepared, err := mc.interpolateParams(query, args)	2,952✔
409	if err != nil {	2,952✔
410	return nil, err	×
411	}	×
412	query = prepared	2,952✔
413	}
414	// Send command
415	err := mc.writeCommandPacketStr(comQuery, query)	14,820✔
416	if err != nil {	14,820✔
417	return nil, mc.markBadConn(err)	×
418	}	×
419
420	// Read Result
421	var resLen int	14,820✔
422	resLen, err = handleOk.readResultSetHeaderPacket()	14,820✔
423	if err != nil {	14,918✔
424	return nil, err	98✔
425	}	98✔
426
427	rows := new(textRows)	14,722✔
428	rows.mc = mc	14,722✔
429		14,722✔
430	if resLen == 0 {	14,882✔
431	rows.rs.done = true	160✔
432		160✔
433	switch err := rows.NextResultSet(); err {	160✔
434	case nil, io.EOF:	160✔
435	return rows, nil	160✔
436	default:	×
437	return nil, err	×
438	}
439	}
440
441	// Columns
442	rows.rs.columns, err = mc.readColumns(resLen)	14,562✔
443	return rows, err	14,562✔
444	}
445
446	// Gets the value of the given MySQL System Variable
447	// The returned byte slice is only valid until the next read
448	func (mc *mysqlConn) getSystemVar(name string) ([]byte, error) {	96✔
449	// Send command	96✔
450	handleOk := mc.clearResult()	96✔
451	if err := mc.writeCommandPacketStr(comQuery, "SELECT @@"+name); err != nil {	96✔
452	return nil, err	×
453	}	×
454
455	// Read Result
456	resLen, err := handleOk.readResultSetHeaderPacket()	96✔
457	if err == nil {	192✔
458	rows := new(textRows)	96✔
459	rows.mc = mc	96✔
460	rows.rs.columns = []mysqlField{{fieldType: fieldTypeVarChar}}	96✔
461		96✔
462	if resLen > 0 {	192✔
463	// Columns	96✔
464	if err := mc.readUntilEOF(); err != nil {	96✔
465	return nil, err	×
466	}	×
467	}
468
469	dest := make([]driver.Value, resLen)	96✔
470	if err = rows.readRow(dest); err == nil {	192✔
471	return dest[0].([]byte), mc.readUntilEOF()	96✔
472	}	96✔
473	}
474	return nil, err	×
475	}
476
477	// cancel is called when the query has canceled.
478	func (mc *mysqlConn) cancel(err error) {	6,474✔
479	mc.canceled.Set(err)	6,474✔
480	mc.cleanup()	6,474✔
481	}	6,474✔
482
483	// finish is called when the query has succeeded.
484	func (mc *mysqlConn) finish() {	80,370✔
485	if !mc.watching \|\| mc.finished == nil {	151,514✔
486	return	71,144✔
487	}	71,144✔
488	select {	9,226✔
489	case mc.finished <- struct{}{}:	2,752✔
490	mc.watching = false	2,752✔
491	case <-mc.closech:	6,474✔
492	}
493	}
494
495	// Ping implements driver.Pinger interface
496	func (mc *mysqlConn) Ping(ctx context.Context) (err error) {	2,430✔
497	if mc.closed.Load() {	2,474✔
498	return driver.ErrBadConn	44✔
499	}	44✔
500
501	if err = mc.watchCancel(ctx); err != nil {	2,486✔
502	return	100✔
503	}	100✔
504	defer mc.finish()	2,286✔
505		2,286✔
506	handleOk := mc.clearResult()	2,286✔
507	if err = mc.writeCommandPacket(comPing); err != nil {	2,351✔
508	return mc.markBadConn(err)	65✔
509	}	65✔
510
511	return handleOk.readResultOK()	2,221✔
512	}
513
514	// BeginTx implements driver.ConnBeginTx interface
515	func (mc *mysqlConn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, error) {	4,854✔
516	if mc.closed.Load() {	4,866✔
517	return nil, driver.ErrBadConn	12✔
518	}	12✔
519
520	if err := mc.watchCancel(ctx); err != nil {	4,842✔
521	return nil, err	×
522	}	×
523	defer mc.finish()	4,842✔
524		4,842✔
525	if sql.IsolationLevel(opts.Isolation) != sql.LevelDefault {	4,906✔
526	level, err := mapIsolationLevel(opts.Isolation)	64✔
527	if err != nil {	64✔
528	return nil, err	×
529	}	×
530	err = mc.exec("SET TRANSACTION ISOLATION LEVEL " + level)	64✔
531	if err != nil {	64✔
532	return nil, err	×
533	}	×
534	}
535
536	return mc.begin(opts.ReadOnly)	4,842✔
537	}
538
539	func (mc *mysqlConn) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {	20,884✔
540	dargs, err := namedValueToValue(args)	20,884✔
541	if err != nil {	20,884✔
542	return nil, err	×
543	}	×
544
545	if err := mc.watchCancel(ctx); err != nil {	20,884✔
546	return nil, err	×
547	}	×
548
549	rows, err := mc.query(query, dargs)	20,884✔
550	if err != nil {	27,206✔
551	mc.finish()	6,322✔
552	return nil, err	6,322✔
553	}	6,322✔
554	rows.finish = mc.finish	14,562✔
555	return rows, err	14,562✔
556	}
557
558	func (mc *mysqlConn) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) {	18,128✔
559	dargs, err := namedValueToValue(args)	18,128✔
560	if err != nil {	18,128✔
561	return nil, err	×
562	}	×
563
564	if err := mc.watchCancel(ctx); err != nil {	18,128✔
565	return nil, err	×
566	}	×
567	defer mc.finish()	18,128✔
568		18,128✔
569	return mc.Exec(query, dargs)	18,128✔
570	}
571
572	func (mc *mysqlConn) PrepareContext(ctx context.Context, query string) (driver.Stmt, error) {	8,432✔
573	if err := mc.watchCancel(ctx); err != nil {	8,432✔
574	return nil, err	×
575	}	×
576
577	stmt, err := mc.Prepare(query)	8,432✔
578	mc.finish()	8,432✔
579	if err != nil {	8,432✔
580	return nil, err	×
581	}	×
582
583	select {	8,432✔
584	default:	8,432✔
585	case <-ctx.Done():	×
586	stmt.Close()	×
587	return nil, ctx.Err()	×
588	}
589	return stmt, nil	8,432✔
590	}
591
592	func (stmt *mysqlStmt) QueryContext(ctx context.Context, args []driver.NamedValue) (driver.Rows, error) {	7,024✔
593	dargs, err := namedValueToValue(args)	7,024✔
594	if err != nil {	7,024✔
595	return nil, err	×
596	}	×
597
598	if err := stmt.mc.watchCancel(ctx); err != nil {	7,024✔
599	return nil, err	×
600	}	×
601
602	rows, err := stmt.query(dargs)	7,024✔
603	if err != nil {	7,056✔
604	stmt.mc.finish()	32✔
605	return nil, err	32✔
606	}	32✔
607	rows.finish = stmt.mc.finish	6,992✔
608	return rows, err	6,992✔
609	}
610
611	func (stmt *mysqlStmt) ExecContext(ctx context.Context, args []driver.NamedValue) (driver.Result, error) {	1,824✔
612	dargs, err := namedValueToValue(args)	1,824✔
613	if err != nil {	1,824✔
614	return nil, err	×
615	}	×
616
617	if err := stmt.mc.watchCancel(ctx); err != nil {	1,824✔
618	return nil, err	×
619	}	×
620	defer stmt.mc.finish()	1,824✔
621		1,824✔
622	return stmt.Exec(dargs)	1,824✔
623	}
624
625	func (mc *mysqlConn) watchCancel(ctx context.Context) error {	80,608✔
626	if mc.watching {	80,608✔
627	// Reach here if canceled,	×
628	// so the connection is already invalid	×
629	mc.cleanup()	×
630	return nil	×
631	}	×
632	// When ctx is already cancelled, don't watch it.
633	if err := ctx.Err(); err != nil {	80,814✔
634	return err	206✔
635	}	206✔
636	// When ctx is not cancellable, don't watch it.
637	if ctx.Done() == nil {	151,578✔
638	return nil	71,176✔
639	}	71,176✔
640	// When watcher is not alive, can't watch it.
641	if mc.watcher == nil {	9,226✔
642	return nil	×
643	}	×
644
645	mc.watching = true	9,226✔
646	mc.watcher <- ctx	9,226✔
647	return nil	9,226✔
648	}
649
650	func (mc *mysqlConn) startWatcher() {	17,120✔
651	watcher := make(chan context.Context, 1)	17,120✔
652	mc.watcher = watcher	17,120✔
653	finished := make(chan struct{})	17,120✔
654	mc.finished = finished	17,120✔
655	go func() {	34,240✔
656	for {	43,466✔
657	var ctx context.Context	26,346✔
658	select {	26,346✔
659	case ctx = <-watcher:	9,226✔
660	case <-mc.closech:	16,928✔
661	return	16,928✔
662	}
663
664	select {	9,226✔
665	case <-ctx.Done():	6,474✔
666	mc.cancel(ctx.Err())	6,474✔
667	case <-finished:	2,752✔
668	case <-mc.closech:	×
669	return	×
670	}
671	}
672	}()
673	}
674
675	func (mc mysqlConn) CheckNamedValue(nv driver.NamedValue) (err error) {	12,120✔
676	nv.Value, err = converter{}.ConvertValue(nv.Value)	12,120✔
677	return	12,120✔
678	}	12,120✔
679
680	// ResetSession implements driver.SessionResetter.
681	// (From Go 1.10)
682	func (mc *mysqlConn) ResetSession(ctx context.Context) error {	30,028✔
683	if mc.closed.Load() \|\| mc.buf.busy() {	30,028✔
684	return driver.ErrBadConn	×
685	}	×
686
687	// Perform a stale connection check. We only perform this check for
688	// the first query on a connection that has been checked out of the
689	// connection pool: a fresh connection from the pool is more likely
690	// to be stale, and it has not performed any previous writes that
691	// could cause data corruption, so it's safe to return ErrBadConn
692	// if the check fails.
693	if mc.cfg.CheckConnLiveness {	60,056✔
694	conn := mc.netConn	30,028✔
695	if mc.rawConn != nil {	60,056✔
696	conn = mc.rawConn	30,028✔
697	}	30,028✔
698	var err error	30,028✔
699	if mc.cfg.ReadTimeout != 0 {	30,028✔
700	err = conn.SetReadDeadline(time.Now().Add(mc.cfg.ReadTimeout))	×
701	}	×
702	if err == nil {	60,056✔
703	err = connCheck(conn)	30,028✔
704	}	30,028✔
705	if err != nil {	30,050✔
706	mc.log("closing bad idle connection: ", err)	22✔
707	return driver.ErrBadConn	22✔
708	}	22✔
709	}
710
711	return nil	30,006✔
712	}
713
714	// IsValid implements driver.Validator interface
715	// (From Go 1.15)
716	func (mc *mysqlConn) IsValid() bool {	42,216✔
717	return !mc.closed.Load() && !mc.buf.busy()	42,216✔
718	}	42,216✔
719
720	var _ driver.SessionResetter = &mysqlConn{}
721	var _ driver.Validator = &mysqlConn{}

go-sql-driver / mysql / 15624673805

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous