Merge pull request #902 from hieblmi/fix-deadlock

staticaddr: lock mutex for active deposits access

looprpc/go.mod

@@ -1,6 +1,7 @@
 module github.com/lightninglabs/loop/looprpc
 
-go 1.22.3
+go 1.23.6
+
 toolchain go1.23.7
 
 require (

staticaddr/deposit/actions.go

@@ -147,12 +147,9 @@ func (f *FSM) SweptExpiredDepositAction(ctx context.Context,
 	case <-ctx.Done():
 		return fsm.OnError
 
-	default:
-		f.finalizedDepositChan <- f.deposit.OutPoint
-		ctx.Done()
+	case f.finalizedDepositChan <- f.deposit.OutPoint:
+		return fsm.NoOp
 	}
-
-	return fsm.NoOp
 }
 
 // FinalizeDepositAction is the final action after a withdrawal. It signals to
@@ -164,9 +161,7 @@ func (f *FSM) FinalizeDepositAction(ctx context.Context,
 	case <-ctx.Done():
 		return fsm.OnError
 
-	default:
-		f.finalizedDepositChan <- f.deposit.OutPoint
+	case f.finalizedDepositChan <- f.deposit.OutPoint:
+		return fsm.NoOp
 	}
-
-	return fsm.NoOp
 }

staticaddr/deposit/fsm.go

@@ -141,6 +141,11 @@ type FSM struct {
 
 	blockNtfnChan chan uint32
 
+	// quitChan stops after the FSM stops consuming blockNtfnChan.
+	quitChan chan struct{}
+
+	// finalizedDepositChan is used to signal that the deposit has been
+	// finalized and the FSM can be removed from the manager's memory.
 	finalizedDepositChan chan wire.OutPoint
 }
 
@@ -167,6 +172,7 @@ func NewFSM(ctx context.Context, deposit *Deposit, cfg *ManagerConfig,
 		params:               params,
 		address:              address,
 		blockNtfnChan:        make(chan uint32),
+		quitChan:             make(chan struct{}),
 		finalizedDepositChan: finalizedDepositChan,
 	}
 
@@ -191,10 +197,12 @@ func NewFSM(ctx context.Context, deposit *Deposit, cfg *ManagerConfig,
 
 	depoFsm.ActionEntryFunc = depoFsm.updateDeposit
 
-	go func() {
+	go func(fsm *FSM) {
+		defer close(fsm.quitChan)
+
 		for {
 			select {
-			case currentHeight := <-depoFsm.blockNtfnChan:
+			case currentHeight := <-fsm.blockNtfnChan:
 				depoFsm.handleBlockNotification(
 					ctx, currentHeight,
 				)
@@ -203,7 +211,7 @@ func NewFSM(ctx context.Context, deposit *Deposit, cfg *ManagerConfig,
 				return
 			}
 		}
-	}()
+	}(depoFsm)
 
 	return depoFsm, nil
 }

staticaddr/deposit/manager.go

@@ -131,19 +131,31 @@ func (m *Manager) Run(ctx context.Context) error {
 		select {
 		case height := <-newBlockChan:
 			// Inform all active deposits about a new block arrival.
+			m.mu.Lock()
+			activeDeposits := make([]*FSM, 0, len(m.activeDeposits))
 			for _, fsm := range m.activeDeposits {
+				activeDeposits = append(activeDeposits, fsm)
+			}
+			m.mu.Unlock()
+
+			for _, fsm := range activeDeposits {
 				select {
 				case fsm.blockNtfnChan <- uint32(height):
 
+				case <-fsm.quitChan:
+					continue
+
 				case <-ctx.Done():
 					return ctx.Err()
 				}
 			}
+
 		case outpoint := <-m.finalizedDepositChan:
-			// If deposits notify us about their finalization, we
-			// update the manager's internal state and flush the
-			// finalized deposit from memory.
+			// If deposits notify us about their finalization, flush
+			// the finalized deposit from memory.
+			m.mu.Lock()
 			delete(m.activeDeposits, outpoint)
+			m.mu.Unlock()
 
 		case err = <-newBlockErrChan:
 			return err
@@ -192,7 +204,9 @@ func (m *Manager) recoverDeposits(ctx context.Context) error {
 			}
 		}()
 
+		m.mu.Lock()
 		m.activeDeposits[d.OutPoint] = fsm
+		m.mu.Unlock()
 	}
 
 	return nil
@@ -488,9 +502,9 @@ func (m *Manager) AllStringOutpointsActiveDeposits(outpoints []string,
 // TransitionDeposits allows a caller to transition a set of deposits to a new
 // state.
 // Caveat: The action triggered by the state transitions should not compute
-// heavy things or call external endpoints that can block for a long time.
-// Deposits will be released if a transition takes longer than
-// DefaultTransitionTimeout which is set to 5 seconds.
+// heavy things or call external endpoints that can block for a long time as
+// this function blocks until the expectedFinalState is reached. The default
+// timeout for the transition is set to DefaultTransitionTimeout.
 func (m *Manager) TransitionDeposits(ctx context.Context, deposits []*Deposit,
 	event fsm.EventType, expectedFinalState fsm.StateType) error {
 
@@ -500,9 +514,9 @@ func (m *Manager) TransitionDeposits(ctx context.Context, deposits []*Deposit,
 	}
 
 	m.mu.Lock()
-	defer m.mu.Unlock()
-
 	stateMachines, _ := m.toActiveDeposits(&outpoints)
+	m.mu.Unlock()
+
 	if stateMachines == nil {
 		return fmt.Errorf("deposits not found in active deposits")
 	}