wip

Author: methane

auth.go

@@ -338,7 +338,7 @@ func (mc *mysqlConn) auth(authData []byte, plugin string) ([]byte, error) {
 		return authEd25519(authData, mc.cfg.Passwd)
 
 	default:
-		mc.cfg.Logger.Print("unknown auth plugin:", plugin)
+		mc.log("unknown auth plugin:", plugin)
 		return nil, ErrUnknownPlugin
 	}
 }

compress.go

@@ -123,17 +123,18 @@ func (c *decompressor) uncompressPacket() error {
 	uncompressedLength := int(uint32(header[4]) | uint32(header[5])<<8 | uint32(header[6])<<16)
 	compressionSequence := uint8(header[3])
 	if debugTrace {
-		c.mc.cfg.Logger.Print(
-			fmt.Sprintf("uncompress cmplen=%v uncomplen=%v pkt_cmp_seq=%v expected_cmp_seq=%v\n",
-				comprLength, uncompressedLength, compressionSequence, c.mc.sequence))
+		traceLogger.Printf("uncompress cmplen=%v uncomplen=%v pkt_cmp_seq=%v expected_cmp_seq=%v\n",
+			comprLength, uncompressedLength, compressionSequence, c.mc.sequence)
 	}
 	if compressionSequence != c.mc.sequence {
 		// return ErrPktSync
 		// server may return error packet (e.g. 1153 Got a packet bigger than 'max_allowed_packet' bytes)
 		// before receiving all packets from client. In this case, seqnr is younger than expected.
-		c.mc.cfg.Logger.Print(
-			fmt.Sprintf("[warn] unexpected cmpress seq nr: expected %v, got %v",
-				c.mc.sequence, compressionSequence))
+		if debugTrace {
+			traceLogger.Printf("WARN: unexpected cmpress seq nr: expected %v, got %v",
+				c.mc.sequence, compressionSequence)
+		}
+		c.mc.invalid = true
 	}
 	c.mc.sequence = compressionSequence + 1
 	c.mc.compressSequence = c.mc.sequence
@@ -218,10 +219,9 @@ func (mc *mysqlConn) writeCompressed(packets []byte) (int, error) {
 func (mc *mysqlConn) writeCompressedPacket(data []byte, uncompressedLen int) error {
 	comprLength := len(data) - 7
 	if debugTrace {
-		mc.cfg.Logger.Print(
-			fmt.Sprintf(
-				"writeCompressedPacket: comprLength=%v, uncompressedLen=%v, seq=%v",
-				comprLength, uncompressedLen, mc.compressSequence))
+		traceLogger.Printf(
+			"writeCompressedPacket: comprLength=%v, uncompressedLen=%v, seq=%v",
+			comprLength, uncompressedLen, mc.compressSequence)
 	}
 
 	// compression header
@@ -237,7 +237,7 @@ func (mc *mysqlConn) writeCompressedPacket(data []byte, uncompressedLen int) err
 	data[6] = byte(0xff & (uncompressedLen >> 16))
 
 	if _, err := mc.netConn.Write(data); err != nil {
-		mc.cfg.Logger.Print(err)
+		mc.log("writing compressed packet:", err)
 		return err
 	}
 

connection.go

@@ -37,6 +37,7 @@ type mysqlConn struct {
 	compressSequence uint8
 	parseTime        bool
 	compress         bool
+	invalid          bool // true when the connection is in invalid state and will be closed later.
 
 	// for context support (Go 1.8+)
 	watching bool
@@ -132,7 +133,6 @@ func (mc *mysqlConn) Begin() (driver.Tx, error) {
 
 func (mc *mysqlConn) begin(readOnly bool) (driver.Tx, error) {
 	if mc.closed.Load() {
-		mc.cfg.Logger.Print(ErrInvalidConn)
 		return nil, driver.ErrBadConn
 	}
 	var q string
@@ -173,7 +173,7 @@ func (mc *mysqlConn) cleanup() {
 		return
 	}
 	if err := mc.netConn.Close(); err != nil {
-		mc.cfg.Logger.Print(err)
+		mc.log("closing connection:", err)
 	}
 	mc.clearResult()
 }
@@ -698,5 +698,5 @@ func (mc *mysqlConn) ResetSession(ctx context.Context) error {
 // IsValid implements driver.Validator interface
 // (From Go 1.15)
 func (mc *mysqlConn) IsValid() bool {
-	return !mc.closed.Load()
+	return !mc.closed.Load() && !mc.invalid
 }

errors.go

@@ -37,18 +37,41 @@ var (
 	errBadConnNoWrite = errors.New("bad connection")
 )
 
-var defaultLogger = Logger(log.New(os.Stderr, "[mysql] ", log.Ldate|log.Ltime|log.Lshortfile))
+// traceLogger is used for debug trace log.
+var traceLogger *log.Logger
+
+func init() {
+	if debugTrace {
+		traceLogger = log.New(os.Stderr, "[mysql.trace]", log.Lmicroseconds|log.Lshortfile)
+	}
+}
+
+func trace(format string, v ...any) {
+	if debugTrace {
+		traceLogger.Printf(format, v...)
+	}
+}
 
 // Logger is used to log critical error messages.
 type Logger interface {
-	Print(v ...interface{})
+	Print(v ...any)
+}
+
+var defaultLogger = Logger(log.New(os.Stderr, "[mysql] ", log.Ldate|log.Ltime|log.Lshortfile))
+
+func (mc *mysqlConn) log(v ...any) {
+	mc.cfg.Logger.Print(v...)
+}
+
+func (mc *mysqlConn) logf(format string, v ...any) {
+	mc.cfg.Logger.Print(fmt.Sprintf(format, v...))
 }
 
 // NopLogger is a nop implementation of the Logger interface.
 type NopLogger struct{}
 
 // Print implements Logger interface.
-func (nl *NopLogger) Print(_ ...interface{}) {}
+func (nl *NopLogger) Print(_ ...any) {}
 
 // SetLogger is used to set the default logger for critical errors.
 // The initial logger is os.Stderr.

packets.go

@@ -28,6 +28,7 @@ import (
 // Read packet to buffer 'data'
 func (mc *mysqlConn) readPacket() ([]byte, error) {
 	var prevData []byte
+	var rerr error = nil
 	for {
 		// read packet header
 		data, err := mc.packetReader.readNext(4)
@@ -46,10 +47,18 @@ func (mc *mysqlConn) readPacket() ([]byte, error) {
 		// packet length [24 bit]
 		pktLen := int(uint32(data[0]) | uint32(data[1])<<8 | uint32(data[2])<<16)
 
-		if !mc.compress { // MySQL and MariaDB doesn't check packet nr in compressed packet.
+		if mc.compress {
+			// MySQL and MariaDB doesn't check packet nr in compressed packet.
+			if debugTrace && data[3] != mc.compressSequence {
+				mc.cfg.Logger.Print
+			}
+			mc.compressSequence = data[3]+1
+		} else mc.compress {
 			// check packet sync [8 bit]
 			if data[3] != mc.sequence {
 				mc.cfg.Logger.Print(fmt.Sprintf("[warn] unexpected seq nr: expected %v, got %v", mc.sequence, data[3]))
+				mc.invalid = true
+				rerr = ErrInvalidConn
 			}
 			mc.sequence++
 		}
@@ -117,7 +126,7 @@ func (mc *mysqlConn) writePacket(data []byte) error {
 
 		// Write packet
 		if debugTrace {
-			mc.cfg.Logger.Print(fmt.Sprintf("writePacket: size=%v seq=%v", size, mc.sequence))
+			traceLogger.Printf("writePacket: size=%v seq=%v", size, mc.sequence)
 		}
 		if mc.writeTimeout > 0 {
 			if err := mc.netConn.SetWriteDeadline(time.Now().Add(mc.writeTimeout)); err != nil {