staticaddr: poll new deposits on block arrival

Author: hieblmi

staticaddr/deposit/manager.go

@@ -19,9 +19,8 @@ import (
 )
 
 const (
-	// PollInterval is the interval in which we poll for new deposits to our
-	// static address.
-	PollInterval = 10 * time.Second
+	// PollTimeout is the timeout for polling for new deposits.
+	PollTimeout = 10 * time.Second
 
 	// MinConfs is the minimum number of confirmations we require for a
 	// deposit to be considered available for loop-ins, coop-spends and
@@ -74,16 +73,16 @@ 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
@@ -119,8 +118,11 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error {
 		return err
 	}
 
-	// Start the deposit notifier.
-	m.pollDeposits(ctx)
+	// Initially poll for new deposits after a restart, so we catch up with
+	// missed deposits while we were offline.
+	timeoutCtx, cancel := context.WithTimeout(ctx, PollTimeout)
+	defer cancel()
+	m.pollDeposits(timeoutCtx)
 
 	// Communicate to the caller that the address manager has completed its
 	// initialization.
@@ -149,6 +151,16 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error {
 				}
 			}
 
+			// Poll for new deposits.
+			func() {
+				timeoutCtx, cancel = context.WithTimeout(
+					ctx, PollTimeout,
+				)
+				defer cancel()
+
+				m.pollDeposits(timeoutCtx)
+			}()
+
 		case outpoint := <-m.finalizedDepositChan:
 			// If deposits notify us about their finalization, flush
 			// the finalized deposit from memory.
@@ -213,28 +225,30 @@ func (m *Manager) recoverDeposits(ctx context.Context) error {
 	return nil
 }
 
+// pollDepositsWithTimeout creates a timeout context and polls for new deposits.
+func (m *Manager) pollDepositsWithTimeout(ctx context.Context) error {
+	timeoutCtx, cancel := context.WithTimeout(ctx, PollTimeout)
+	defer cancel()
+
+	m.pollDeposits(timeoutCtx)
+	return nil
+}
+
 // pollDeposits polls new deposits to our static address and notifies the
 // manager's event loop about them.
 func (m *Manager) pollDeposits(ctx context.Context) {
-	log.Debugf("Waiting for new static address deposits...")
+	log.Debugf("Polling for 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)
-				}
+	select {
+	case <-ctx.Done():
+		log.Debugf("Context cancelled, exiting deposit notifier...")
 
-			case <-ctx.Done():
-				return
-			}
+	default:
+		err := m.reconcileDeposits(ctx)
+		if err != nil {
+			log.Errorf("unable to reconcile deposits: %v", err)
 		}
-	}()
+	}
 }
 
 // reconcileDeposits fetches all spends to our static address from our lnd