sqldb+kvdb: unify SQL error mapping and transaction retry for between the kvdb and sqldb packages

autopilot/agent.go

@@ -648,7 +648,7 @@ func (a *Agent) openChans(availableFunds btcutil.Amount, numChans uint32,
 	// to open channels to.
 	scores, err = chooseN(numChans, scores)
 	if err != nil {
-		return fmt.Errorf("unable to make weighted choice: %v",
+		return fmt.Errorf("unable to make weighted choice: %w",
 			err)
 	}
 

autopilot/combinedattach.go

@@ -84,7 +84,7 @@ func (c *WeightedCombAttachment) NodeScores(g ChannelGraph, chans []LocalChannel
 			g, chans, chanSize, nodes,
 		)
 		if err != nil {
-			return nil, fmt.Errorf("unable to get sub score: %v",
+			return nil, fmt.Errorf("unable to get sub score: %w",
 				err)
 		}
 

autopilot/manager.go

@@ -341,7 +341,7 @@ func (m *Manager) queryHeuristics(nodes map[NodeID]struct{}, localState bool) (
 			m.cfg.PilotCfg.Graph, totalChans, chanSize, nodes,
 		)
 		if err != nil {
-			return nil, fmt.Errorf("unable to get sub score: %v",
+			return nil, fmt.Errorf("unable to get sub score: %w",
 				err)
 		}
 

brontide/listener.go

@@ -91,7 +91,7 @@ func (l *Listener) listen() {
 // rejectedConnErr is a helper function that prepends the remote address of the
 // failed connection attempt to the original error message.
 func rejectedConnErr(err error, remoteAddr string) error {
-	return fmt.Errorf("unable to accept connection from %v: %v", remoteAddr,
+	return fmt.Errorf("unable to accept connection from %v: %w", remoteAddr,
 		err)
 }
 

cert/selfsigned.go

@@ -261,7 +261,7 @@ func GenCertPair(org string, tlsExtraIPs, tlsExtraDomains []string,
 		&template, &priv.PublicKey, priv,
 	)
 	if err != nil {
-		return nil, nil, fmt.Errorf("failed to create certificate: %v",
+		return nil, nil, fmt.Errorf("failed to create certificate: %w",
 			err)
 	}
 
@@ -270,21 +270,21 @@ func GenCertPair(org string, tlsExtraIPs, tlsExtraDomains []string,
 		certBuf, &pem.Block{Type: "CERTIFICATE", Bytes: derBytes},
 	)
 	if err != nil {
-		return nil, nil, fmt.Errorf("failed to encode certificate: %v",
+		return nil, nil, fmt.Errorf("failed to encode certificate: %w",
 			err)
 	}
 
 	keybytes, err := x509.MarshalECPrivateKey(priv)
 	if err != nil {
-		return nil, nil, fmt.Errorf("unable to encode privkey: %v",
+		return nil, nil, fmt.Errorf("unable to encode privkey: %w",
 			err)
 	}
 	keyBuf := &bytes.Buffer{}
 	err = pem.Encode(
 		keyBuf, &pem.Block{Type: "EC PRIVATE KEY", Bytes: keybytes},
 	)
 	if err != nil {
-		return nil, nil, fmt.Errorf("failed to encode private key: %v",
+		return nil, nil, fmt.Errorf("failed to encode private key: %w",
 			err)
 	}
 

chainntnfs/bitcoindnotify/bitcoind.go

@@ -792,8 +792,8 @@ func (b *BitcoindNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint,
 		// proceed with fallback methods.
 		jsonErr, ok := err.(*btcjson.RPCError)
 		if !ok || jsonErr.Code != btcjson.ErrRPCNoTxInfo {
-			return nil, fmt.Errorf("unable to query for txid %v: %v",
-				outpoint.Hash, err)
+			return nil, fmt.Errorf("unable to query for txid "+
+				"%v: %w", outpoint.Hash, err)
 		}
 	}
 

chainntnfs/btcdnotify/btcd.go

@@ -886,8 +886,8 @@ func (b *BtcdNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint,
 		// proceed with fallback methods.
 		jsonErr, ok := err.(*btcjson.RPCError)
 		if !ok || jsonErr.Code != btcjson.ErrRPCNoTxInfo {
-			return nil, fmt.Errorf("unable to query for txid %v: %v",
-				outpoint.Hash, err)
+			return nil, fmt.Errorf("unable to query for txid %v: "+
+				"%w", outpoint.Hash, err)
 		}
 	}
 

chainntnfs/interface.go

@@ -483,22 +483,22 @@ func GetCommonBlockAncestorHeight(chainConn ChainConn, reorgHash,
 	for reorgHash != chainHash {
 		reorgHeader, err := chainConn.GetBlockHeader(&reorgHash)
 		if err != nil {
-			return 0, fmt.Errorf("unable to get header for hash=%v: %v",
-				reorgHash, err)
+			return 0, fmt.Errorf("unable to get header for "+
+				"hash=%v: %w", reorgHash, err)
 		}
 		chainHeader, err := chainConn.GetBlockHeader(&chainHash)
 		if err != nil {
-			return 0, fmt.Errorf("unable to get header for hash=%v: %v",
-				chainHash, err)
+			return 0, fmt.Errorf("unable to get header for "+
+				"hash=%v: %w", chainHash, err)
 		}
 		reorgHash = reorgHeader.PrevBlock
 		chainHash = chainHeader.PrevBlock
 	}
 
 	verboseHeader, err := chainConn.GetBlockHeaderVerbose(&chainHash)
 	if err != nil {
-		return 0, fmt.Errorf("unable to get verbose header for hash=%v: %v",
-			chainHash, err)
+		return 0, fmt.Errorf("unable to get verbose header for "+
+			"hash=%v: %w", chainHash, err)
 	}
 
 	return verboseHeader.Height, nil
@@ -719,7 +719,7 @@ func ConfDetailsFromTxIndex(chainConn TxIndexConn, r ConfRequest,
 		}
 
 		return nil, TxNotFoundIndex,
-			fmt.Errorf("unable to query for txid %v: %v",
+			fmt.Errorf("unable to query for txid %v: %w",
 				r.TxID, err)
 	}
 
@@ -728,13 +728,13 @@ func ConfDetailsFromTxIndex(chainConn TxIndexConn, r ConfRequest,
 	rawTx, err := hex.DecodeString(rawTxRes.Hex)
 	if err != nil {
 		return nil, TxNotFoundIndex,
-			fmt.Errorf("unable to deserialize tx %v: %v",
+			fmt.Errorf("unable to deserialize tx %v: %w",
 				r.TxID, err)
 	}
 	var tx wire.MsgTx
 	if err := tx.Deserialize(bytes.NewReader(rawTx)); err != nil {
 		return nil, TxNotFoundIndex,
-			fmt.Errorf("unable to deserialize tx %v: %v",
+			fmt.Errorf("unable to deserialize tx %v: %w",
 				r.TxID, err)
 	}
 
@@ -759,13 +759,14 @@ func ConfDetailsFromTxIndex(chainConn TxIndexConn, r ConfRequest,
 	if err != nil {
 		return nil, TxNotFoundIndex,
 			fmt.Errorf("unable to get block hash %v for "+
-				"historical dispatch: %v", rawTxRes.BlockHash, err)
+				"historical dispatch: %w", rawTxRes.BlockHash,
+				err)
 	}
 	block, err := chainConn.GetBlock(blockHash)
 	if err != nil {
 		return nil, TxNotFoundIndex,
 			fmt.Errorf("unable to get block with hash %v for "+
-				"historical dispatch: %v", blockHash, err)
+				"historical dispatch: %w", blockHash, err)
 	}
 
 	// In the modern chain (the only one we really care about for LN), the

chainntnfs/neutrinonotify/neutrino.go

@@ -582,8 +582,8 @@ func (n *NeutrinoNotifier) historicalConfDetails(confRequest chainntnfs.ConfRequ
 		// can compute the current block hash.
 		blockHash, err := n.p2pNode.GetBlockHash(int64(scanHeight))
 		if err != nil {
-			return nil, fmt.Errorf("unable to get header for height=%v: %v",
-				scanHeight, err)
+			return nil, fmt.Errorf("unable to get header for "+
+				"height=%v: %w", scanHeight, err)
 		}
 
 		// With the hash computed, we can now fetch the basic filter for this
@@ -600,8 +600,8 @@ func (n *NeutrinoNotifier) historicalConfDetails(confRequest chainntnfs.ConfRequ
 			neutrino.MaxBatchSize(int64(scanHeight-startHeight+1)),
 		)
 		if err != nil {
-			return nil, fmt.Errorf("unable to retrieve regular filter for "+
-				"height=%v: %v", scanHeight, err)
+			return nil, fmt.Errorf("unable to retrieve regular "+
+				"filter for height=%v: %w", scanHeight, err)
 		}
 
 		// In the case that the filter exists, we'll attempt to see if

chainreg/chainregistry.go

@@ -763,11 +763,11 @@ func NewChainControl(walletConfig lnwallet.Config,
 
 	lnWallet, err := lnwallet.NewLightningWallet(walletConfig)
 	if err != nil {
-		return nil, ccCleanup, fmt.Errorf("unable to create wallet: %v",
+		return nil, ccCleanup, fmt.Errorf("unable to create wallet: %w",
 			err)
 	}
 	if err := lnWallet.Startup(); err != nil {
-		return nil, ccCleanup, fmt.Errorf("unable to create wallet: %v",
+		return nil, ccCleanup, fmt.Errorf("unable to create wallet: %w",
 			err)
 	}
 

channeldb/codec.go

@@ -78,7 +78,8 @@ func WriteElement(w io.Writer, element interface{}) error {
 
 		if e.PubKey != nil {
 			if err := binary.Write(w, byteOrder, true); err != nil {
-				return fmt.Errorf("error writing serialized element: %s", err)
+				return fmt.Errorf("error writing serialized "+
+					"element: %w", err)
 			}
 
 			return WriteElement(w, e.PubKey)

channeldb/db.go

@@ -643,7 +643,7 @@ func (c *ChannelStateDB) fetchNodeChannels(chainBucket kvdb.RBucket) (
 		oChannel, err := fetchOpenChannel(chanBucket, &outPoint)
 		if err != nil {
 			return fmt.Errorf("unable to read channel data for "+
-				"chan_point=%v: %v", outPoint, err)
+				"chan_point=%v: %w", outPoint, err)
 		}
 		oChannel.Db = c
 

channeldb/graph.go

@@ -1508,9 +1508,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",

channeldb/migration21/current/current_codec.go

@@ -83,7 +83,8 @@ func WriteElement(w io.Writer, element interface{}) error {
 
 		if e.PubKey != nil {
 			if err := binary.Write(w, byteOrder, true); err != nil {
-				return fmt.Errorf("error writing serialized element: %s", err)
+				return fmt.Errorf("error writing serialized "+
+					"element: %w", err)
 			}
 
 			return WriteElement(w, e.PubKey)

channeldb/migration21/legacy/legacy_codec.go

@@ -81,7 +81,8 @@ func WriteElement(w io.Writer, element interface{}) error {
 
 		if e.PubKey != nil {
 			if err := binary.Write(w, byteOrder, true); err != nil {
-				return fmt.Errorf("error writing serialized element: %s", err)
+				return fmt.Errorf("error writing serialized "+
+					"element: %w", err)
 			}
 
 			return WriteElement(w, e.PubKey)

channeldb/migration25/migration.go

@@ -93,7 +93,7 @@ func findOpenChannels(openChanBucket kvdb.RBucket) ([]*OpenChannel, error) {
 		// open channels as they don't have any revocation logs and
 		// their current commitments reflect the initial balances.
 		if err := FetchChanCommitments(chanBucket, c); err != nil {
-			return fmt.Errorf("unable to fetch chan commits: %v",
+			return fmt.Errorf("unable to fetch chan commits: %w",
 				err)
 		}
 

channeldb/migration27/migration.go

@@ -79,7 +79,7 @@ func findHistoricalChannels(historicalBucket kvdb.RBucket) ([]*OpenChannel,
 		// Try to fetch channel info in old format.
 		err = fetchChanInfoCompatible(chanBucket, c, true)
 		if err != nil {
-			return fmt.Errorf("%s: fetch chan info got: %v",
+			return fmt.Errorf("%s: fetch chan info got: %w",
 				c.FundingOutpoint, err)
 		}
 

channeldb/migration30/migration.go

@@ -254,7 +254,7 @@ func writeRevocationLogs(openChanBucket kvdb.RwBucket,
 			logEntrykey := mig24.MakeLogKey(entry.commitHeight)
 			err = logBucket.Put(logEntrykey[:], b.Bytes())
 			if err != nil {
-				return fmt.Errorf("putRevocationLog err: %v",
+				return fmt.Errorf("putRevocationLog err: %w",
 					err)
 			}
 		}

channeldb/migration_01_to_11/codec.go

@@ -70,7 +70,7 @@ func WriteElement(w io.Writer, element interface{}) error {
 
 		if e.PubKey != nil {
 			if err := binary.Write(w, byteOrder, true); err != nil {
-				return fmt.Errorf("error writing serialized element: %s", err)
+				return fmt.Errorf("error writing serialized element: %w", err)
 			}
 
 			return WriteElement(w, e.PubKey)

channeldb/migration_01_to_11/graph.go

@@ -1140,7 +1140,7 @@ func deserializeChanEdgePolicy(r io.Reader,
 
 	node, err := fetchLightningNode(nodes, pub[:])
 	if err != nil {
-		return nil, fmt.Errorf("unable to fetch node: %x, %v",
+		return nil, fmt.Errorf("unable to fetch node: %x, %w",
 			pub[:], err)
 	}
 	edge.Node = &node

channeldb/migration_01_to_11/migrations.go

@@ -516,7 +516,7 @@ func MigratePruneEdgeUpdateIndex(tx kvdb.RwTx) error {
 	// well.
 	edgeIndex, err := edges.CreateBucketIfNotExists(edgeIndexBucket)
 	if err != nil {
-		return fmt.Errorf("error creating edge index bucket: %s", err)
+		return fmt.Errorf("error creating edge index bucket: %w", err)
 	}
 	if edgeIndex == nil {
 		return fmt.Errorf("unable to create/fetch edge index " +
@@ -546,7 +546,7 @@ func MigratePruneEdgeUpdateIndex(tx kvdb.RwTx) error {
 		return nil
 	})
 	if err != nil {
-		return fmt.Errorf("unable to gather existing edge policies: %v",
+		return fmt.Errorf("unable to gather existing edge policies: %w",
 			err)
 	}
 
@@ -560,7 +560,7 @@ func MigratePruneEdgeUpdateIndex(tx kvdb.RwTx) error {
 		return nil
 	})
 	if err != nil {
-		return fmt.Errorf("unable to gather existing edge updates: %v",
+		return fmt.Errorf("unable to gather existing edge updates: %w",
 			err)
 	}
 

channeldb/migration_01_to_11/zpay32/amountunits.go

@@ -136,7 +136,7 @@ func encodeAmount(msat lnwire.MilliSatoshi) (string, error) {
 	// Should always be expressible in pico BTC.
 	pico, err := fromMSat['p'](msat)
 	if err != nil {
-		return "", fmt.Errorf("unable to express %d msat as pBTC: %v",
+		return "", fmt.Errorf("unable to express %d msat as pBTC: %w",
 			msat, err)
 	}
 	shortened := strconv.FormatUint(pico, 10) + "p"

channeldb/payment_control.go

@@ -229,7 +229,7 @@ func (p *PaymentControl) InitPayment(paymentHash lntypes.Hash,
 		return bucket.Delete(paymentFailInfoKey)
 	})
 	if err != nil {
-		return err
+		return fmt.Errorf("unable to init payment: %w", err)
 	}
 
 	return updateErr

channeldb/payment_control_test.go

@@ -181,7 +181,7 @@ func TestPaymentControlSwitchFail(t *testing.T) {
 	// Attempt a final payment, which should now fail since the prior
 	// payment succeed.
 	err = pControl.InitPayment(info.PaymentIdentifier, info)
-	if err != ErrAlreadyPaid {
+	if !errors.Is(err, ErrAlreadyPaid) {
 		t.Fatalf("unable to send htlc message: %v", err)
 	}
 }
@@ -216,9 +216,7 @@ func TestPaymentControlSwitchDoubleSend(t *testing.T) {
 	// payment hash, should result in error indicating that payment has
 	// already been sent.
 	err = pControl.InitPayment(info.PaymentIdentifier, info)
-	require.Equal(t, ErrPaymentExists, err, "payment control wrong "+
-		"behaviour: init payment again must trigger ErrPaymentExists "+
-		"error")
+	require.ErrorIs(t, err, ErrPaymentExists)
 
 	// Record an attempt.
 	_, err = pControl.RegisterAttempt(info.PaymentIdentifier, attempt)
@@ -234,7 +232,7 @@ func TestPaymentControlSwitchDoubleSend(t *testing.T) {
 
 	// Sends base htlc message which initiate StatusInFlight.
 	err = pControl.InitPayment(info.PaymentIdentifier, info)
-	if err != ErrPaymentInFlight {
+	if !errors.Is(err, ErrPaymentInFlight) {
 		t.Fatalf("payment control wrong behaviour: " +
 			"double sending must trigger ErrPaymentInFlight error")
 	}
@@ -253,7 +251,7 @@ func TestPaymentControlSwitchDoubleSend(t *testing.T) {
 	assertPaymentInfo(t, pControl, info.PaymentIdentifier, info, nil, htlc)
 
 	err = pControl.InitPayment(info.PaymentIdentifier, info)
-	if err != ErrAlreadyPaid {
+	if !errors.Is(err, ErrAlreadyPaid) {
 		t.Fatalf("unable to send htlc message: %v", err)
 	}
 }

channeldb/payments.go

@@ -615,7 +615,7 @@ func (d *DB) QueryPayments(query PaymentsQuery) (PaymentsResponse, error) {
 				err = indexes.ForEach(countFn)
 			}
 			if err != nil {
-				return fmt.Errorf("error counting payments: %v",
+				return fmt.Errorf("error counting payments: %w",
 					err)
 			}