feat: add lock retry strategy

* previouly we relied on the task being re-queued which generally reduces the throughput of tasks

internal/tasker/task/account_gift_gas.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/bsm/redislock"
 	"github.com/celo-org/celo-blockchain/common/hexutil"
 	"github.com/grassrootseconomics/celoutils"
 	"github.com/grassrootseconomics/cic-custodial/internal/custodial"
@@ -36,7 +37,9 @@ func AccountGiftGasProcessor(cu *custodial.Custodial) func(context.Context, *asy
 			ctx,
 			lockPrefix+cu.SystemContainer.PublicKey,
 			cu.SystemContainer.LockTimeout,
-			nil,
+			&redislock.Options{
+				RetryStrategy: lockRetry(),
+			},
 		)
 		if err != nil {
 			return err

internal/tasker/task/account_gift_voucher.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 
+	"github.com/bsm/redislock"
 	"github.com/celo-org/celo-blockchain/common/hexutil"
 	"github.com/grassrootseconomics/celoutils"
 	"github.com/grassrootseconomics/cic-custodial/internal/custodial"
@@ -30,7 +31,9 @@ func GiftVoucherProcessor(cu *custodial.Custodial) func(context.Context, *asynq.
 			ctx,
 			lockPrefix+cu.SystemContainer.PublicKey,
 			cu.SystemContainer.LockTimeout,
-			nil,
+			&redislock.Options{
+				RetryStrategy: lockRetry(),
+			},
 		)
 		if err != nil {
 			return err

internal/tasker/task/account_refill_gas.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/bsm/redislock"
 	"github.com/celo-org/celo-blockchain/common/hexutil"
 	"github.com/grassrootseconomics/celoutils"
 	"github.com/grassrootseconomics/cic-custodial/internal/custodial"
@@ -55,7 +56,9 @@ func AccountRefillGasProcessor(cu *custodial.Custodial) func(context.Context, *a
 			ctx,
 			lockPrefix+cu.SystemContainer.PublicKey,
 			cu.SystemContainer.LockTimeout,
-			nil,
+			&redislock.Options{
+				RetryStrategy: lockRetry(),
+			},
 		)
 		if err != nil {
 			return err

internal/tasker/task/account_register_onchain.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 
+	"github.com/bsm/redislock"
 	"github.com/celo-org/celo-blockchain/common/hexutil"
 	"github.com/grassrootseconomics/celoutils"
 	"github.com/grassrootseconomics/cic-custodial/internal/custodial"
@@ -31,7 +32,9 @@ func AccountRegisterOnChainProcessor(cu *custodial.Custodial) func(context.Conte
 			ctx,
 			lockPrefix+cu.SystemContainer.PublicKey,
 			cu.SystemContainer.LockTimeout,
-			nil,
+			&redislock.Options{
+				RetryStrategy: lockRetry(),
+			},
 		)
 		if err != nil {
 			return err

internal/tasker/task/sign_transfer.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"math/big"
 
+	"github.com/bsm/redislock"
 	"github.com/celo-org/celo-blockchain/common/hexutil"
 	"github.com/grassrootseconomics/celoutils"
 	"github.com/grassrootseconomics/cic-custodial/internal/custodial"
@@ -49,7 +50,9 @@ func SignTransfer(cu *custodial.Custodial) func(context.Context, *asynq.Task) er
 			ctx,
 			lockPrefix+payload.From,
 			cu.SystemContainer.LockTimeout,
-			nil,
+			&redislock.Options{
+				RetryStrategy: lockRetry(),
+			},
 		)
 		if err != nil {
 			return err

internal/tasker/task/utils.go

@@ -1,5 +1,21 @@
 package task
 
+import (
+	"time"
+
+	"github.com/bsm/redislock"
+)
+
 const (
 	lockPrefix     = "lock:"
+	lockRetryDelay = 25 * time.Millisecond
 )
+
+// lockRetry will at most try to obtain the lock 20 times within ~0.5s.
+// it is expected to prevent immidiate requeue of the task at the expense of more redis calls.
+func lockRetry() redislock.RetryStrategy {
+	return redislock.LimitRetry(
+		redislock.LinearBackoff(lockRetryDelay),
+		20,
+	)
+}