Retry balance checks on failures

We want to use a retry mechanism for single executions of the balance check.
The retries will be performed during the period of `retryTimeout`,
logging a warning on each error. When the timeout is hit an error
will be logged and retries stopped. A next balance check will be triggered at the tick.

Author: nkuba

pkg/chain/celo/celoutil/balance_monitor.go

@@ -2,10 +2,11 @@ package celoutil
 
 import (
 	"context"
+	"time"
+
 	"github.com/celo-org/celo-blockchain/common"
 	"github.com/keep-network/keep-common/pkg/chain/celo"
 	"github.com/keep-network/keep-common/pkg/chain/ethlike"
-	"time"
 )
 
 // BalanceSource provides a balance info for the given address.
@@ -38,16 +39,19 @@ func NewBalanceMonitor(balanceSource BalanceSource) *BalanceMonitor {
 // Observe starts a process which checks the address balance with the given
 // tick and triggers an alert in case the balance falls below the
 // alert threshold value.
+// The balance check will be retried in case of an error up to the retry timeout.
 func (bm *BalanceMonitor) Observe(
 	ctx context.Context,
 	address common.Address,
 	alertThreshold *celo.Wei,
 	tick time.Duration,
+	retryTimeout time.Duration,
 ) {
 	bm.delegate.Observe(
 		ctx,
 		ethlike.Address(address),
 		&alertThreshold.Token,
 		tick,
+		retryTimeout,
 	)
 }

pkg/chain/ethereum/ethutil/balance_monitor.go

@@ -2,10 +2,11 @@ package ethutil
 
 import (
 	"context"
+	"time"
+
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/keep-network/keep-common/pkg/chain/ethereum"
 	"github.com/keep-network/keep-common/pkg/chain/ethlike"
-	"time"
 )
 
 // BalanceSource provides a balance info for the given address.
@@ -38,16 +39,19 @@ func NewBalanceMonitor(balanceSource BalanceSource) *BalanceMonitor {
 // Observe starts a process which checks the address balance with the given
 // tick and triggers an alert in case the balance falls below the
 // alert threshold value.
+// The balance check will be retried in case of an error up to the retry timeout.
 func (bm *BalanceMonitor) Observe(
 	ctx context.Context,
 	address common.Address,
 	alertThreshold *ethereum.Wei,
 	tick time.Duration,
+	retryTimeout time.Duration,
 ) {
 	bm.delegate.Observe(
 		ctx,
 		ethlike.Address(address),
 		&alertThreshold.Token,
 		tick,
+		retryTimeout,
 	)
 }

pkg/chain/ethlike/balance_monitor.go

@@ -2,7 +2,10 @@ package ethlike
 
 import (
 	"context"
+	"fmt"
 	"time"
+
+	"github.com/keep-network/keep-common/pkg/wrappers"
 )
 
 // BalanceSource provides a balance info for the given address.
@@ -22,17 +25,26 @@ func NewBalanceMonitor(balanceSource BalanceSource) *BalanceMonitor {
 // Observe starts a process which checks the address balance with the given
 // tick and triggers an alert in case the balance falls below the
 // alert threshold value.
+// The balance check will be retried in case of an error up to the retry timeout.
 func (bm *BalanceMonitor) Observe(
 	ctx context.Context,
 	address Address,
 	alertThreshold *Token,
 	tick time.Duration,
+	retryTimeout time.Duration,
 ) {
-	check := func() {
+	check := func(ctx context.Context) error {
 		balance, err := bm.balanceSource(address)
 		if err != nil {
-			logger.Errorf("balance monitor error: [%v]", err)
-			return
+			wrappedErr := fmt.Errorf(
+				"failed to get balance for account [%s]: [%w]",
+				address.TerminalString(),
+				err,
+			)
+
+			logger.Warning(wrappedErr)
+
+			return wrappedErr
 		}
 
 		if balance.Cmp(alertThreshold.Int) == -1 {
@@ -43,6 +55,8 @@ func (bm *BalanceMonitor) Observe(
 				alertThreshold.Text(10),
 			)
 		}
+
+		return nil
 	}
 
 	go func() {
@@ -52,7 +66,9 @@ func (bm *BalanceMonitor) Observe(
 		for {
 			select {
 			case <-ticker.C:
-				check()
+				err := wrappers.DoWithDefaultRetry(retryTimeout, check)
+
+				logger.Errorf("balance monitor error: [%v]", err)
 			case <-ctx.Done():
 				return
 			}