Merge pull request #10428 from ziggie1984/fix-sql-pool-exhaustion

graphdb: fix potential sql tx exhaustion

Author: yyforyongyu

docs/release-notes/release-notes-0.20.1.md

@@ -42,6 +42,10 @@
   in the mission control store. Now we skip over potential errors and also
   delete them from the store.
 
+* [Fix potential sql tx exhaustion 
+  issue](https://github.com/lightningnetwork/lnd/pull/10428) in LND which might
+  happen when running postgres with a limited number of connections configured.
+
 # New Features
 
 ## Functional Enhancements

graph/db/kv_store.go

@@ -2108,6 +2108,13 @@ func (c *KVStore) fetchNextChanUpdateBatch(
 		batch   []ChannelEdge
 		hasMore bool
 	)
+
+	// Acquire read lock before starting transaction to ensure
+	// consistent lock ordering (cacheMu -> DB) and prevent
+	// deadlock with write operations.
+	c.cacheMu.RLock()
+	defer c.cacheMu.RUnlock()
+
 	err := kvdb.View(c.db, func(tx kvdb.RTx) error {
 		edges := tx.ReadBucket(edgeBucket)
 		if edges == nil {
@@ -2187,9 +2194,7 @@ func (c *KVStore) fetchNextChanUpdateBatch(
 				continue
 			}
 
-			// Before we read the edge info, we'll see if this
-			// element is already in the cache or not.
-			c.cacheMu.RLock()
+			// Check cache (we already hold shared read lock).
 			if channel, ok := c.chanCache.get(chanIDInt); ok {
 				state.edgesSeen[chanIDInt] = struct{}{}
 
@@ -2200,11 +2205,8 @@ func (c *KVStore) fetchNextChanUpdateBatch(
 
 				indexKey, _ = updateCursor.Next()
 
-				c.cacheMu.RUnlock()
-
 				continue
 			}
-			c.cacheMu.RUnlock()
 
 			// The edge wasn't in the cache, so we'll fetch it along
 			// w/ the edge policies and nodes.

graph/db/sql_store.go

@@ -1127,6 +1127,11 @@ func (s *SQLStore) ChanUpdatesInHorizon(startTime, endTime time.Time,
 		for hasMore {
 			var batch []ChannelEdge
 
+			// Acquire read lock before starting transaction to
+			// ensure consistent lock ordering (cacheMu -> DB) and
+			// prevent deadlock with write operations.
+			s.cacheMu.RLock()
+
 			err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(),
 				func(db SQLQueries) error {
 					//nolint:ll
@@ -1179,11 +1184,11 @@ func (s *SQLStore) ChanUpdatesInHorizon(startTime, endTime time.Time,
 							continue
 						}
 
-						s.cacheMu.RLock()
+						// Check cache (we already hold
+						// shared read lock).
 						channel, ok := s.chanCache.get(
 							chanIDInt,
 						)
-						s.cacheMu.RUnlock()
 						if ok {
 							hits++
 							total++
@@ -1217,6 +1222,9 @@ func (s *SQLStore) ChanUpdatesInHorizon(startTime, endTime time.Time,
 					)
 				})
 
+			// Release read lock after transaction completes.
+			s.cacheMu.RUnlock()
+
 			if err != nil {
 				log.Errorf("ChanUpdatesInHorizon "+
 					"batch error: %v", err)