Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
3 changes: 2 additions & 1 deletion looprpc/go.mod
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
module github.com/lightninglabs/loop/looprpc

go 1.22.3
go 1.23.6

toolchain go1.23.7

require (
Expand Down
13 changes: 4 additions & 9 deletions staticaddr/deposit/actions.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
}
14 changes: 11 additions & 3 deletions staticaddr/deposit/fsm.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
}

Expand All @@ -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,
}

Expand All @@ -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,
)
Expand All @@ -203,7 +211,7 @@ func NewFSM(ctx context.Context, deposit *Deposit, cfg *ManagerConfig,
return
}
}
}()
}(depoFsm)
Copy link
Collaborator

Choose a reason for hiding this comment

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

Why is this passing-as-an-argument needed? Previous version seems also ok.

Copy link
Collaborator Author

Choose a reason for hiding this comment

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

usually this pattern is better within loops, but I also find in more clear, so keeping it.


return depoFsm, nil
}
Expand Down
30 changes: 22 additions & 8 deletions staticaddr/deposit/manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -136,19 +136,31 @@ func (m *Manager) Run(ctx context.Context, currentHeight uint32) 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
Expand Down Expand Up @@ -197,7 +209,9 @@ func (m *Manager) recoverDeposits(ctx context.Context) error {
}
}()

m.mu.Lock()
m.activeDeposits[d.OutPoint] = fsm
m.mu.Unlock()
}

return nil
Expand Down Expand Up @@ -493,9 +507,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 {

Expand All @@ -505,9 +519,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")
}
Expand Down