licenses/github.com/go-sql-driver/mysql/errors_test.go

@@ -16,9 +16,9 @@ import (
 )
 
 func TestErrorsSetLogger(t *testing.T) {
-	previous := errLog
+	previous := defaultLogger
 	defer func() {
-		errLog = previous
+		defaultLogger = previous
 	}()
 
 	// set up logger
@@ -28,7 +28,7 @@ func TestErrorsSetLogger(t *testing.T) {
 
 	// print
 	SetLogger(logger)
-	errLog.Print("test")
+	defaultLogger.Print("test")
 
 	// check result
 	if actual := buffer.String(); actual != expected {

licenses/github.com/go-sql-driver/mysql/fields.go

@@ -18,7 +18,7 @@ func (mf *mysqlField) typeDatabaseName() string {
 	case fieldTypeBit:
 		return "BIT"
 	case fieldTypeBLOB:
-		if mf.charSet != collations[binaryCollation] {
+		if mf.charSet != binaryCollationID {
 			return "TEXT"
 		}
 		return "BLOB"
@@ -37,6 +37,9 @@ func (mf *mysqlField) typeDatabaseName() string {
 	case fieldTypeGeometry:
 		return "GEOMETRY"
 	case fieldTypeInt24:
+		if mf.flags&flagUnsigned != 0 {
+			return "UNSIGNED MEDIUMINT"
+		}
 		return "MEDIUMINT"
 	case fieldTypeJSON:
 		return "JSON"
@@ -46,7 +49,7 @@ func (mf *mysqlField) typeDatabaseName() string {
 		}
 		return "INT"
 	case fieldTypeLongBLOB:
-		if mf.charSet != collations[binaryCollation] {
+		if mf.charSet != binaryCollationID {
 			return "LONGTEXT"
 		}
 		return "LONGBLOB"
@@ -56,7 +59,7 @@ func (mf *mysqlField) typeDatabaseName() string {
 		}
 		return "BIGINT"
 	case fieldTypeMediumBLOB:
-		if mf.charSet != collations[binaryCollation] {
+		if mf.charSet != binaryCollationID {
 			return "MEDIUMTEXT"
 		}
 		return "MEDIUMBLOB"
@@ -74,7 +77,12 @@ func (mf *mysqlField) typeDatabaseName() string {
 		}
 		return "SMALLINT"
 	case fieldTypeString:
-		if mf.charSet == collations[binaryCollation] {
+		if mf.flags&flagEnum != 0 {
+			return "ENUM"
+		} else if mf.flags&flagSet != 0 {
+			return "SET"
+		}
+		if mf.charSet == binaryCollationID {
 			return "BINARY"
 		}
 		return "CHAR"
@@ -88,17 +96,17 @@ func (mf *mysqlField) typeDatabaseName() string {
 		}
 		return "TINYINT"
 	case fieldTypeTinyBLOB:
-		if mf.charSet != collations[binaryCollation] {
+		if mf.charSet != binaryCollationID {
 			return "TINYTEXT"
 		}
 		return "TINYBLOB"
 	case fieldTypeVarChar:
-		if mf.charSet == collations[binaryCollation] {
+		if mf.charSet == binaryCollationID {
 			return "VARBINARY"
 		}
 		return "VARCHAR"
 	case fieldTypeVarString:
-		if mf.charSet == collations[binaryCollation] {
+		if mf.charSet == binaryCollationID {
 			return "VARBINARY"
 		}
 		return "VARCHAR"
@@ -123,7 +131,9 @@ var (
 	scanTypeUint16     = reflect.TypeOf(uint16(0))
 	scanTypeUint32     = reflect.TypeOf(uint32(0))
 	scanTypeUint64     = reflect.TypeOf(uint64(0))
-	scanTypeRawBytes  = reflect.TypeOf(sql.RawBytes{})
+	scanTypeString     = reflect.TypeOf("")
+	scanTypeNullString = reflect.TypeOf(sql.NullString{})
+	scanTypeBytes      = reflect.TypeOf([]byte{})
 	scanTypeUnknown    = reflect.TypeOf(new(interface{}))
 )
 
@@ -187,12 +197,18 @@ func (mf *mysqlField) scanType() reflect.Type {
 		}
 		return scanTypeNullFloat
 
+	case fieldTypeBit, fieldTypeTinyBLOB, fieldTypeMediumBLOB, fieldTypeLongBLOB,
+		fieldTypeBLOB, fieldTypeVarString, fieldTypeString, fieldTypeGeometry:
+		if mf.charSet == binaryCollationID {
+			return scanTypeBytes
+		}
+		fallthrough
 	case fieldTypeDecimal, fieldTypeNewDecimal, fieldTypeVarChar,
-		fieldTypeBit, fieldTypeEnum, fieldTypeSet, fieldTypeTinyBLOB,
-		fieldTypeMediumBLOB, fieldTypeLongBLOB, fieldTypeBLOB,
-		fieldTypeVarString, fieldTypeString, fieldTypeGeometry, fieldTypeJSON,
-		fieldTypeTime:
-		return scanTypeRawBytes
+		fieldTypeEnum, fieldTypeSet, fieldTypeJSON, fieldTypeTime:
+		if mf.flags&flagNotNULL != 0 {
+			return scanTypeString
+		}
+		return scanTypeNullString
 
 	case fieldTypeDate, fieldTypeNewDate,
 		fieldTypeTimestamp, fieldTypeDateTime:

licenses/github.com/go-sql-driver/mysql/fuzz.go

-// Go MySQL Driver - A MySQL-Driver for Go's database/sql package.
-//
-// Copyright 2020 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/.
-
-//go:build gofuzz
-// +build gofuzz
-
-package mysql
-
-import (
-	"database/sql"
-)
-
-func Fuzz(data []byte) int {
-	db, err := sql.Open("mysql", string(data))
-	if err != nil {
-		return 0
-	}
-	db.Close()
-	return 1
-}

licenses/github.com/go-sql-driver/mysql/go.mod

 module github.com/go-sql-driver/mysql
 
-go 1.13
+go 1.18
+
+require filippo.io/edwards25519 v1.1.0

licenses/github.com/go-sql-driver/mysql/go.sum

+filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
+filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=

licenses/github.com/go-sql-driver/mysql/infile.go

@@ -93,7 +93,7 @@ func deferredClose(err *error, closer io.Closer) {
 
 const defaultPacketSize = 16 * 1024 // 16KB is small enough for disk readahead and large enough for TCP
 
-func (mc *mysqlConn) handleInFileRequest(name string) (err error) {
+func (mc *okHandler) handleInFileRequest(name string) (err error) {
 	var rdr io.Reader
 	var data []byte
 	packetSize := defaultPacketSize
@@ -116,10 +116,10 @@ func (mc *mysqlConn) handleInFileRequest(name string) (err error) {
 					defer deferredClose(&err, cl)
 				}
 			} else {
-				err = fmt.Errorf("Reader '%s' is <nil>", name)
+				err = fmt.Errorf("reader '%s' is <nil>", name)
 			}
 		} else {
-			err = fmt.Errorf("Reader '%s' is not registered", name)
+			err = fmt.Errorf("reader '%s' is not registered", name)
 		}
 	} else { // File
 		name = strings.Trim(name, `"`)
@@ -154,7 +154,7 @@ func (mc *mysqlConn) handleInFileRequest(name string) (err error) {
 		for err == nil {
 			n, err = rdr.Read(data[4:])
 			if n > 0 {
-				if ioErr := mc.writePacket(data[:4+n]); ioErr != nil {
+				if ioErr := mc.conn().writePacket(data[:4+n]); ioErr != nil {
 					return ioErr
 				}
 			}
@@ -168,7 +168,7 @@ func (mc *mysqlConn) handleInFileRequest(name string) (err error) {
 	if data == nil {
 		data = make([]byte, 4)
 	}
-	if ioErr := mc.writePacket(data[:4]); ioErr != nil {
+	if ioErr := mc.conn().writePacket(data[:4]); ioErr != nil {
 		return ioErr
 	}
 
@@ -177,6 +177,6 @@ func (mc *mysqlConn) handleInFileRequest(name string) (err error) {
 		return mc.readResultOK()
 	}
 
-	mc.readPacket()
+	mc.conn().readPacket()
 	return err
 }

licenses/github.com/go-sql-driver/mysql/nulltime.go

@@ -59,7 +59,7 @@ func (nt *NullTime) Scan(value interface{}) (err error) {
 	}
 
 	nt.Valid = false
-	return fmt.Errorf("Can't convert %T to time.Time", value)
+	return fmt.Errorf("can't convert %T to time.Time", value)
 }
 
 // Value implements the driver Valuer interface.

licenses/github.com/go-sql-driver/mysql/packets.go

@@ -14,10 +14,10 @@ import (
 	"database/sql/driver"
 	"encoding/binary"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"io"
 	"math"
+	"strconv"
 	"time"
 )
 
@@ -34,7 +34,7 @@ func (mc *mysqlConn) readPacket() ([]byte, error) {
 			if cerr := mc.canceled.Value(); cerr != nil {
 				return nil, cerr
 			}
-			errLog.Print(err)
+			mc.cfg.Logger.Print(err)
 			mc.Close()
 			return nil, ErrInvalidConn
 		}
@@ -44,6 +44,7 @@ func (mc *mysqlConn) readPacket() ([]byte, error) {
 
 		// check packet sync [8 bit]
 		if data[3] != mc.sequence {
+			mc.Close()
 			if data[3] > mc.sequence {
 				return nil, ErrPktSyncMul
 			}
@@ -56,7 +57,7 @@ func (mc *mysqlConn) readPacket() ([]byte, error) {
 		if pktLen == 0 {
 			// there was no previous packet
 			if prevData == nil {
-				errLog.Print(ErrMalformPkt)
+				mc.cfg.Logger.Print(ErrMalformPkt)
 				mc.Close()
 				return nil, ErrInvalidConn
 			}
@@ -70,7 +71,7 @@ func (mc *mysqlConn) readPacket() ([]byte, error) {
 			if cerr := mc.canceled.Value(); cerr != nil {
 				return nil, cerr
 			}
-			errLog.Print(err)
+			mc.cfg.Logger.Print(err)
 			mc.Close()
 			return nil, ErrInvalidConn
 		}
@@ -97,34 +98,6 @@ func (mc *mysqlConn) writePacket(data []byte) error {
 		return ErrPktTooLarge
 	}
 
-	// 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.reset {
-		mc.reset = false
-		conn := mc.netConn
-		if mc.rawConn != nil {
-			conn = mc.rawConn
-		}
-		var err error
-		if mc.cfg.CheckConnLiveness {
-			if mc.cfg.ReadTimeout != 0 {
-				err = conn.SetReadDeadline(time.Now().Add(mc.cfg.ReadTimeout))
-			}
-			if err == nil {
-				err = connCheck(conn)
-			}
-		}
-		if err != nil {
-			errLog.Print("closing bad idle connection: ", err)
-			mc.Close()
-			return driver.ErrBadConn
-		}
-	}
-
 	for {
 		var size int
 		if pktLen >= maxPacketSize {
@@ -161,7 +134,7 @@ func (mc *mysqlConn) writePacket(data []byte) error {
 		// Handle error
 		if err == nil { // n != len(data)
 			mc.cleanup()
-			errLog.Print(ErrMalformPkt)
+			mc.cfg.Logger.Print(ErrMalformPkt)
 		} else {
 			if cerr := mc.canceled.Value(); cerr != nil {
 				return cerr
@@ -171,7 +144,7 @@ func (mc *mysqlConn) writePacket(data []byte) error {
 				return errBadConnNoWrite
 			}
 			mc.cleanup()
-			errLog.Print(err)
+			mc.cfg.Logger.Print(err)
 		}
 		return ErrInvalidConn
 	}
@@ -239,7 +212,7 @@ func (mc *mysqlConn) readHandshakePacket() (data []byte, plugin string, err erro
 		// reserved (all [00]) [10 bytes]
 		pos += 1 + 2 + 2 + 1 + 10
 
-		// second part of the password cipher [mininum 13 bytes],
+		// second part of the password cipher [minimum 13 bytes],
 		// where len=MAX(13, length of auth-plugin-data - 8)
 		//
 		// The web documentation is ambiguous about the length. However,
@@ -285,6 +258,7 @@ func (mc *mysqlConn) writeHandshakeResponsePacket(authResp []byte, plugin string
 		clientLocalFiles |
 		clientPluginAuth |
 		clientMultiResults |
+		clientConnectAttrs |
 		mc.flags&clientLongFlag
 
 	if mc.cfg.ClientFoundRows {
@@ -318,11 +292,17 @@ func (mc *mysqlConn) writeHandshakeResponsePacket(authResp []byte, plugin string
 		pktLen += n + 1
 	}
 
+	// encode length of the connection attributes
+	var connAttrsLEIBuf [9]byte
+	connAttrsLen := len(mc.connector.encodedAttributes)
+	connAttrsLEI := appendLengthEncodedInteger(connAttrsLEIBuf[:0], uint64(connAttrsLen))
+	pktLen += len(connAttrsLEI) + len(mc.connector.encodedAttributes)
+
 	// Calculate packet length and get buffer with that size
-	data, err := mc.buf.takeSmallBuffer(pktLen + 4)
+	data, err := mc.buf.takeBuffer(pktLen + 4)
 	if err != nil {
 		// cannot take the buffer. Something must be wrong with the connection
-		errLog.Print(err)
+		mc.cfg.Logger.Print(err)
 		return errBadConnNoWrite
 	}
 
@@ -338,14 +318,18 @@ func (mc *mysqlConn) writeHandshakeResponsePacket(authResp []byte, plugin string
 	data[10] = 0x00
 	data[11] = 0x00
 
-	// Charset [1 byte]
+	// Collation ID [1 byte]
+	cname := mc.cfg.Collation
+	if cname == "" {
+		cname = defaultCollation
+	}
 	var found bool
-	data[12], found = collations[mc.cfg.Collation]
+	data[12], found = collations[cname]
 	if !found {
 		// Note possibility for false negatives:
 		// could be triggered  although the collation is valid if the
 		// collations map does not contain entries the server supports.
-		return errors.New("unknown collation")
+		return fmt.Errorf("unknown collation: %q", cname)
 	}
 
 	// Filler [23 bytes] (all 0x00)
@@ -394,6 +378,10 @@ func (mc *mysqlConn) writeHandshakeResponsePacket(authResp []byte, plugin string
 	data[pos] = 0x00
 	pos++
 
+	// Connection Attributes
+	pos += copy(data[pos:], connAttrsLEI)
+	pos += copy(data[pos:], []byte(mc.connector.encodedAttributes))
+
 	// Send Auth packet
 	return mc.writePacket(data[:pos])
 }
@@ -404,7 +392,7 @@ func (mc *mysqlConn) writeAuthSwitchPacket(authData []byte) error {
 	data, err := mc.buf.takeSmallBuffer(pktLen)
 	if err != nil {
 		// cannot take the buffer. Something must be wrong with the connection
-		errLog.Print(err)
+		mc.cfg.Logger.Print(err)
 		return errBadConnNoWrite
 	}
 
@@ -424,7 +412,7 @@ func (mc *mysqlConn) writeCommandPacket(command byte) error {
 	data, err := mc.buf.takeSmallBuffer(4 + 1)
 	if err != nil {
 		// cannot take the buffer. Something must be wrong with the connection
-		errLog.Print(err)
+		mc.cfg.Logger.Print(err)
 		return errBadConnNoWrite
 	}
 
@@ -443,7 +431,7 @@ func (mc *mysqlConn) writeCommandPacketStr(command byte, arg string) error {
 	data, err := mc.buf.takeBuffer(pktLen + 4)
 	if err != nil {
 		// cannot take the buffer. Something must be wrong with the connection
-		errLog.Print(err)
+		mc.cfg.Logger.Print(err)
 		return errBadConnNoWrite
 	}
 
@@ -464,7 +452,7 @@ func (mc *mysqlConn) writeCommandPacketUint32(command byte, arg uint32) error {
 	data, err := mc.buf.takeSmallBuffer(4 + 1 + 4)
 	if err != nil {
 		// cannot take the buffer. Something must be wrong with the connection
-		errLog.Print(err)
+		mc.cfg.Logger.Print(err)
 		return errBadConnNoWrite
 	}
 
@@ -495,7 +483,9 @@ func (mc *mysqlConn) readAuthResult() ([]byte, string, error) {
 	switch data[0] {
 
 	case iOK:
-		return nil, "", mc.handleOkPacket(data)
+		// resultUnchanged, since auth happens before any queries or
+		// commands have been executed.
+		return nil, "", mc.resultUnchanged().handleOkPacket(data)
 
 	case iAuthMoreData:
 		return data[1:], "", err
@@ -518,9 +508,9 @@ func (mc *mysqlConn) readAuthResult() ([]byte, string, error) {
 	}
 }
 
-// Returns error if Packet is not an 'Result OK'-Packet
-func (mc *mysqlConn) readResultOK() error {
-	data, err := mc.readPacket()
+// Returns error if Packet is not a 'Result OK'-Packet
+func (mc *okHandler) readResultOK() error {
+	data, err := mc.conn().readPacket()
 	if err != nil {
 		return err
 	}
@@ -528,13 +518,17 @@ func (mc *mysqlConn) readResultOK() error {
 	if data[0] == iOK {
 		return mc.handleOkPacket(data)
 	}
-	return mc.handleErrorPacket(data)
+	return mc.conn().handleErrorPacket(data)
 }
 
 // Result Set Header Packet
 // http://dev.mysql.com/doc/internals/en/com-query-response.html#packet-ProtocolText::Resultset
-func (mc *mysqlConn) readResultSetHeaderPacket() (int, error) {
-	data, err := mc.readPacket()
+func (mc *okHandler) readResultSetHeaderPacket() (int, error) {
+	// handleOkPacket replaces both values; other cases leave the values unchanged.
+	mc.result.affectedRows = append(mc.result.affectedRows, 0)
+	mc.result.insertIds = append(mc.result.insertIds, 0)
+
+	data, err := mc.conn().readPacket()
 	if err == nil {
 		switch data[0] {
 
@@ -542,20 +536,17 @@ func (mc *mysqlConn) readResultSetHeaderPacket() (int, error) {
 			return 0, mc.handleOkPacket(data)
 
 		case iERR:
-			return 0, mc.handleErrorPacket(data)
+			return 0, mc.conn().handleErrorPacket(data)
 
 		case iLocalInFile:
 			return 0, mc.handleInFileRequest(string(data[1:]))
 		}
 
 		// column count
-		num, _, n := readLengthEncodedInteger(data)
-		if n-len(data) == 0 {
+		num, _, _ := readLengthEncodedInteger(data)
+		// ignore remaining data in the packet. see #1478.
 		return int(num), nil
 	}
-
-		return 0, ErrMalformPkt
-	}
 	return 0, err
 }
 
@@ -607,18 +598,61 @@ func readStatus(b []byte) statusFlag {
 	return statusFlag(b[0]) | statusFlag(b[1])<<8
 }
 
+// Returns an instance of okHandler for codepaths where mysqlConn.result doesn't
+// need to be cleared first (e.g. during authentication, or while additional
+// resultsets are being fetched.)
+func (mc *mysqlConn) resultUnchanged() *okHandler {
+	return (*okHandler)(mc)
+}
+
+// okHandler represents the state of the connection when mysqlConn.result has
+// been prepared for processing of OK packets.
+//
+// To correctly populate mysqlConn.result (updated by handleOkPacket()), all
+// callpaths must either:
+//
+// 1. first clear it using clearResult(), or
+// 2. confirm that they don't need to (by calling resultUnchanged()).
+//
+// Both return an instance of type *okHandler.
+type okHandler mysqlConn
+
+// Exposes the underlying type's methods.
+func (mc *okHandler) conn() *mysqlConn {
+	return (*mysqlConn)(mc)
+}
+
+// clearResult clears the connection's stored affectedRows and insertIds
+// fields.
+//
+// It returns a handler that can process OK responses.
+func (mc *mysqlConn) clearResult() *okHandler {
+	mc.result = mysqlResult{}
+	return (*okHandler)(mc)
+}
+
 // Ok Packet
 // http://dev.mysql.com/doc/internals/en/generic-response-packets.html#packet-OK_Packet
-func (mc *mysqlConn) handleOkPacket(data []byte) error {
+func (mc *okHandler) handleOkPacket(data []byte) error {
 	var n, m int
+	var affectedRows, insertId uint64
 
 	// 0x00 [1 byte]
 
 	// Affected rows [Length Coded Binary]
-	mc.affectedRows, _, n = readLengthEncodedInteger(data[1:])
+	affectedRows, _, n = readLengthEncodedInteger(data[1:])
 
 	// Insert id [Length Coded Binary]
-	mc.insertId, _, m = readLengthEncodedInteger(data[1+n:])
+	insertId, _, m = readLengthEncodedInteger(data[1+n:])
+
+	// Update for the current statement result (only used by
+	// readResultSetHeaderPacket).
+	if len(mc.result.affectedRows) > 0 {
+		mc.result.affectedRows[len(mc.result.affectedRows)-1] = int64(affectedRows)
+	}
+	if len(mc.result.insertIds) > 0 {
+		mc.result.insertIds[len(mc.result.insertIds)-1] = int64(insertId)
+	}
 
 	// server_status [2 bytes]
 	mc.status = readStatus(data[1+n+m : 1+n+m+2])
@@ -769,7 +803,8 @@ func (rows *textRows) readRow(dest []driver.Value) error {
 
 	for i := range dest {
 		// Read bytes and convert to string
-		dest[i], isNull, n, err = readLengthEncodedString(data[pos:])
+		var buf []byte
+		buf, isNull, n, err = readLengthEncodedString(data[pos:])
 		pos += n
 
 		if err != nil {
@@ -781,19 +816,40 @@ func (rows *textRows) readRow(dest []driver.Value) error {
 			continue
 		}
 
-		if !mc.parseTime {
-			continue
-		}
-
-		// Parse time field
 		switch rows.rs.columns[i].fieldType {
 		case fieldTypeTimestamp,
 			fieldTypeDateTime,
 			fieldTypeDate,
 			fieldTypeNewDate:
-			if dest[i], err = parseDateTime(dest[i].([]byte), mc.cfg.Loc); err != nil {
-				return err
+			if mc.parseTime {
+				dest[i], err = parseDateTime(buf, mc.cfg.Loc)
+			} else {
+				dest[i] = buf
 			}
+
+		case fieldTypeTiny, fieldTypeShort, fieldTypeInt24, fieldTypeYear, fieldTypeLong:
+			dest[i], err = strconv.ParseInt(string(buf), 10, 64)
+
+		case fieldTypeLongLong:
+			if rows.rs.columns[i].flags&flagUnsigned != 0 {
+				dest[i], err = strconv.ParseUint(string(buf), 10, 64)
+			} else {
+				dest[i], err = strconv.ParseInt(string(buf), 10, 64)
+			}
+
+		case fieldTypeFloat:
+			var d float64
+			d, err = strconv.ParseFloat(string(buf), 32)
+			dest[i] = float32(d)
+
+		case fieldTypeDouble:
+			dest[i], err = strconv.ParseFloat(string(buf), 64)
+
+		default:
+			dest[i] = buf
+		}
+		if err != nil {
+			return err
 		}
 	}
 
@@ -938,7 +994,7 @@ func (stmt *mysqlStmt) writeExecutePacket(args []driver.Value) error {
 	}
 	if err != nil {
 		// cannot take the buffer. Something must be wrong with the connection
-		errLog.Print(err)
+		mc.cfg.Logger.Print(err)
 		return errBadConnNoWrite
 	}
 
@@ -1116,7 +1172,7 @@ func (stmt *mysqlStmt) writeExecutePacket(args []driver.Value) error {
 				if v.IsZero() {
 					b = append(b, "0000-00-00"...)
 				} else {
-					b, err = appendDateTime(b, v.In(mc.cfg.Loc))
+					b, err = appendDateTime(b, v.In(mc.cfg.Loc), mc.cfg.timeTruncate)
 					if err != nil {
 						return err
 					}
@@ -1137,7 +1193,7 @@ func (stmt *mysqlStmt) writeExecutePacket(args []driver.Value) error {
 		if valuesCap != cap(paramValues) {
 			data = append(data[:pos], paramValues...)
 			if err = mc.buf.store(data); err != nil {
-				errLog.Print(err)
+				mc.cfg.Logger.Print(err)
 				return errBadConnNoWrite
 			}
 		}
@@ -1149,7 +1205,9 @@ func (stmt *mysqlStmt) writeExecutePacket(args []driver.Value) error {
 	return mc.writePacket(data)
 }
 
-func (mc *mysqlConn) discardResults() error {
+// For each remaining resultset in the stream, discards its rows and updates
+// mc.affectedRows and mc.insertIds.
+func (mc *okHandler) discardResults() error {
 	for mc.status&statusMoreResultsExists != 0 {
 		resLen, err := mc.readResultSetHeaderPacket()
 		if err != nil {
@@ -1157,11 +1215,11 @@ func (mc *mysqlConn) discardResults() error {
 		}
 		if resLen > 0 {
 			// columns
-			if err := mc.readUntilEOF(); err != nil {
+			if err := mc.conn().readUntilEOF(); err != nil {
 				return err
 			}
 			// rows
-			if err := mc.readUntilEOF(); err != nil {
+			if err := mc.conn().readUntilEOF(); err != nil {
 				return err
 			}
 		}

licenses/github.com/go-sql-driver/mysql/packets_test.go

@@ -96,9 +96,11 @@ var _ net.Conn = new(mockConn)
 
 func newRWMockConn(sequence uint8) (*mockConn, *mysqlConn) {
 	conn := new(mockConn)
+	connector := newConnector(NewConfig())
 	mc := &mysqlConn{
 		buf:              newBuffer(conn),
-		cfg:              NewConfig(),
+		cfg:              connector.cfg,
+		connector:        connector,
 		netConn:          conn,
 		closech:          make(chan struct{}),
 		maxAllowedPacket: defaultMaxAllowedPacket,
@@ -128,30 +130,34 @@ func TestReadPacketSingleByte(t *testing.T) {
 }
 
 func TestReadPacketWrongSequenceID(t *testing.T) {
-	conn := new(mockConn)
-	mc := &mysqlConn{
-		buf: newBuffer(conn),
-	}
-
-	// too low sequence id
-	conn.data = []byte{0x01, 0x00, 0x00, 0x00, 0xff}
-	conn.maxReads = 1
-	mc.sequence = 1
+	for _, testCase := range []struct {
+		ClientSequenceID byte
+		ServerSequenceID byte
+		ExpectedErr      error
+	}{
+		{
+			ClientSequenceID: 1,
+			ServerSequenceID: 0,
+			ExpectedErr:      ErrPktSync,
+		},
+		{
+			ClientSequenceID: 0,
+			ServerSequenceID: 0x42,
+			ExpectedErr:      ErrPktSyncMul,
+		},
+	} {
+		conn, mc := newRWMockConn(testCase.ClientSequenceID)
+
+		conn.data = []byte{0x01, 0x00, 0x00, testCase.ServerSequenceID, 0xff}
 		_, err := mc.readPacket()
-	if err != ErrPktSync {
-		t.Errorf("expected ErrPktSync, got %v", err)
+		if err != testCase.ExpectedErr {
+			t.Errorf("expected %v, got %v", testCase.ExpectedErr, err)
 		}
 
-	// reset
-	conn.reads = 0
-	mc.sequence = 0
-	mc.buf = newBuffer(conn)
-
-	// too high sequence id
-	conn.data = []byte{0x01, 0x00, 0x00, 0x42, 0xff}
-	_, err = mc.readPacket()
-	if err != ErrPktSyncMul {
-		t.Errorf("expected ErrPktSyncMul, got %v", err)
+		// connection should not be returned to the pool in this state
+		if mc.IsValid() {
+			t.Errorf("expected IsValid() to be false")
+		}
 	}
 }
 
@@ -179,7 +185,7 @@ func TestReadPacketSplit(t *testing.T) {
 	data[4] = 0x11
 	data[maxPacketSize+3] = 0x22
 
-	// 2nd packet has payload length 0 and squence id 1
+	// 2nd packet has payload length 0 and sequence id 1
 	// 00 00 00 01
 	data[pkt2ofs+3] = 0x01
 
@@ -211,7 +217,7 @@ func TestReadPacketSplit(t *testing.T) {
 	data[pkt2ofs+4] = 0x33
 	data[pkt2ofs+maxPacketSize+3] = 0x44
 
-	// 3rd packet has payload length 0 and squence id 2
+	// 3rd packet has payload length 0 and sequence id 2
 	// 00 00 00 02
 	data[pkt3ofs+3] = 0x02
 
@@ -265,6 +271,7 @@ func TestReadPacketFail(t *testing.T) {
 	mc := &mysqlConn{
 		buf:     newBuffer(conn),
 		closech: make(chan struct{}),
+		cfg:     NewConfig(),
 	}
 
 	// illegal empty (stand-alone) packet

licenses/github.com/go-sql-driver/mysql/result.go

@@ -8,15 +8,43 @@
 
 package mysql
 
+import "database/sql/driver"
+
+// Result exposes data not available through *connection.Result.
+//
+// This is accessible by executing statements using sql.Conn.Raw() and
+// downcasting the returned result:
+//
+//	res, err := rawConn.Exec(...)
+//	res.(mysql.Result).AllRowsAffected()
+type Result interface {
+	driver.Result
+	// AllRowsAffected returns a slice containing the affected rows for each
+	// executed statement.
+	AllRowsAffected() []int64
+	// AllLastInsertIds returns a slice containing the last inserted ID for each
+	// executed statement.
+	AllLastInsertIds() []int64
+}
+
 type mysqlResult struct {
-	affectedRows int64
-	insertId     int64
+	// One entry in both slices is created for every executed statement result.
+	affectedRows []int64
+	insertIds    []int64
 }
 
 func (res *mysqlResult) LastInsertId() (int64, error) {
-	return res.insertId, nil
+	return res.insertIds[len(res.insertIds)-1], nil
 }
 
 func (res *mysqlResult) RowsAffected() (int64, error) {
-	return res.affectedRows, nil
+	return res.affectedRows[len(res.affectedRows)-1], nil
+}
+
+func (res *mysqlResult) AllLastInsertIds() []int64 {
+	return append([]int64{}, res.insertIds...) // defensive copy
+}
+
+func (res *mysqlResult) AllRowsAffected() []int64 {
+	return append([]int64{}, res.affectedRows...) // defensive copy
 }

licenses/github.com/go-sql-driver/mysql/rows.go

@@ -123,7 +123,8 @@ func (rows *mysqlRows) Close() (err error) {
 		err = mc.readUntilEOF()
 	}
 	if err == nil {
-		if err = mc.discardResults(); err != nil {
+		handleOk := mc.clearResult()
+		if err = handleOk.discardResults(); err != nil {
 			return err
 		}
 	}
@@ -160,7 +161,15 @@ func (rows *mysqlRows) nextResultSet() (int, error) {
 		return 0, io.EOF
 	}
 	rows.rs = resultSet{}
-	return rows.mc.readResultSetHeaderPacket()
+	// rows.mc.affectedRows and rows.mc.insertIds accumulate on each call to
+	// nextResultSet.
+	resLen, err := rows.mc.resultUnchanged().readResultSetHeaderPacket()
+	if err != nil {
+		// Clean up about multi-results flag
+		rows.rs.done = true
+		rows.mc.status = rows.mc.status & (^statusMoreResultsExists)
+	}
+	return resLen, err
 }
 
 func (rows *mysqlRows) nextNotEmptyResultSet() (int, error) {

licenses/github.com/go-sql-driver/mysql/statement.go

@@ -51,7 +51,7 @@ func (stmt *mysqlStmt) CheckNamedValue(nv *driver.NamedValue) (err error) {
 
 func (stmt *mysqlStmt) Exec(args []driver.Value) (driver.Result, error) {
 	if stmt.mc.closed.Load() {
-		errLog.Print(ErrInvalidConn)
+		stmt.mc.cfg.Logger.Print(ErrInvalidConn)
 		return nil, driver.ErrBadConn
 	}
 	// Send command
@@ -61,12 +61,10 @@ func (stmt *mysqlStmt) Exec(args []driver.Value) (driver.Result, error) {
 	}
 
 	mc := stmt.mc
-
-	mc.affectedRows = 0
-	mc.insertId = 0
+	handleOk := stmt.mc.clearResult()
 
 	// Read Result
-	resLen, err := mc.readResultSetHeaderPacket()
+	resLen, err := handleOk.readResultSetHeaderPacket()
 	if err != nil {
 		return nil, err
 	}
@@ -83,14 +81,12 @@ func (stmt *mysqlStmt) Exec(args []driver.Value) (driver.Result, error) {
 		}
 	}
 
-	if err := mc.discardResults(); err != nil {
+	if err := handleOk.discardResults(); err != nil {
 		return nil, err
 	}
 
-	return &mysqlResult{
-		affectedRows: int64(mc.affectedRows),
-		insertId:     int64(mc.insertId),
-	}, nil
+	copied := mc.result
+	return &copied, nil
 }
 
 func (stmt *mysqlStmt) Query(args []driver.Value) (driver.Rows, error) {
@@ -99,7 +95,7 @@ func (stmt *mysqlStmt) Query(args []driver.Value) (driver.Rows, error) {
 
 func (stmt *mysqlStmt) query(args []driver.Value) (*binaryRows, error) {
 	if stmt.mc.closed.Load() {
-		errLog.Print(ErrInvalidConn)
+		stmt.mc.cfg.Logger.Print(ErrInvalidConn)
 		return nil, driver.ErrBadConn
 	}
 	// Send command
@@ -111,7 +107,8 @@ func (stmt *mysqlStmt) query(args []driver.Value) (*binaryRows, error) {
 	mc := stmt.mc
 
 	// Read Result
-	resLen, err := mc.readResultSetHeaderPacket()
+	handleOk := stmt.mc.clearResult()
+	resLen, err := handleOk.readResultSetHeaderPacket()
 	if err != nil {
 		return nil, err
 	}

licenses/github.com/go-sql-driver/mysql/utils.go

@@ -36,7 +36,7 @@ var (
 // registering it.
 //
 //	rootCertPool := x509.NewCertPool()
-//	pem, err := ioutil.ReadFile("/path/ca-cert.pem")
+//	pem, err := os.ReadFile("/path/ca-cert.pem")
 //	if err != nil {
 //	    log.Fatal(err)
 //	}
@@ -265,7 +265,11 @@ func parseBinaryDateTime(num uint64, data []byte, loc *time.Location) (driver.Va
 	return nil, fmt.Errorf("invalid DATETIME packet length %d", num)
 }
 
-func appendDateTime(buf []byte, t time.Time) ([]byte, error) {
+func appendDateTime(buf []byte, t time.Time, timeTruncate time.Duration) ([]byte, error) {
+	if timeTruncate > 0 {
+		t = t.Truncate(timeTruncate)
+	}
+
 	year, month, day := t.Date()
 	hour, min, sec := t.Clock()
 	nsec := t.Nanosecond()
@@ -616,6 +620,11 @@ func appendLengthEncodedInteger(b []byte, n uint64) []byte {
 		byte(n>>32), byte(n>>40), byte(n>>48), byte(n>>56))
 }
 
+func appendLengthEncodedString(b []byte, s string) []byte {
+	b = appendLengthEncodedInteger(b, uint64(len(s)))
+	return append(b, s...)
+}
+
 // reserveBuffer checks cap(buf) and expand buffer to len(buf) + appendSize.
 // If cap(buf) is not enough, reallocate new buffer.
 func reserveBuffer(buf []byte, appendSize int) []byte {

licenses/github.com/go-sql-driver/mysql/utils_test.go

@@ -239,6 +239,8 @@ func TestAppendDateTime(t *testing.T) {
 	tests := []struct {
 		t            time.Time
 		str          string
+		timeTruncate time.Duration
+		expectedErr  bool
 	}{
 		{
 			t:   time.Date(1234, 5, 6, 0, 0, 0, 0, time.UTC),
@@ -276,32 +278,73 @@ func TestAppendDateTime(t *testing.T) {
 			t:   time.Date(1, 1, 1, 0, 0, 0, 0, time.UTC),
 			str: "0001-01-01",
 		},
-	}
-	for _, v := range tests {
-		buf := make([]byte, 0, 32)
-		buf, _ = appendDateTime(buf, v.t)
-		if str := string(buf); str != v.str {
-			t.Errorf("appendDateTime(%v), have: %s, want: %s", v.t, str, v.str)
-		}
-	}
-
+		// Truncated time
+		{
+			t:            time.Date(1234, 5, 6, 0, 0, 0, 0, time.UTC),
+			str:          "1234-05-06",
+			timeTruncate: time.Second,
+		},
+		{
+			t:            time.Date(4567, 12, 31, 12, 0, 0, 0, time.UTC),
+			str:          "4567-12-31 12:00:00",
+			timeTruncate: time.Minute,
+		},
+		{
+			t:            time.Date(2020, 5, 30, 12, 34, 0, 0, time.UTC),
+			str:          "2020-05-30 12:34:00",
+			timeTruncate: 0,
+		},
+		{
+			t:            time.Date(2020, 5, 30, 12, 34, 56, 0, time.UTC),
+			str:          "2020-05-30 12:34:56",
+			timeTruncate: time.Second,
+		},
+		{
+			t:            time.Date(2020, 5, 30, 22, 33, 44, 123000000, time.UTC),
+			str:          "2020-05-30 22:33:44",
+			timeTruncate: time.Second,
+		},
+		{
+			t:            time.Date(2020, 5, 30, 22, 33, 44, 123456000, time.UTC),
+			str:          "2020-05-30 22:33:44.123",
+			timeTruncate: time.Millisecond,
+		},
+		{
+			t:            time.Date(2020, 5, 30, 22, 33, 44, 123456789, time.UTC),
+			str:          "2020-05-30 22:33:44",
+			timeTruncate: time.Second,
+		},
+		{
+			t:            time.Date(9999, 12, 31, 23, 59, 59, 999999999, time.UTC),
+			str:          "9999-12-31 23:59:59.999999999",
+			timeTruncate: 0,
+		},
+		{
+			t:            time.Date(1, 1, 1, 1, 1, 1, 1, time.UTC),
+			str:          "0001-01-01",
+			timeTruncate: 365 * 24 * time.Hour,
+		},
 		// year out of range
 		{
-		v := time.Date(0, 1, 1, 0, 0, 0, 0, time.UTC)
+			t:           time.Date(0, 1, 1, 0, 0, 0, 0, time.UTC),
+			expectedErr: true,
+		},
+		{
+			t:           time.Date(10000, 1, 1, 0, 0, 0, 0, time.UTC),
+			expectedErr: true,
+		},
+	}
+	for _, v := range tests {
 		buf := make([]byte, 0, 32)
-		_, err := appendDateTime(buf, v)
-		if err == nil {
-			t.Error("want an error")
-			return
+		buf, err := appendDateTime(buf, v.t, v.timeTruncate)
+		if err != nil {
+			if !v.expectedErr {
+				t.Errorf("appendDateTime(%v) returned an errror: %v", v.t, err)
 			}
+			continue
 		}
-	{
-		v := time.Date(10000, 1, 1, 0, 0, 0, 0, time.UTC)
-		buf := make([]byte, 0, 32)
-		_, err := appendDateTime(buf, v)
-		if err == nil {
-			t.Error("want an error")
-			return
+		if str := string(buf); str != v.str {
+			t.Errorf("appendDateTime(%v), have: %s, want: %s", v.t, str, v.str)
 		}
 	}
 }

manager.go

@@ -35,7 +35,7 @@ type Manager struct {
 	//logger       Logger
 
 	database  *gorm.DB
-	dbSaver  *DBSaver
+	jobSyncer *JobSyncer
 
 	mu sync.Mutex
 }
@@ -152,9 +152,9 @@ func (m *Manager) DeleteJob(id JobID) error {
 		return err
 	}
 
-	if m.dbSaver != nil {
+	if m.jobSyncer != nil {
 
-		err := m.dbSaver.DeleteJob(job)
+		err := m.jobSyncer.DeleteJob(job)
 		if err != nil {
 			return err
 		}
@@ -186,9 +186,9 @@ func (m *Manager) ResetJobLogs(id JobID) error {
 		return ErrJobNotActive
 	}
 
-	if m.dbSaver != nil {
+	if m.jobSyncer != nil {
 
-		err := m.dbSaver.ResetLogs(m.activeJobs[id])
+		err := m.jobSyncer.ResetLogs(m.activeJobs[id])
 		if err != nil {
 			return err
 		}
@@ -206,9 +206,9 @@ func (m *Manager) ResetJobStats(id JobID) error {
 		return ErrJobNotActive
 	}
 
-	if m.dbSaver != nil {
+	if m.jobSyncer != nil {
 
-		err := m.dbSaver.ResetStats(m.activeJobs[id])
+		err := m.jobSyncer.ResetStats(m.activeJobs[id])
 		if err != nil {
 			return err
 		}
@@ -290,12 +290,10 @@ func (m *Manager) SetDB(db *gorm.DB) *Manager {
 	defer m.mu.Unlock()
 	m.database = db
 
-	if m.dbSaver != nil {
+	if m.jobSyncer != nil {
 		return m
 	}
-
-	m.dbSaver = NewDBSaver()
-	m.dbSaver.SetManager(m)
+	m.jobSyncer = NewJobSyncer(m)
 
 	return m
 }
@@ -391,7 +389,6 @@ func (m *Manager) RemoveWorker(worker Worker) error {
 
 // Start starts the manager
 func (m *Manager) Start() error {
-	var err error
 
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -400,21 +397,31 @@ func (m *Manager) Start() error {
 		return ErrManagerAlreadyRunning
 	}
 
-	if m.dbSaver != nil {
-		p := StartDBSaver(m.dbSaver)
+	if m.jobSyncer != nil {
+		p := CreateAndStartJobSyncer(m)
 
 		ready := make(chan struct{})
 
-		Then[bool, bool](p, func(value bool) (bool, error) {
+		var jobSyncerErr error
+
+		Then[*JobSyncer, *JobSyncer](p, func(value *JobSyncer) (*JobSyncer, error) {
 			close(ready)
+			m.mu.Lock()
+			m.jobSyncer = value
+			m.mu.Unlock()
 			return value, nil
 		}, func(e error) error {
 			close(ready)
-			Error("Error while starting db saver", "error", err)
+			Error("Error while starting db saver", "error", e)
+			jobSyncerErr = e
 			return nil
 		})
 
 		<-ready
+
+		if jobSyncerErr != nil {
+			return jobSyncerErr
+		}
 	}
 
 	if len(m.workerMap) == 0 {
@@ -449,7 +456,7 @@ func (m *Manager) Start() error {
 
 	go m.handleJobEvents()
 
-	err = m.checkAndSetRunningState()
+	err := m.checkAndSetRunningState()
 
 	if err != nil {
 		wrappedErr = fmt.Errorf("%w\n%s", wrappedErr, err.Error())
@@ -481,7 +488,7 @@ func (m *Manager) Stop() error {
 
 	for _, worker := range m.workerMap {
 		err := worker.Stop()
-		if err != nil && err != ErrWorkerAlreadyStopped {
+		if err != nil && !errors.Is(err, ErrWorkerAlreadyStopped) {
 			if wrappedErr == nil {
 				wrappedErr = fmt.Errorf("Error: ")
 			}
@@ -500,8 +507,16 @@ func (m *Manager) Stop() error {
 		m.cronInstance.Stop()
 	}
 
-	if m.dbSaver != nil {
-		m.dbSaver.Stop()
+	if m.jobSyncer != nil {
+		err = m.jobSyncer.Stop()
+		if err != nil {
+			if wrappedErr == nil {
+				wrappedErr = fmt.Errorf("Error: ")
+			}
+
+			wrappedErr = fmt.Errorf("%w\n%s", wrappedErr, err.Error())
+
+		}
 	}
 
 	return wrappedErr
@@ -564,11 +579,8 @@ func (m *Manager) ScheduleJob(job GenericJob, scheduler Scheduler) error {
 
 	m.activeJobs[job.GetID()] = job
 
-	if m.dbSaver != nil {
-		err := m.dbSaver.SaveJob(job)
-		if err != nil {
-			return err
-		}
+	if m.jobSyncer != nil {
+		m.jobSyncer.AddJob(job)
 	}
 
 	return nil
@@ -621,7 +633,7 @@ func (m *Manager) handleJobEvents() {
 
 				job := event.Data.(GenericJob)
 				err := m.queue.Enqueue(job)
-				if err != nil && err != ErrJobAlreadyExists {
+				if err != nil && !errors.Is(err, ErrJobAlreadyExists) {
 					Error("Error while queueing job", "error", err)
 
 				}

manager_test.go

@@ -64,6 +64,10 @@ func (m *MockGenericJob) ResetStats() {
 
 }
 
+func (m *MockGenericJob) GetStats() JobStats {
+	return JobStats{}
+}
+
 func (m *MockGenericJob) GetMaxRetries() uint {
 	return 0
 }
@@ -232,7 +236,7 @@ func TestManager_CancelJob(t *testing.T) {
 
 func TestManagerEventHandling(t *testing.T) {
 	mgr := NewManager()
-	worker := NewLocalWorker(1)
+	worker := NewLocalWorker(10)
 	err := mgr.AddWorker(worker)
 	assert.Nil(t, err)
 

runnable-counter.go

@@ -68,6 +68,8 @@ func (c *CounterRunnable) GetType() string {
 }
 
 func (c *CounterRunnable) GetPersistence() RunnableImport {
+	c.mu.Lock()
+	defer c.mu.Unlock()
 
 	data := JSONMap{
 		"count": c.Count,

runnable-dummy.go

@@ -3,7 +3,10 @@
 
 package jobqueue
 
-import "context"
+import (
+	"context"
+	"sync"
+)
 
 func NewDummyRunnableFromMap(data map[string]any) (*DummyRunnable, error) {
 	return &DummyRunnable{}, nil
@@ -13,7 +16,9 @@ func NewDummyRunnableFromMap(data map[string]any) (*DummyRunnable, error) {
 type DummyResult struct {
 }
 
-type DummyRunnable struct{}
+type DummyRunnable struct {
+	mu sync.Mutex
+}
 
 func (d *DummyRunnable) Run(_ context.Context) (RunResult[DummyResult], error) {
 	return RunResult[DummyResult]{
@@ -26,6 +31,8 @@ func (d *DummyRunnable) GetType() string {
 }
 
 func (c *DummyRunnable) GetPersistence() RunnableImport {
+	c.mu.Lock()
+	defer c.mu.Unlock()
 
 	data := JSONMap{}
 

runnable-fileoperation.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"sync"
 )
 
 func NewFileOperationRunnableFromMap(data map[string]interface{}) (*FileOperationRunnable, error) {
@@ -60,6 +61,7 @@ type FileOperationRunnable struct {
 	Operation string // z.B. "read", "write", "delete"
 	FilePath  string
 	Content   string // Optional, je nach Operation
+	mu        sync.Mutex
 }
 
 func (f *FileOperationRunnable) Run(_ context.Context) (RunResult[FileOperationResult], error) {
@@ -142,6 +144,8 @@ func (f *FileOperationRunnable) GetType() string {
 }
 
 func (f *FileOperationRunnable) GetPersistence() RunnableImport {
+	f.mu.Lock()
+	defer f.mu.Unlock()
 
 	data := JSONMap{
 		"operation": f.Operation,