Files
916f6b7225 MM-68332: consistently enforce query timeouts (#36522)
* Remove QueryRowx (no-timeout) from sqlxDBWrapper; migrate caller to QueryRowX

QueryRowx forwarded to context.Background() with no timeout, while QueryRowX
(uppercase) already enforces the wrapper timeout. Removing the no-timeout
variant eliminates an accidental footgun and migrates the one sqlstore caller
(plugin_store) to the timeout-enforcing method.

* sqlxRow, with timeout cancel after Scan

Introduce sqlxRow, which pairs *sqlx.Row with its context cancel function.
QueryRowX (on both sqlxDBWrapper and sqlxTxWrapper) now returns *sqlxRow;
Scan calls cancel immediately after the row is consumed, releasing the
timeout context as soon as possible rather than waiting for the timer.

* sqlxRows, with timeout cancel on Close

Introduce sqlxRows, which embeds *sqlx.Rows and holds the timeout context's
cancel function. Close() cancels the context immediately after the rows are
done, releasing timeout resources as soon as iteration completes rather than
waiting for the timer to fire.

Introduce rowScanner interface (Next/Scan/Err) so the internal helpers
scanRowsIntoMap and scanRetentionIdsForDeletion accept any row iterator
rather than the concrete *sql.Rows, accommodating the new return types
without threading *sqlxRows through every caller.

* abolish the X suffix altogether

* delete unused NamedQuery

* add timeout tests for Query and QueryRow on db and tx wrappers

* fix tx/timeout tests: handle pq driver.ErrBadConn on killed connection

* fixup! fix tx/timeout tests: handle pq driver.ErrBadConn on killed connection

* override Next() on sqlxRows to cancel on EOF

* rm redundant row.Err(), handled by Scan

* apply timeout to context unless deadline set

* rebind consistently

* address review feedback: rename, trace, and QueryContext fix

- Rename withQueryTimeout → ensureQueryTimeout to better convey that it
  respects existing deadlines rather than overriding them.
- Add missing w.trace blocks to QueryRowContext and ExecContext.
- Change QueryContext to use ensureQueryTimeout and return *sqlxRows
  (instead of *sql.Rows) so the cancel is deferred to Close/Next
  rather than released prematurely.

* fix Beginx → Begin after master merge

* fix golangci-lint inline warning in sqlx_wrapper_test

---------

Co-authored-by: Mattermost Build <[email protected]>
2026-05-15 13:02:13 +00:00

224 lines
6.0 KiB
Go

// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
// See LICENSE.txt for license information.
package sqlstore
import (
"database/sql"
"errors"
"fmt"
"io"
"maps"
"net/url"
"strconv"
"strings"
"unicode"
"github.com/wiggin77/merror"
"github.com/mattermost/mattermost/server/public/model"
"github.com/mattermost/mattermost/server/public/shared/mlog"
)
// defaultMaxInsertParams is a conservative threshold (76% of PostgreSQL's
// 65,535 parameter limit) used to chunk bulk INSERT statements so they never
// overflow the wire-protocol's 16-bit parameter counter.
const defaultMaxInsertParams = 50_000
// chunkSlice splits items into sub-slices sized so that each chunk uses at most
// maxParams query parameters (columnsPerRow params per item). When the input
// already fits in one chunk the original slice is returned with zero allocation
// overhead.
func chunkSlice[T any](items []T, columnsPerRow int, maxParams int) [][]T {
if columnsPerRow <= 0 {
panic(fmt.Sprintf("chunkSlice: columnsPerRow must be > 0, got %d", columnsPerRow))
}
if maxParams <= 0 {
panic(fmt.Sprintf("chunkSlice: maxParams must be > 0, got %d", maxParams))
}
if columnsPerRow > maxParams {
panic(fmt.Sprintf("chunkSlice: columnsPerRow (%d) must be <= maxParams (%d)", columnsPerRow, maxParams))
}
if len(items) == 0 {
return nil
}
chunkSize := maxParams / columnsPerRow
if len(items) <= chunkSize {
return [][]T{items}
}
var chunks [][]T
for i := 0; i < len(items); i += chunkSize {
end := min(i+chunkSize, len(items))
chunks = append(chunks, items[i:end])
}
return chunks
}
var escapeLikeSearchChar = []string{
"%",
"_",
}
func sanitizeSearchTerm(term string, escapeChar string) string {
term = strings.Replace(term, escapeChar, "", -1)
for _, c := range escapeLikeSearchChar {
term = strings.Replace(term, c, escapeChar+c, -1)
}
return term
}
// Converts a list of strings into a list of query parameters and a named parameter map that can
// be used as part of a SQL query.
func MapStringsToQueryParams(list []string, paramPrefix string) (string, map[string]any) {
var keys strings.Builder
params := make(map[string]any, len(list))
for i, entry := range list {
if keys.Len() > 0 {
keys.WriteString(",")
}
key := paramPrefix + strconv.Itoa(i)
keys.WriteString(":" + key)
params[key] = entry
}
return "(" + keys.String() + ")", params
}
// finalizeTransactionX ensures a transaction is closed after use, rolling back if not already committed.
func finalizeTransactionX(transaction *sqlxTxWrapper, perr *error) {
// Rollback returns sql.ErrTxDone if the transaction was already closed.
if err := transaction.Rollback(); err != nil && !errors.Is(err, sql.ErrTxDone) {
*perr = merror.Append(*perr, err)
}
}
func deferClose(c io.Closer, perr *error) {
err := c.Close()
*perr = merror.Append(*perr, err)
}
// removeNonAlphaNumericUnquotedTerms removes all unquoted words that only contain
// non-alphanumeric chars from given line
func removeNonAlphaNumericUnquotedTerms(line, separator string) string {
words := strings.Split(line, separator)
filteredResult := make([]string, 0, len(words))
for _, w := range words {
if isQuotedWord(w) || containsAlphaNumericChar(w) {
filteredResult = append(filteredResult, strings.TrimSpace(w))
}
}
return strings.Join(filteredResult, separator)
}
// containsAlphaNumericChar returns true in case any letter or digit is present, false otherwise
func containsAlphaNumericChar(s string) bool {
for _, r := range s {
if unicode.IsLetter(r) || unicode.IsDigit(r) {
return true
}
}
return false
}
// isQuotedWord return true if the input string is quoted, false otherwise. Ex :-
//
// "quoted string" - will return true
// unquoted string - will return false
func isQuotedWord(s string) bool {
if len(s) < 2 {
return false
}
return s[0] == '"' && s[len(s)-1] == '"'
}
func constructArrayArgs(ids []string) (string, []any) {
var placeholder strings.Builder
values := make([]any, 0, len(ids))
for _, entry := range ids {
if placeholder.Len() > 0 {
placeholder.WriteString(",")
}
placeholder.WriteString("?")
values = append(values, entry)
}
return "(" + placeholder.String() + ")", values
}
func wrapBinaryParamStringMap(ok bool, props model.StringMap) model.StringMap {
if props == nil {
props = make(model.StringMap)
}
props[model.BinaryParamKey] = strconv.FormatBool(ok)
return props
}
// morphWriter is a target to pass to the logger instance of morph.
// For now, everything is just logged at a debug level. If we need to log
// errors/warnings from the library also, that needs to be seen later.
type morphWriter struct{}
func (l *morphWriter) Write(in []byte) (int, error) {
mlog.Debug(strings.TrimSpace(string(in)))
return len(in), nil
}
func DSNHasBinaryParam(dsn string) (bool, error) {
url, err := url.Parse(dsn)
if err != nil {
return false, err
}
return url.Query().Get("binary_parameters") == "yes", nil
}
// AppendBinaryFlag updates the byte slice to work using binary_parameters=yes.
func AppendBinaryFlag(buf []byte) []byte {
return append([]byte{0x01}, buf...)
}
const maxTokenSize = 50
// trimInput limits the string to a max size to prevent clogging up disk space
// while logging
func trimInput(input string) string {
if len(input) > maxTokenSize {
input = input[:maxTokenSize] + "..."
}
return input
}
// rowScanner is the minimal interface needed to iterate over SQL result rows.
type rowScanner interface {
Next() bool
Scan(dest ...any) error
Err() error
}
// scanRowsIntoMap scans SQL rows into a map, using a provided scanner function to extract key-value pairs
func scanRowsIntoMap[K comparable, V any](rows rowScanner, scanner func(rows rowScanner) (K, V, error), defaults map[K]V) (map[K]V, error) {
results := make(map[K]V, len(defaults))
// Initialize with default values if provided
maps.Copy(results, defaults)
for rows.Next() {
key, value, err := scanner(rows)
if err != nil {
return nil, err
}
results[key] = value
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("error while iterating rows: %w", err)
}
return results, nil
}