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

317 lines
8.5 KiB
Go

// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
// See LICENSE.txt for license information.
package sqlstore
import (
"database/sql"
"fmt"
sq "github.com/mattermost/squirrel"
"github.com/pkg/errors"
"github.com/mattermost/mattermost/server/public/model"
"github.com/mattermost/mattermost/server/v8/channels/store"
)
const (
defaultPluginKeyFetchLimit = 10
)
type SqlPluginStore struct {
*SqlStore
}
func newSqlPluginStore(sqlStore *SqlStore) store.PluginStore {
return &SqlPluginStore{sqlStore}
}
func (ps SqlPluginStore) SaveOrUpdate(kv *model.PluginKeyValue) (*model.PluginKeyValue, error) {
if err := kv.IsValid(); err != nil {
return nil, err
}
if kv.Value == nil {
// Setting a key to nil is the same as removing it
err := ps.Delete(kv.PluginId, kv.Key)
if err != nil {
return nil, err
}
return kv, nil
}
query := ps.getQueryBuilder().
Insert("PluginKeyValueStore").
Columns("PluginId", "PKey", "PValue", "ExpireAt").
Values(kv.PluginId, kv.Key, kv.Value, kv.ExpireAt)
query = query.SuffixExpr(sq.Expr("ON CONFLICT (pluginid, pkey) DO UPDATE SET PValue = ?, ExpireAt = ?", kv.Value, kv.ExpireAt))
queryString, args, err := query.ToSql()
if err != nil {
return nil, errors.Wrap(err, "plugin_tosql")
}
if _, err := ps.GetMaster().Exec(queryString, args...); err != nil {
return nil, errors.Wrap(err, "failed to upsert PluginKeyValue")
}
return kv, nil
}
func (ps SqlPluginStore) CompareAndSet(kv *model.PluginKeyValue, oldValue []byte) (bool, error) {
if err := kv.IsValid(); err != nil {
return false, err
}
if kv.Value == nil {
// Setting a key to nil is the same as removing it
return ps.CompareAndDelete(kv, oldValue)
}
if oldValue == nil {
// Delete any existing, expired value.
query := ps.getQueryBuilder().
Delete("PluginKeyValueStore").
Where(sq.Eq{"PluginId": kv.PluginId}).
Where(sq.Eq{"PKey": kv.Key}).
Where(sq.NotEq{"ExpireAt": int(0)}).
Where(sq.Lt{"ExpireAt": model.GetMillis()})
queryString, args, err := query.ToSql()
if err != nil {
return false, errors.Wrap(err, "plugin_tosql")
}
if _, err = ps.GetMaster().Exec(queryString, args...); err != nil {
return false, errors.Wrap(err, "failed to delete PluginKeyValue")
}
// Insert if oldValue is nil
queryString, args, err = ps.getQueryBuilder().
Insert("PluginKeyValueStore").
Columns("PluginId", "PKey", "PValue", "ExpireAt").
Values(kv.PluginId, kv.Key, kv.Value, kv.ExpireAt).ToSql()
if err != nil {
return false, errors.Wrap(err, "plugin_tosql")
}
if _, err := ps.GetMaster().Exec(queryString, args...); err != nil {
// If the error is from unique constraints violation, it's the result of a
// race condition, return false and no error. Otherwise we have a real error and
// need to return it.
if IsUniqueConstraintError(err, []string{"PRIMARY", "PluginId", "Key", "PKey", "pkey"}) {
return false, nil
}
return false, errors.Wrap(err, "failed to insert PluginKeyValue")
}
} else {
currentTime := model.GetMillis()
// Update if oldValue is not nil
query := ps.getQueryBuilder().
Update("PluginKeyValueStore").
Set("PValue", kv.Value).
Set("ExpireAt", kv.ExpireAt).
Where(sq.Eq{"PluginId": kv.PluginId}).
Where(sq.Eq{"PKey": kv.Key}).
Where(sq.Eq{"PValue": oldValue}).
Where(sq.Or{
sq.Eq{"ExpireAt": int(0)},
sq.Gt{"ExpireAt": currentTime},
})
queryString, args, err := query.ToSql()
if err != nil {
return false, errors.Wrap(err, "plugin_tosql")
}
updateResult, err := ps.GetMaster().Exec(queryString, args...)
if err != nil {
return false, errors.Wrap(err, "failed to update PluginKeyValue")
}
if rowsAffected, err := updateResult.RowsAffected(); err != nil {
// Failed to update
return false, errors.Wrap(err, "unable to get rows affected")
} else if rowsAffected == 0 {
// No rows were affected by the update, where condition was not satisfied,
// return false, but no error.
return false, nil
}
}
return true, nil
}
func (ps SqlPluginStore) CompareAndDelete(kv *model.PluginKeyValue, oldValue []byte) (bool, error) {
if err := kv.IsValid(); err != nil {
return false, err
}
if oldValue == nil {
// nil can't be stored. Return showing that we didn't do anything
return false, nil
}
query := ps.getQueryBuilder().
Delete("PluginKeyValueStore").
Where(sq.Eq{"PluginId": kv.PluginId}).
Where(sq.Eq{"PKey": kv.Key}).
Where(sq.Eq{"PValue": oldValue}).
Where(sq.Or{
sq.Eq{"ExpireAt": int(0)},
sq.Gt{"ExpireAt": model.GetMillis()},
})
queryString, args, err := query.ToSql()
if err != nil {
return false, errors.Wrap(err, "plugin_tosql")
}
deleteResult, err := ps.GetMaster().Exec(queryString, args...)
if err != nil {
return false, errors.Wrap(err, "failed to delete PluginKeyValue")
}
if rowsAffected, err := deleteResult.RowsAffected(); err != nil {
return false, errors.Wrap(err, "unable to get rows affected")
} else if rowsAffected == 0 {
return false, nil
}
return true, nil
}
func (ps SqlPluginStore) SetWithOptions(pluginId string, key string, value []byte, opt model.PluginKVSetOptions) (bool, error) {
if err := opt.IsValid(); err != nil {
return false, err
}
kv, err := model.NewPluginKeyValueFromOptions(pluginId, key, value, opt)
if err != nil {
return false, err
}
if opt.Atomic {
return ps.CompareAndSet(kv, opt.OldValue)
}
savedKv, nErr := ps.SaveOrUpdate(kv)
if nErr != nil {
return false, nErr
}
return savedKv != nil, nil
}
func (ps SqlPluginStore) Get(pluginId, key string) (*model.PluginKeyValue, error) {
currentTime := model.GetMillis()
query := ps.getQueryBuilder().Select("PluginId, PKey, PValue, ExpireAt").
From("PluginKeyValueStore").
Where(sq.Eq{"PluginId": pluginId}).
Where(sq.Eq{"PKey": key}).
Where(sq.Or{sq.Eq{"ExpireAt": 0}, sq.Gt{"ExpireAt": currentTime}})
queryString, args, err := query.ToSql()
if err != nil {
return nil, errors.Wrap(err, "plugin_tosql")
}
row := ps.GetReplica().QueryRow(queryString, args...)
var kv model.PluginKeyValue
if err := row.Scan(&kv.PluginId, &kv.Key, &kv.Value, &kv.ExpireAt); err != nil {
if err == sql.ErrNoRows {
return nil, store.NewErrNotFound("PluginKeyValue", fmt.Sprintf("pluginId=%s, key=%s", pluginId, key))
}
return nil, errors.Wrapf(err, "failed to get PluginKeyValue with pluginId=%s and key=%s", pluginId, key)
}
return &kv, nil
}
func (ps SqlPluginStore) Delete(pluginId, key string) error {
query := ps.getQueryBuilder().
Delete("PluginKeyValueStore").
Where(sq.Eq{"PluginId": pluginId}).
Where(sq.Eq{"Pkey": key})
queryString, args, err := query.ToSql()
if err != nil {
return errors.Wrap(err, "plugin_tosql")
}
if _, err := ps.GetMaster().Exec(queryString, args...); err != nil {
return errors.Wrapf(err, "failed to delete PluginKeyValue with pluginId=%s and key=%s", pluginId, key)
}
return nil
}
func (ps SqlPluginStore) DeleteAllForPlugin(pluginId string) error {
query := ps.getQueryBuilder().
Delete("PluginKeyValueStore").
Where(sq.Eq{"PluginId": pluginId})
queryString, args, err := query.ToSql()
if err != nil {
return errors.Wrap(err, "plugin_tosql")
}
if _, err := ps.GetMaster().Exec(queryString, args...); err != nil {
return errors.Wrapf(err, "failed to get all PluginKeyValues with pluginId=%s ", pluginId)
}
return nil
}
func (ps SqlPluginStore) DeleteAllExpired() error {
currentTime := model.GetMillis()
query := ps.getQueryBuilder().
Delete("PluginKeyValueStore").
Where(sq.NotEq{"ExpireAt": 0}).
Where(sq.Lt{"ExpireAt": currentTime})
queryString, args, err := query.ToSql()
if err != nil {
return errors.Wrap(err, "plugin_tosql")
}
if _, err := ps.GetMaster().Exec(queryString, args...); err != nil {
return errors.Wrap(err, "failed to delete all expired PluginKeyValues")
}
return nil
}
func (ps SqlPluginStore) List(pluginId string, offset int, limit int) ([]string, error) {
if limit <= 0 {
limit = defaultPluginKeyFetchLimit
}
if offset <= 0 {
offset = 0
}
query := ps.getQueryBuilder().
Select("Pkey").
From("PluginKeyValueStore").
Where(sq.Eq{"PluginId": pluginId}).
Where(sq.Or{
sq.Eq{"ExpireAt": int(0)},
sq.Gt{"ExpireAt": model.GetMillis()},
}).
OrderBy("PKey").
Limit(uint64(limit)).
Offset(uint64(offset))
queryString, args, err := query.ToSql()
if err != nil {
return nil, errors.Wrap(err, "plugin_tosql")
}
keys := []string{}
err = ps.GetReplica().Select(&keys, queryString, args...)
if err != nil {
return nil, errors.Wrapf(err, "failed to get PluginKeyValues with pluginId=%s", pluginId)
}
return keys, nil
}