staticaddr: poll deposits on block arrival

In this commit we remove the ticker-based polling
of new deposits as this is wasteful, and we replace
it by polling on the arrival of new blocks.

Author: hieblmi

staticaddr/deposit/manager.go

@@ -74,22 +74,25 @@ type ManagerConfig struct {
 type Manager struct {
 	cfg *ManagerConfig
 
-	// mu guards access to activeDeposits map.
+	// mu guards access to the activeDeposits map.
 	mu sync.Mutex
 
 	// activeDeposits contains all the active static address outputs.
 	activeDeposits map[wire.OutPoint]*FSM
 
-	// deposits contains all the deposits that have ever been made to the
+	// deposits contain all the deposits that have ever been made to the
 	// static address. This field is used to store and recover deposits. It
-	// also serves as basis for reconciliation of newly detected deposits by
-	// matching them against deposits in this map that were already seen.
+	// also serves as a basis for reconciliation of newly detected deposits
+	// by matching them against deposits in this map that were already seen.
 	deposits map[wire.OutPoint]*Deposit
 
 	// finalizedDepositChan is a channel that receives deposits that have
 	// been finalized. The manager will adjust its internal state and flush
 	// finalized deposits from its memory.
 	finalizedDepositChan chan wire.OutPoint
+
+	// pollingGate ensures only one reconcileDeposits runs at a time.
+	pollingGate chan struct{}
 }
 
 // NewManager creates a new deposit manager.
@@ -99,6 +102,7 @@ func NewManager(cfg *ManagerConfig) *Manager {
 		activeDeposits:       make(map[wire.OutPoint]*FSM),
 		deposits:             make(map[wire.OutPoint]*Deposit),
 		finalizedDepositChan: make(chan wire.OutPoint),
+		pollingGate:          make(chan struct{}, 1),
 	}
 }
 
@@ -119,8 +123,9 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error {
 		return err
 	}
 
-	// Start the deposit notifier.
-	m.pollDeposits(ctx)
+	// Poll for deposits once at startup to detect any that occurred while
+	// offline.
+	go m.pollDeposits(ctx)
 
 	// Communicate to the caller that the address manager has completed its
 	// initialization.
@@ -129,7 +134,11 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error {
 	for {
 		select {
 		case height := <-newBlockChan:
-			// Inform all active deposits about a new block arrival.
+			// Poll for new deposits and spin up fsms for them.
+			go m.pollDeposits(ctx)
+
+			// Now inform all active deposits about a new block
+			// arrival.
 			m.mu.Lock()
 			activeDeposits := make([]*FSM, 0, len(m.activeDeposits))
 			for _, fsm := range m.activeDeposits {
@@ -213,28 +222,25 @@ func (m *Manager) recoverDeposits(ctx context.Context) error {
 	return nil
 }
 
-// pollDeposits polls new deposits to our static address and notifies the
-// manager's event loop about them.
+// pollDeposits polls new deposits to our static addresses.
 func (m *Manager) pollDeposits(ctx context.Context) {
-	log.Debugf("Waiting for new static address deposits...")
+	log.Debugf("Polling new static address deposits...")
 
-	go func() {
-		ticker := time.NewTicker(PollInterval)
-		defer ticker.Stop()
-		for {
-			select {
-			case <-ticker.C:
-				err := m.reconcileDeposits(ctx)
-				if err != nil {
-					log.Errorf("unable to reconcile "+
-						"deposits: %v", err)
-				}
-
-			case <-ctx.Done():
-				return
-			}
+	select {
+	case m.pollingGate <- struct{}{}:
+		defer func() { <-m.pollingGate }()
+		if err := m.reconcileDeposits(ctx); err != nil {
+			log.Errorf("unable to reconcile deposits: %v",
+				err)
 		}
-	}()
+
+	case <-ctx.Done():
+		return
+
+	default:
+		log.Tracef("Polling new static address deposits already in " +
+			"progress, skipping")
+	}
 }
 
 // reconcileDeposits fetches all spends to our static address from our lnd