Skip to content

[sql-43] Add support for sqldb/v2 in litd #1085

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 12 commits into
base: master
Choose a base branch
from
Draft

[sql-43] Add support for sqldb/v2 in litd #1085

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
ea20fb5
mod: add sqldb/v2 dependency to `litd`
ViktorT-11 Jun 4, 2025
47ab4f8
db: add `LitdMigrationStream`
ViktorT-11 Jul 24, 2025
4bffead
db+sqlc: use sqldb/v2 `BackendType` definition
ViktorT-11 Jul 24, 2025
c7e3b9d
sqlc: introduce `NewForType` helper method
ViktorT-11 Jul 24, 2025
d8bbce8
db: add sqldb/v2 `PostgresStore` creation helper
ViktorT-11 Jul 24, 2025
b00d77d
accounts: use `sqldb/v2` in accounts package
ViktorT-11 Jul 22, 2025
d869a91
multi: use `sqldb/v2` in session package
ViktorT-11 Jul 22, 2025
431c3d6
firewalldb: add `ListAllKVStoresRecords` to queries
ViktorT-11 Jul 10, 2025
8d6b9d0
firewalldb: rename sqlStore to store in mig test
ViktorT-11 Jul 11, 2025
b5d4438
firewalldb: use `sqldb/v2` in firewalldb package
ViktorT-11 Jul 22, 2025
133ac68
multi: remove unused db code
ViktorT-11 Jul 23, 2025
7792848
mutli: rename `db.NewTestPostgresV2DB` function
ViktorT-11 Jul 23, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
306 changes: 0 additions & 306 deletions db/interfaces.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,311 +1,5 @@
package db

import (
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤩

"context"
"database/sql"
"math"
prand "math/rand"
"time"

"github.com/lightninglabs/lightning-terminal/db/sqlc"
"github.com/lightningnetwork/lnd/sqldb/v2"
)

var (
// DefaultStoreTimeout is the default timeout used for any interaction
// with the storage/database.
DefaultStoreTimeout = time.Second * 10
)

const (
// DefaultNumTxRetries is the default number of times we'll retry a
// transaction if it fails with an error that permits transaction
// repetition.
DefaultNumTxRetries = 10

// DefaultInitialRetryDelay is the default initial delay between
// retries. This will be used to generate a random delay between -50%
// and +50% of this value, so 20 to 60 milliseconds. The retry will be
// doubled after each attempt until we reach DefaultMaxRetryDelay. We
// start with a random value to avoid multiple goroutines that are
// created at the same time to effectively retry at the same time.
DefaultInitialRetryDelay = time.Millisecond * 40

// DefaultMaxRetryDelay is the default maximum delay between retries.
DefaultMaxRetryDelay = time.Second * 3
)

// TxOptions represents a set of options one can use to control what type of
// database transaction is created. Transaction can wither be read or write.
type TxOptions interface {
// ReadOnly returns true if the transaction should be read only.
ReadOnly() bool
}

// BatchedTx is a generic interface that represents the ability to execute
// several operations to a given storage interface in a single atomic
// transaction. Typically, Q here will be some subset of the main sqlc.Querier
// interface allowing it to only depend on the routines it needs to implement
// any additional business logic.
type BatchedTx[Q any] interface {
// ExecTx will execute the passed txBody, operating upon generic
// parameter Q (usually a storage interface) in a single transaction.
// The set of TxOptions are passed in in order to allow the caller to
// specify if a transaction should be read-only and optionally what
// type of concurrency control should be used.
ExecTx(ctx context.Context, txOptions TxOptions,
txBody func(Q) error) error

// Backend returns the type of the database backend used.
Backend() sqldb.BackendType
}

// Tx represents a database transaction that can be committed or rolled back.
type Tx interface {
// Commit commits the database transaction, an error should be returned
// if the commit isn't possible.
Commit() error

// Rollback rolls back an incomplete database transaction.
// Transactions that were able to be committed can still call this as a
// noop.
Rollback() error
}

// QueryCreator is a generic function that's used to create a Querier, which is
// a type of interface that implements storage related methods from a database
// transaction. This will be used to instantiate an object callers can use to
// apply multiple modifications to an object interface in a single atomic
// transaction.
type QueryCreator[Q any] func(*sql.Tx) Q

// BatchedQuerier is a generic interface that allows callers to create a new
// database transaction based on an abstract type that implements the TxOptions
// interface.
type BatchedQuerier interface {
// Querier is the underlying query source, this is in place so we can
// pass a BatchedQuerier implementation directly into objects that
// create a batched version of the normal methods they need.
sqlc.Querier

// CustomQueries is the set of custom queries that we have manually
// defined in addition to the ones generated by sqlc.
sqlc.CustomQueries

// BeginTx creates a new database transaction given the set of
// transaction options.
BeginTx(ctx context.Context, options TxOptions) (*sql.Tx, error)
}

// txExecutorOptions is a struct that holds the options for the transaction
// executor. This can be used to do things like retry a transaction due to an
// error a certain amount of times.
type txExecutorOptions struct {
numRetries int
initialRetryDelay time.Duration
maxRetryDelay time.Duration
}

// defaultTxExecutorOptions returns the default options for the transaction
// executor.
func defaultTxExecutorOptions() *txExecutorOptions {
return &txExecutorOptions{
numRetries: DefaultNumTxRetries,
initialRetryDelay: DefaultInitialRetryDelay,
maxRetryDelay: DefaultMaxRetryDelay,
}
}

// randRetryDelay returns a random retry delay between -50% and +50%
// of the configured delay that is doubled for each attempt and capped at a max
// value.
func (t *txExecutorOptions) randRetryDelay(attempt int) time.Duration {
halfDelay := t.initialRetryDelay / 2
randDelay := prand.Int63n(int64(t.initialRetryDelay)) //nolint:gosec

// 50% plus 0%-100% gives us the range of 50%-150%.
initialDelay := halfDelay + time.Duration(randDelay)

// If this is the first attempt, we just return the initial delay.
if attempt == 0 {
return initialDelay
}

// For each subsequent delay, we double the initial delay. This still
// gives us a somewhat random delay, but it still increases with each
// attempt. If we double something n times, that's the same as
// multiplying the value with 2^n. We limit the power to 32 to avoid
// overflows.
factor := time.Duration(math.Pow(2, math.Min(float64(attempt), 32)))
actualDelay := initialDelay * factor

// Cap the delay at the maximum configured value.
if actualDelay > t.maxRetryDelay {
return t.maxRetryDelay
}

return actualDelay
}

// TxExecutorOption is a functional option that allows us to pass in optional
// argument when creating the executor.
type TxExecutorOption func(*txExecutorOptions)

// WithTxRetries is a functional option that allows us to specify the number of
// times a transaction should be retried if it fails with a repeatable error.
func WithTxRetries(numRetries int) TxExecutorOption {
return func(o *txExecutorOptions) {
o.numRetries = numRetries
}
}

// WithTxRetryDelay is a functional option that allows us to specify the delay
// to wait before a transaction is retried.
func WithTxRetryDelay(delay time.Duration) TxExecutorOption {
return func(o *txExecutorOptions) {
o.initialRetryDelay = delay
}
}

// TransactionExecutor is a generic struct that abstracts away from the type of
// query a type needs to run under a database transaction, and also the set of
// options for that transaction. The QueryCreator is used to create a query
// given a database transaction created by the BatchedQuerier.
type TransactionExecutor[Query any] struct {
BatchedQuerier

createQuery QueryCreator[Query]

opts *txExecutorOptions
}

// NewTransactionExecutor creates a new instance of a TransactionExecutor given
// a Querier query object and a concrete type for the type of transactions the
// Querier understands.
func NewTransactionExecutor[Querier any](db BatchedQuerier,
createQuery QueryCreator[Querier],
opts ...TxExecutorOption) *TransactionExecutor[Querier] {

txOpts := defaultTxExecutorOptions()
for _, optFunc := range opts {
optFunc(txOpts)
}

return &TransactionExecutor[Querier]{
BatchedQuerier: db,
createQuery: createQuery,
opts: txOpts,
}
}

// ExecTx is a wrapper for txBody to abstract the creation and commit of a db
// transaction. The db transaction is embedded in a `*Queries` that txBody
// needs to use when executing each one of the queries that need to be applied
// atomically. This can be used by other storage interfaces to parameterize the
// type of query and options run, in order to have access to batched operations
// related to a storage object.
func (t *TransactionExecutor[Q]) ExecTx(ctx context.Context,
txOptions TxOptions, txBody func(Q) error) error {

waitBeforeRetry := func(attemptNumber int) {
retryDelay := t.opts.randRetryDelay(attemptNumber)

log.Tracef("Retrying transaction due to tx serialization or "+
"deadlock error, attempt_number=%v, delay=%v",
attemptNumber, retryDelay)

// Before we try again, we'll wait with a random backoff based
// on the retry delay.
time.Sleep(retryDelay)
}

for i := 0; i < t.opts.numRetries; i++ {
// Create the db transaction.
tx, err := t.BatchedQuerier.BeginTx(ctx, txOptions)
if err != nil {
dbErr := MapSQLError(err)
if IsSerializationOrDeadlockError(dbErr) {
// Nothing to roll back here, since we didn't
// even get a transaction yet.
waitBeforeRetry(i)
continue
}

return dbErr
}

// Rollback is safe to call even if the tx is already closed,
// so if the tx commits successfully, this is a no-op.
defer func() {
_ = tx.Rollback()
}()

if err := txBody(t.createQuery(tx)); err != nil {
dbErr := MapSQLError(err)
if IsSerializationOrDeadlockError(dbErr) {
// Roll back the transaction, then pop back up
// to try once again.
_ = tx.Rollback()

waitBeforeRetry(i)
continue
}

return dbErr
}

// Commit transaction.
if err = tx.Commit(); err != nil {
dbErr := MapSQLError(err)
if IsSerializationOrDeadlockError(dbErr) {
// Roll back the transaction, then pop back up
// to try once again.
_ = tx.Rollback()

waitBeforeRetry(i)
continue
}

return dbErr
}

return nil
}

// If we get to this point, then we weren't able to successfully commit
// a tx given the max number of retries.
return ErrRetriesExceeded
}

// Backend returns the type of the database backend used.
func (t *TransactionExecutor[Q]) Backend() sqldb.BackendType {
return t.BatchedQuerier.Backend()
}

// BaseDB is the base database struct that each implementation can embed to
// gain some common functionality.
type BaseDB struct {
*sql.DB

*sqlc.Queries
}

// BeginTx wraps the normal sql specific BeginTx method with the TxOptions
// interface. This interface is then mapped to the concrete sql tx options
// struct.
func (s *BaseDB) BeginTx(ctx context.Context, opts TxOptions) (*sql.Tx, error) {
sqlOptions := sql.TxOptions{
ReadOnly: opts.ReadOnly(),
Isolation: sql.LevelSerializable,
}
return s.DB.BeginTx(ctx, &sqlOptions)
}

// Backend returns the type of the database backend used.
func (s *BaseDB) Backend() sqldb.BackendType {
return s.Queries.Backend()
}

// QueriesTxOptions defines the set of db txn options the SQLQueries
// understands.
type QueriesTxOptions struct {
Expand Down
Loading