kvdb/postgres: remove global application level lock

channeldb/graph.go

@@ -1492,9 +1492,15 @@ func (c *ChannelGraph) pruneGraphNodes(nodes kvdb.RwBucket,
 		// If we reach this point, then there are no longer any edges
 		// that connect this node, so we can delete it.
 		if err := c.deleteLightningNode(nodes, nodePubKey[:]); err != nil {
-			log.Warnf("Unable to prune node %x from the "+
-				"graph: %v", nodePubKey, err)
-			continue
+			if errors.Is(err, ErrGraphNodeNotFound) ||
+				errors.Is(err, ErrGraphNodesNotFound) {
+
+				log.Warnf("Unable to prune node %x from the "+
+					"graph: %v", nodePubKey, err)
+				continue
+			}
+
+			return err
 		}
 
 		log.Infof("Pruned unconnected node %x from channel graph",

go.mod

@@ -212,4 +212,6 @@ replace google.golang.org/protobuf => github.com/lightninglabs/protobuf-go-hex-d
 // docs/INSTALL.md.
 go 1.19
 
+replace github.com/lightningnetwork/lnd/kvdb => ./kvdb
+
 retract v0.0.2

go.sum

@@ -448,8 +448,6 @@ github.com/lightningnetwork/lnd/clock v1.1.1 h1:OfR3/zcJd2RhH0RU+zX/77c0ZiOnIMsD
 github.com/lightningnetwork/lnd/clock v1.1.1/go.mod h1:mGnAhPyjYZQJmebS7aevElXKTFDuO+uNFFfMXK1W8xQ=
 github.com/lightningnetwork/lnd/healthcheck v1.2.3 h1:oqhOOy8WmIEa6RBkYKC0mmYZkhl8T2kGD97n9jpML8o=
 github.com/lightningnetwork/lnd/healthcheck v1.2.3/go.mod h1:eDxH3dEwV9DeBW/6inrmlVh1qBOFV0AI14EEPnGt9gc=
-github.com/lightningnetwork/lnd/kvdb v1.4.4 h1:bCv63rVCvzqj1BkagN/EWTov6NDDgYEG/t0z2HepRMk=
-github.com/lightningnetwork/lnd/kvdb v1.4.4/go.mod h1:9SuaIqMA9ugrVkdvgQkYXa8CAKYNYd4vsEYORP4V698=
 github.com/lightningnetwork/lnd/queue v1.1.1 h1:99ovBlpM9B0FRCGYJo6RSFDlt8/vOkQQZznVb18iNMI=
 github.com/lightningnetwork/lnd/queue v1.1.1/go.mod h1:7A6nC1Qrm32FHuhx/mi1cieAiBZo5O6l8IBIoQxvkz4=
 github.com/lightningnetwork/lnd/ticker v1.1.1 h1:J/b6N2hibFtC7JLV77ULQp++QLtCwT6ijJlbdiZFbSM=

kvdb/postgres/db.go

@@ -28,7 +28,6 @@ func newPostgresBackend(ctx context.Context, config *Config, prefix string) (
 		Schema:                "public",
 		TableNamePrefix:       prefix,
 		SQLiteCmdReplacements: sqliteCmdReplacements,
-		WithTxLevelLock:       true,
 	}
 
 	return sqlbase.NewSqlBackend(ctx, cfg)

kvdb/sqlbase/db.go

@@ -66,10 +66,6 @@ type Config struct {
 	// commands. Note that the sqlite keywords to be replaced are
 	// case-sensitive.
 	SQLiteCmdReplacements SQLiteCmdReplacements
-
-	// WithTxLevelLock when set will ensure that there is a transaction
-	// level lock.
-	WithTxLevelLock bool
 }
 
 // db holds a reference to the sql db connection.
@@ -90,10 +86,6 @@ type db struct {
 	// db is the underlying database connection instance.
 	db *sql.DB
 
-	// lock is the global write lock that ensures single writer. This is
-	// only used if cfg.WithTxLevelLock is set.
-	lock sync.RWMutex
-
 	// table is the name of the table that contains the data for all
 	// top-level buckets that have keys that cannot be mapped to a distinct
 	// sql table.
@@ -181,7 +173,6 @@ func (db *db) getPrefixedTableName(table string) string {
 func catchPanic(f func() error) (err error) {
 	defer func() {
 		if r := recover(); r != nil {
-			log.Criticalf("Caught unhandled error: %v", r)
 
 			switch data := r.(type) {
 			case error:
@@ -190,6 +181,18 @@ func catchPanic(f func() error) (err error) {
 			default:
 				err = errors.New(fmt.Sprintf("%v", data))
 			}
+
+			// Before we issue a critical log which'll cause the
+			// daemon to shutdown, we'll first check if this is a
+			// DB serialization error. If so, then we don't need to
+			// log as we can retry safely and avoid tearing
+			// everything down.
+			if IsSerializationError(MapSQLError(err)) {
+				log.Tracef("detected db serialization error "+
+					"via panic: %v", err)
+			} else {
+				log.Criticalf("Caught unhandled error: %v", r)
+			}
 		}
 	}()
 
@@ -308,26 +311,39 @@ func (db *db) executeTransaction(f func(tx walletdb.ReadWriteTx) error,
 			return dbErr
 		}
 
-		err = catchPanic(func() error { return f(tx) })
-		if err != nil {
+		fnErr := catchPanic(func() error { return f(tx) })
+		if fnErr != nil {
 			if rollbackErr := tx.Rollback(); rollbackErr != nil {
 				log.Errorf("Error rolling back tx: %v",
 					rollbackErr)
 			}
 
-			return err
+			dbErr := MapSQLError(fnErr)
+			if IsSerializationError(dbErr) {
+				// Nothing to roll back here, since we didn't
+				// even get a transaction yet.
+				if waitBeforeRetry(i) {
+					continue
+				}
+			}
+
+			return fnErr
 		}
 
-		dbErr := tx.Commit()
-		if IsSerializationError(dbErr) {
+		commitErr := tx.Commit()
+		if commitErr != nil {
 			_ = tx.Rollback()
 
-			if waitBeforeRetry(i) {
-				continue
+			dbErr := MapSQLError(fnErr)
+			if IsSerializationError(dbErr) {
+
+				if waitBeforeRetry(i) {
+					continue
+				}
 			}
 		}
 
-		return dbErr
+		return commitErr
 	}
 
 	// If we get to this point, then we weren't able to successfully commit

kvdb/sqlbase/readwrite_tx.go

@@ -5,7 +5,6 @@ package sqlbase
 import (
 	"context"
 	"database/sql"
-	"sync"
 
 	"github.com/btcsuite/btcwallet/walletdb"
 )
@@ -20,28 +19,11 @@ type readWriteTx struct {
 
 	// active is true if the transaction hasn't been committed yet.
 	active bool
-
-	// locker is a pointer to the global db lock.
-	locker sync.Locker
 }
 
 // newReadWriteTx creates an rw transaction using a connection from the
 // specified pool.
 func newReadWriteTx(db *db, readOnly bool) (*readWriteTx, error) {
-	locker := newNoopLocker()
-	if db.cfg.WithTxLevelLock {
-		// Obtain the global lock instance. An alternative here is to
-		// obtain a database lock from Postgres. Unfortunately there is
-		// no database-level lock in Postgres, meaning that each table
-		// would need to be locked individually. Perhaps an advisory
-		// lock could perform this function too.
-		locker = &db.lock
-		if readOnly {
-			locker = db.lock.RLocker()
-		}
-	}
-	locker.Lock()
-
 	// Start the transaction. Don't use the timeout context because it would
 	// be applied to the transaction as a whole. If possible, mark the
 	// transaction as read-only to make sure that potential programming
@@ -54,15 +36,13 @@ func newReadWriteTx(db *db, readOnly bool) (*readWriteTx, error) {
 		},
 	)
 	if err != nil {
-		locker.Unlock()
 		return nil, err
 	}
 
 	return &readWriteTx{
 		db:     db,
 		tx:     tx,
 		active: true,
-		locker: locker,
 	}, nil
 }
 
@@ -94,7 +74,6 @@ func (tx *readWriteTx) Rollback() error {
 
 	// Unlock the transaction regardless of the error result.
 	tx.active = false
-	tx.locker.Unlock()
 	return err
 }
 
@@ -162,7 +141,6 @@ func (tx *readWriteTx) Commit() error {
 
 	// Unlock the transaction regardless of the error result.
 	tx.active = false
-	tx.locker.Unlock()
 
 	return err
 }
@@ -204,25 +182,3 @@ func (tx *readWriteTx) Exec(query string, args ...interface{}) (sql.Result,
 
 	return tx.tx.ExecContext(ctx, query, args...)
 }
-
-// noopLocker is an implementation of a no-op sync.Locker.
-type noopLocker struct{}
-
-// newNoopLocker creates a new noopLocker.
-func newNoopLocker() sync.Locker {
-	return &noopLocker{}
-}
-
-// Lock is a noop.
-//
-// NOTE: this is part of the sync.Locker interface.
-func (n *noopLocker) Lock() {
-}
-
-// Unlock is a noop.
-//
-// NOTE: this is part of the sync.Locker interface.
-func (n *noopLocker) Unlock() {
-}
-
-var _ sync.Locker = (*noopLocker)(nil)

kvdb/sqlbase/sqlerrors_postgres.go

@@ -5,6 +5,7 @@ package sqlbase
 import (
 	"errors"
 	"fmt"
+	"strings"
 
 	"github.com/jackc/pgconn"
 	"github.com/jackc/pgerrcode"
@@ -13,6 +14,15 @@ import (
 // parsePostgresError attempts to parse a postgres error as a database agnostic
 // SQL error.
 func parsePostgresError(err error) error {
+	// Sometimes the error won't be properly wrapped, so we'll need to
+	// inspect raw error itself to detect something we can wrap properly.
+	const postgresErrMsg = "could not serialize access"
+	if strings.Contains(err.Error(), postgresErrMsg) {
+		return &ErrSerializationError{
+			DBError: err,
+		}
+	}
+
 	var pqErr *pgconn.PgError
 	if !errors.As(err, &pqErr) {
 		return nil

kvdb/sqlbase/sqlerrors_sqlite.go

@@ -5,6 +5,7 @@ package sqlbase
 import (
 	"errors"
 	"fmt"
+	"strings"
 
 	"modernc.org/sqlite"
 	sqlite3 "modernc.org/sqlite/lib"
@@ -13,6 +14,17 @@ import (
 // parseSqliteError attempts to parse a sqlite error as a database agnostic
 // SQL error.
 func parseSqliteError(err error) error {
+	// If the error isn't wrapped properly, the errors.As call with fail,
+	// so we'll also try to check the expected error message directly.
+	// This is taken from:
+	// https://gitlab.com/cznic/sqlite/-/blob/v1.25.0/sqlite.go#L75.
+	const sqliteErrMsg = "SQLITE_BUSY"
+	if strings.Contains(err.Error(), sqliteErrMsg) {
+		return &ErrSerializationError{
+			DBError: err,
+		}
+	}
+
 	var sqliteErr *sqlite.Error
 	if !errors.As(err, &sqliteErr) {
 		return nil

sqldb/sqlerrors.go

@@ -3,6 +3,7 @@ package sqldb
 import (
 	"errors"
 	"fmt"
+	"strings"
 
 	"github.com/jackc/pgconn"
 	"github.com/jackc/pgerrcode"
@@ -31,6 +32,26 @@ func MapSQLError(err error) error {
 		return parsePostgresError(pqErr)
 	}
 
+	// Sometimes the error won't be properly wrapped, so we'll need to
+	// inspect raw error itself to detect something we can wrap properly.
+	// This handles a postgres variant of the error.
+	const postgresErrMsg = "could not serialize access"
+	if strings.Contains(err.Error(), postgresErrMsg) {
+		return &ErrSerializationError{
+			DBError: err,
+		}
+	}
+
+	// We'll also attempt to catch this for sqlite, that uses a slightly
+	// different error message. This is taken from:
+	// https://gitlab.com/cznic/sqlite/-/blob/v1.25.0/sqlite.go#L75.
+	const sqliteErrMsg = "SQLITE_BUSY"
+	if strings.Contains(err.Error(), sqliteErrMsg) {
+		return &ErrSerializationError{
+			DBError: err,
+		}
+	}
+
 	// Return original error if it could not be classified as a database
 	// specific error.
 	return err