rancher/steve
1
0
Fork 0
mirror of https://github.com/rancher/steve.git synced 2025-09-08 18:59:58 +00:00
Code Issues Projects Releases Wiki Activity

sql: use a closure to wrap transactions (#469)

Browse Source 
This introduces the a `WithTransaction` function, which is then used for all transactional work in Steve.

Because `WithTransaction` takes care of all `Begin`s, `Commit`s and `Rollback`s, it eliminates the problem where forgotten open transactions can block all other operations (with long stalling and `SQLITE_BUSY` errors).

This also:

- merges together the disparate `DBClient` interfaces in one only `db.Client` interface with one unexported non-test implementation. I found this much easier to follow
- refactors the transaction package in order to make it as minimal as possible, and as close to the wrapped `sql.Tx` and `sql.Stmt` functions as possible, in order to reduce cognitive load when working with this part of the codebase
- simplifies tests accordingly
- adds a couple of known files to `.gitignore`
    
Credits to @tomleb for suggesting the approach: https://github.com/rancher/lasso/pull/121#pullrequestreview-2515872507
This commit is contained in:
Silvio Moioli
2025-02-05 10:05:52 +01:00
committed by GitHub
parent 6a46a1e091
commit 772dc7577e
28 changed files with 1543 additions and 2059 deletions
Download Patch File Download Diff File Expand all files Collapse all files

159
pkg/sqlcache/store/store.go
View File

@@ -34,7 +34,7 @@ const (
// Store is a SQLite-backed cache.Store
type Store struct {
DBClient
db.Client
name string
typ reflect.Type
@@ -53,49 +53,37 @@ type Store struct {
listStmt *sql.Stmt
listKeysStmt *sql.Stmt
afterUpsert []func(key string, obj any, tx db.TXClient) error
afterDelete []func(key string, tx db.TXClient) error
afterUpsert []func(key string, obj any, tx transaction.Client) error
afterDelete []func(key string, tx transaction.Client) error
}
// Test that Store implements cache.Indexer
var _ cache.Store = (*Store)(nil)
type DBClient interface {
BeginTx(ctx context.Context, forWriting bool) (db.TXClient, error)
Prepare(stmt string) *sql.Stmt
QueryForRows(ctx context.Context, stmt transaction.Stmt, params ...any) (*sql.Rows, error)
ReadObjects(rows db.Rows, typ reflect.Type, shouldDecrypt bool) ([]any, error)
ReadStrings(rows db.Rows) ([]string, error)
ReadInt(rows db.Rows) (int, error)
Upsert(tx db.TXClient, stmt *sql.Stmt, key string, obj any, shouldEncrypt bool) error
CloseStmt(closable db.Closable) error
}
// NewStore creates a SQLite-backed cache.Store for objects of the given example type
func NewStore(example any, keyFunc cache.KeyFunc, c DBClient, shouldEncrypt bool, name string) (*Store, error) {
func NewStore(example any, keyFunc cache.KeyFunc, c db.Client, shouldEncrypt bool, name string) (*Store, error) {
s := &Store{
name: name,
typ: reflect.TypeOf(example),
DBClient: c,
Client: c,
keyFunc: keyFunc,
shouldEncrypt: shouldEncrypt,
afterUpsert: []func(key string, obj any, tx db.TXClient) error{},
afterDelete: []func(key string, tx db.TXClient) error{},
afterUpsert: []func(key string, obj any, tx transaction.Client) error{},
afterDelete: []func(key string, tx transaction.Client) error{},
}
dbName := db.Sanitize(s.name)
// once multiple informerfactories are needed, this can accept the case where table already exists error is received
txC, err := s.BeginTx(context.Background(), true)
if err != nil {
return nil, err
}
dbName := db.Sanitize(s.name)
createTableQuery := fmt.Sprintf(createTableFmt, dbName)
err = txC.Exec(createTableQuery)
if err != nil {
return nil, &db.QueryError{QueryString: createTableQuery, Err: err}
}
err := s.WithTransaction(context.Background(), true, func(tx transaction.Client) error {
createTableQuery := fmt.Sprintf(createTableFmt, dbName)
_, err := tx.Exec(createTableQuery)
if err != nil {
return &db.QueryError{QueryString: createTableQuery, Err: err}
}
err = txC.Commit()
return nil
})
if err != nil {
return nil, err
}
@@ -118,42 +106,36 @@ func NewStore(example any, keyFunc cache.KeyFunc, c DBClient, shouldEncrypt bool
/* Core methods */
// upsert saves an obj with its key, or updates key with obj if it exists in this Store
func (s *Store) upsert(key string, obj any) error {
tx, err := s.BeginTx(context.Background(), true)
if err != nil {
return err
}
return s.WithTransaction(context.Background(), true, func(tx transaction.Client) error {
err := s.Upsert(tx, s.upsertStmt, key, obj, s.shouldEncrypt)
if err != nil {
return &db.QueryError{QueryString: s.upsertQuery, Err: err}
}
err = s.Upsert(tx, s.upsertStmt, key, obj, s.shouldEncrypt)
if err != nil {
return &db.QueryError{QueryString: s.upsertQuery, Err: err}
}
err = s.runAfterUpsert(key, obj, tx)
if err != nil {
return err
}
err = s.runAfterUpsert(key, obj, tx)
if err != nil {
return err
}
return tx.Commit()
return nil
})
}
// deleteByKey deletes the object associated with key, if it exists in this Store
func (s *Store) deleteByKey(key string) error {
tx, err := s.BeginTx(context.Background(), true)
if err != nil {
return err
}
return s.WithTransaction(context.Background(), true, func(tx transaction.Client) error {
_, err := tx.Stmt(s.deleteStmt).Exec(key)
if err != nil {
return &db.QueryError{QueryString: s.deleteQuery, Err: err}
}
err = tx.StmtExec(tx.Stmt(s.deleteStmt), key)
if err != nil {
return &db.QueryError{QueryString: s.deleteQuery, Err: err}
}
err = s.runAfterDelete(key, tx)
if err != nil {
return err
}
err = s.runAfterDelete(key, tx)
if err != nil {
return err
}
return tx.Commit()
return nil
})
}
// GetByKey returns the object associated with the given object's key
@@ -267,45 +249,42 @@ func (s *Store) Replace(objects []any, _ string) error {
// replaceByKey will delete the contents of the Store, using instead the given key to obj map
func (s *Store) replaceByKey(objects map[string]any) error {
txC, err := s.BeginTx(context.Background(), true)
if err != nil {
return err
}
return s.WithTransaction(context.Background(), true, func(txC transaction.Client) error {
txCListKeys := txC.Stmt(s.listKeysStmt)
txCListKeys := txC.Stmt(s.listKeysStmt)
rows, err := s.QueryForRows(context.TODO(), txCListKeys)
if err != nil {
return err
}
keys, err := s.ReadStrings(rows)
if err != nil {
return err
}
for _, key := range keys {
err = txC.StmtExec(txC.Stmt(s.deleteStmt), key)
rows, err := s.QueryForRows(context.TODO(), txCListKeys)
if err != nil {
return err
}
err = s.runAfterDelete(key, txC)
keys, err := s.ReadStrings(rows)
if err != nil {
return err
}
}
for key, obj := range objects {
err = s.Upsert(txC, s.upsertStmt, key, obj, s.shouldEncrypt)
if err != nil {
return err
for _, key := range keys {
_, err = txC.Stmt(s.deleteStmt).Exec(key)
if err != nil {
return err
}
err = s.runAfterDelete(key, txC)
if err != nil {
return err
}
}
err = s.runAfterUpsert(key, obj, txC)
if err != nil {
return err
}
}
return txC.Commit()
for key, obj := range objects {
err = s.Upsert(txC, s.upsertStmt, key, obj, s.shouldEncrypt)
if err != nil {
return err
}
err = s.runAfterUpsert(key, obj, txC)
if err != nil {
return err
}
}
return nil
})
}
// Resync is a no-op and is deprecated
@@ -316,7 +295,7 @@ func (s *Store) Resync() error {
/* Utilities */
// RegisterAfterUpsert registers a func to be called after each upsert
func (s *Store) RegisterAfterUpsert(f func(key string, obj any, txC db.TXClient) error) {
func (s *Store) RegisterAfterUpsert(f func(key string, obj any, txC transaction.Client) error) {
s.afterUpsert = append(s.afterUpsert, f)
}
@@ -334,7 +313,7 @@ func (s *Store) GetType() reflect.Type {
// keep
// runAfterUpsert executes functions registered to run after upsert
func (s *Store) runAfterUpsert(key string, obj any, txC db.TXClient) error {
func (s *Store) runAfterUpsert(key string, obj any, txC transaction.Client) error {
for _, f := range s.afterUpsert {
err := f(key, obj, txC)
if err != nil {
@@ -345,13 +324,13 @@ func (s *Store) runAfterUpsert(key string, obj any, txC db.TXClient) error {
}
// RegisterAfterDelete registers a func to be called after each deletion
func (s *Store) RegisterAfterDelete(f func(key string, txC db.TXClient) error) {
func (s *Store) RegisterAfterDelete(f func(key string, txC transaction.Client) error) {
s.afterDelete = append(s.afterDelete, f)
}
// keep
// runAfterDelete executes functions registered to run after upsert
func (s *Store) runAfterDelete(key string, txC db.TXClient) error {
func (s *Store) runAfterDelete(key string, txC transaction.Client) error {
for _, f := range s.afterDelete {
err := f(key, txC)
if err != nil {