assets+loopdb: recover active deposits on startup

This commit adds deposit recovery to the deposit manager's startup
process. All deposits that have not yet been spent by the server or
swept by the client are recovered from the store.

Author: bhandras

assets/deposit/manager.go

@@ -146,6 +146,13 @@ func (m *Manager) Run(ctx context.Context, bestBlock uint32) error {
 
 	m.currentHeight = bestBlock
 
+	err := m.recoverDeposits(ctx)
+	if err != nil {
+		log.Errorf("Unable to recover deposits: %v", err)
+
+		return err
+	}
+
 	blockChan, blockErrChan, err := m.chainNotifier.RegisterBlockEpochNtfn(
 		ctxc,
 	)
@@ -219,6 +226,48 @@ func (m *Manager) criticalError(err error) {
 	}
 }
 
+// recoverDeposits recovers all active deppsits when the deposit manager starts.
+func (m *Manager) recoverDeposits(ctx context.Context) error {
+	// Fetch all active deposits from the store to kick-off the manager.
+	activeDeposits, err := m.store.GetActiveDeposits(ctx)
+	if err != nil {
+		log.Errorf("Unable to fetch deposits from store: %v", err)
+
+		return err
+	}
+
+	for i := range activeDeposits {
+		d := &activeDeposits[i]
+		log.Infof("Recovering deposit %v (state=%s)", d.ID, d.State)
+
+		m.deposits[d.ID] = d
+		_, _, _, err = m.isDepositFunded(ctx, d)
+		if err != nil {
+			return err
+		}
+
+		if d.State == StateInitiated {
+			// If the deposit has just been initiated, then we need
+			// to ensure that it is funded.
+			err = m.fundDepositIfNeeded(ctx, d)
+			if err != nil {
+				log.Errorf("Unable to fund deposit %v: %v",
+					d.ID, err)
+
+				return err
+			}
+		} else {
+			// Cache proof info of the deposit in-memory.
+			err = m.cacheProofInfo(ctx, d)
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
 // handleBlockEpoch is called when a new block is added to the chain.
 func (m *Manager) handleBlockEpoch(ctx context.Context, height uint32) error {
 	for _, d := range m.deposits {

assets/deposit/sql_store.go

@@ -32,6 +32,9 @@ type Querier interface {
 
 	SetAssetDepositSweepAddr(ctx context.Context,
 		arg sqlc.SetAssetDepositSweepAddrParams) error
+
+	GetActiveAssetDeposits(ctx context.Context) (
+		[]sqlc.GetActiveAssetDepositsRow, error)
 }
 
 // DepositBaseDB is the interface that contains all the queries generated
@@ -271,3 +274,27 @@ func sqlcDepositToDeposit(sqlDeposit sqlc.GetAssetDepositsRow,
 		DepositInfo: depositInfo,
 	}, nil
 }
+
+// GetActiveDeposits returns all active deposits from the database. Active
+// deposits are those that have not yet been spent or swept.
+func (s *SQLStore) GetActiveDeposits(ctx context.Context) ([]Deposit, error) {
+	sqlDeposits, err := s.db.GetActiveAssetDeposits(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	deposits := make([]Deposit, 0, len(sqlDeposits))
+	for _, sqlDeposit := range sqlDeposits {
+		deposit, err := sqlcDepositToDeposit(
+			sqlc.GetAssetDepositsRow(sqlDeposit),
+			&s.addressParams,
+		)
+		if err != nil {
+			return nil, err
+		}
+
+		deposits = append(deposits, deposit)
+	}
+
+	return deposits, nil
+}

loopdb/sqlc/asset_deposits.sql.go

@@ -44,3 +44,18 @@ ORDER BY d.created_at ASC;
 UPDATE asset_deposits
 SET sweep_addr = $2
 WHERE deposit_id = $1;
+
+-- name: GetActiveAssetDeposits :many
+SELECT d.*, u.update_state, u.update_timestamp
+FROM asset_deposits d
+JOIN asset_deposit_updates u
+  ON u.deposit_id = d.deposit_id
+WHERE u.id = (
+    SELECT id
+    FROM asset_deposit_updates
+    WHERE deposit_id = d.deposit_id
+    ORDER BY update_timestamp DESC
+    LIMIT 1
+)
+AND u.update_state IN (0, 1, 2, 3, 4, 5, 6);
+