refactor mercury (#11137)

* refactor mercury, rebase on top of PR 10878

* change back to use eth_call, sigh, simulated backend in integration test doesnt like using the registry contract directly

* fixes and include new threadctl changes

* address comments, add time in logs

* rebase and add a test

core/scripts/chaincli/handler/debug.go

@@ -18,16 +18,18 @@ import (
 	gethcommon "github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/ethclient"
+	ocr2keepers "github.com/smartcontractkit/ocr2keepers/pkg/v3/types"
+
 	"github.com/smartcontractkit/chainlink/core/scripts/chaincli/config"
 	"github.com/smartcontractkit/chainlink/core/scripts/common"
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/generated/automation_utils_2_1"
 	iregistry21 "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/generated/i_keeper_registry_master_wrapper_2_1"
-	evm "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ocr2keeper/evm21"
 	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ocr2keeper/evm21/core"
 	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ocr2keeper/evm21/encoding"
+	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ocr2keeper/evm21/mercury/streams"
 	"github.com/smartcontractkit/chainlink/v2/core/utils"
 	bigmath "github.com/smartcontractkit/chainlink/v2/core/utils/big_math"
-	ocr2keepers "github.com/smartcontractkit/ocr2keepers/pkg/v3/types"
 )
 
 const (
@@ -227,6 +229,13 @@ func (k *Keeper) Debug(ctx context.Context, args []string) {
 		message(fmt.Sprintf("checkUpkeep failed with UpkeepFailureReason %d", checkResult.UpkeepFailureReason))
 	}
 	if checkResult.UpkeepFailureReason == uint8(encoding.UpkeepFailureReasonTargetCheckReverted) {
+		// TODO use the new streams lookup lib
+		//mc := &models.MercuryCredentials{k.cfg.MercuryLegacyURL, k.cfg.MercuryURL, k.cfg.MercuryID, k.cfg.MercuryKey}
+		//mercuryConfig := evm.NewMercuryConfig(mc, core.StreamsCompatibleABI)
+		//lggr, _ := logger.NewLogger()
+		//blockSub := &blockSubscriber{k.client}
+		//_ = streams.NewStreamsLookup(packer, mercuryConfig, blockSub, keeperRegistry21, k.rpcClient, lggr)
+
 		streamsLookupErr, err := packer.DecodeStreamsLookupRequest(checkResult.PerformData)
 		if err == nil {
 			message("upkeep reverted with StreamsLookup")
@@ -240,7 +249,7 @@ func (k *Keeper) Debug(ctx context.Context, args []string) {
 				}
 				allowed := false
 				if len(cfg) > 0 {
-					var privilegeConfig evm.UpkeepPrivilegeConfig
+					var privilegeConfig streams.UpkeepPrivilegeConfig
 					if err := json.Unmarshal(cfg, &privilegeConfig); err != nil {
 						failUnknown("failed to unmarshal privilege config ", err)
 					}
@@ -307,6 +316,22 @@ func (k *Keeper) Debug(ctx context.Context, args []string) {
 	}
 }
 
+type blockSubscriber struct {
+	ethClient *ethclient.Client
+}
+
+func (bs *blockSubscriber) LatestBlock() *ocr2keepers.BlockKey {
+	header, err := bs.ethClient.HeaderByNumber(context.Background(), nil)
+	if err != nil {
+		return nil
+	}
+
+	return &ocr2keepers.BlockKey{
+		Number: ocr2keepers.BlockNumber(header.Number.Uint64()),
+		Hash:   header.Hash(),
+	}
+}
+
 func logMatchesTriggerConfig(log *types.Log, config automation_utils_2_1.LogTriggerConfig) bool {
 	if log.Topics[0] != config.Topic0 {
 		return false

core/scripts/chaincli/handler/keeper.go

@@ -14,6 +14,8 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/umbracle/ethgo/abi"
 
+	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ocr2keeper/evm21/mercury/streams"
+
 	"github.com/smartcontractkit/chainlink/core/scripts/chaincli/config"
 	helpers "github.com/smartcontractkit/chainlink/core/scripts/common"
 	"github.com/smartcontractkit/chainlink/v2/core/cmd"
@@ -34,7 +36,6 @@ import (
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/generated/verifiable_load_upkeep_wrapper"
 	"github.com/smartcontractkit/chainlink/v2/core/logger"
 	"github.com/smartcontractkit/chainlink/v2/core/services/keeper"
-	evm "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ocr2keeper/evm21"
 )
 
 // Keeper is the keepers commands handler
@@ -719,7 +720,7 @@ func (k *Keeper) deployUpkeeps(ctx context.Context, registryAddr common.Address,
 		log.Printf("registry version is %s", v)
 		log.Printf("active upkeep ids: %v", activeUpkeepIds)
 
-		adminBytes, err := json.Marshal(evm.UpkeepPrivilegeConfig{
+		adminBytes, err := json.Marshal(streams.UpkeepPrivilegeConfig{
 			MercuryEnabled: true,
 		})
 		if err != nil {

core/scripts/chaincli/handler/mercury_lookup_handler.go

@@ -3,7 +3,9 @@ package handler
 import (
 	"context"
 	"crypto/hmac"
+	"crypto/sha256"
 	"encoding/hex"
+	"encoding/json"
 	"fmt"
 	"io"
 	"math/big"
@@ -13,10 +15,6 @@ import (
 	"strings"
 	"time"
 
-	"crypto/sha256"
-
-	"encoding/json"
-
 	"github.com/avast/retry-go"
 	ethabi "github.com/ethereum/go-ethereum/accounts/abi"
 	"github.com/ethereum/go-ethereum/common"

core/services/ocr2/plugins/ocr2keeper/evm21/block_subscriber.go

@@ -12,6 +12,7 @@ import (
 	ocr2keepers "github.com/smartcontractkit/ocr2keepers/pkg/v3/types"
 
 	"github.com/smartcontractkit/chainlink-relay/pkg/services"
+
 	httypes "github.com/smartcontractkit/chainlink/v2/core/chains/evm/headtracker/types"
 	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller"
 	evmtypes "github.com/smartcontractkit/chainlink/v2/core/chains/evm/types"
@@ -56,6 +57,10 @@ type BlockSubscriber struct {
 	lggr             logger.Logger
 }
 
+func (bs *BlockSubscriber) LatestBlock() *ocr2keepers.BlockKey {
+	return bs.latestBlock.Load()
+}
+
 var _ ocr2keepers.BlockSubscriber = &BlockSubscriber{}
 
 func NewBlockSubscriber(hb httypes.HeadBroadcaster, lp logpoller.LogPoller, finalityDepth uint32, lggr logger.Logger) *BlockSubscriber {

core/services/ocr2/plugins/ocr2keeper/evm21/encoding/interface.go

@@ -5,56 +5,55 @@ import (
 
 	ocr2keepers "github.com/smartcontractkit/ocr2keepers/pkg/v3/types"
 
+	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ocr2keeper/evm21/mercury"
+
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/generated/automation_utils_2_1"
 	iregistry21 "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/generated/i_keeper_registry_master_wrapper_2_1"
 )
 
-type UpkeepFailureReason uint8
-type PipelineExecutionState uint8
-
 const (
 	// upkeep failure onchain reasons
-	UpkeepFailureReasonNone                    UpkeepFailureReason = 0
-	UpkeepFailureReasonUpkeepCancelled         UpkeepFailureReason = 1
-	UpkeepFailureReasonUpkeepPaused            UpkeepFailureReason = 2
-	UpkeepFailureReasonTargetCheckReverted     UpkeepFailureReason = 3
-	UpkeepFailureReasonUpkeepNotNeeded         UpkeepFailureReason = 4
-	UpkeepFailureReasonPerformDataExceedsLimit UpkeepFailureReason = 5
-	UpkeepFailureReasonInsufficientBalance     UpkeepFailureReason = 6
-	UpkeepFailureReasonMercuryCallbackReverted UpkeepFailureReason = 7
-	UpkeepFailureReasonRevertDataExceedsLimit  UpkeepFailureReason = 8
-	UpkeepFailureReasonRegistryPaused          UpkeepFailureReason = 9
+	UpkeepFailureReasonNone                    uint8 = 0
+	UpkeepFailureReasonUpkeepCancelled         uint8 = 1
+	UpkeepFailureReasonUpkeepPaused            uint8 = 2
+	UpkeepFailureReasonTargetCheckReverted     uint8 = 3
+	UpkeepFailureReasonUpkeepNotNeeded         uint8 = 4
+	UpkeepFailureReasonPerformDataExceedsLimit uint8 = 5
+	UpkeepFailureReasonInsufficientBalance     uint8 = 6
+	UpkeepFailureReasonMercuryCallbackReverted uint8 = 7
+	UpkeepFailureReasonRevertDataExceedsLimit  uint8 = 8
+	UpkeepFailureReasonRegistryPaused          uint8 = 9
 	// leaving a gap here for more onchain failure reasons in the future
 	// upkeep failure offchain reasons
-	UpkeepFailureReasonMercuryAccessNotAllowed UpkeepFailureReason = 32
-	UpkeepFailureReasonTxHashNoLongerExists    UpkeepFailureReason = 33
-	UpkeepFailureReasonInvalidRevertDataInput  UpkeepFailureReason = 34
-	UpkeepFailureReasonSimulationFailed        UpkeepFailureReason = 35
-	UpkeepFailureReasonTxHashReorged           UpkeepFailureReason = 36
+	UpkeepFailureReasonMercuryAccessNotAllowed uint8 = 32
+	UpkeepFailureReasonTxHashNoLongerExists    uint8 = 33
+	UpkeepFailureReasonInvalidRevertDataInput  uint8 = 34
+	UpkeepFailureReasonSimulationFailed        uint8 = 35
+	UpkeepFailureReasonTxHashReorged           uint8 = 36
 
 	// pipeline execution error
-	NoPipelineError        PipelineExecutionState = 0
-	CheckBlockTooOld       PipelineExecutionState = 1
-	CheckBlockInvalid      PipelineExecutionState = 2
-	RpcFlakyFailure        PipelineExecutionState = 3
-	MercuryFlakyFailure    PipelineExecutionState = 4
-	PackUnpackDecodeFailed PipelineExecutionState = 5
-	MercuryUnmarshalError  PipelineExecutionState = 6
-	InvalidMercuryRequest  PipelineExecutionState = 7
-	InvalidMercuryResponse PipelineExecutionState = 8 // this will only happen if Mercury server sends bad responses
-	UpkeepNotAuthorized    PipelineExecutionState = 9
+	NoPipelineError        uint8 = 0
+	CheckBlockTooOld       uint8 = 1
+	CheckBlockInvalid      uint8 = 2
+	RpcFlakyFailure        uint8 = 3
+	MercuryFlakyFailure    uint8 = 4
+	PackUnpackDecodeFailed uint8 = 5
+	MercuryUnmarshalError  uint8 = 6
+	InvalidMercuryRequest  uint8 = 7
+	InvalidMercuryResponse uint8 = 8 // this will only happen if Mercury server sends bad responses
+	UpkeepNotAuthorized    uint8 = 9
 )
 
 type UpkeepInfo = iregistry21.KeeperRegistryBase21UpkeepInfo
 
 type Packer interface {
 	UnpackCheckResult(payload ocr2keepers.UpkeepPayload, raw string) (ocr2keepers.CheckResult, error)
-	UnpackCheckCallbackResult(callbackResp []byte) (PipelineExecutionState, bool, []byte, uint8, *big.Int, error)
-	UnpackPerformResult(raw string) (PipelineExecutionState, bool, error)
+	UnpackCheckCallbackResult(callbackResp []byte) (uint8, bool, []byte, uint8, *big.Int, error)
+	UnpackPerformResult(raw string) (uint8, bool, error)
 	UnpackLogTriggerConfig(raw []byte) (automation_utils_2_1.LogTriggerConfig, error)
 	PackReport(report automation_utils_2_1.KeeperRegistryBase21Report) ([]byte, error)
 	UnpackReport(raw []byte) (automation_utils_2_1.KeeperRegistryBase21Report, error)
 	PackGetUpkeepPrivilegeConfig(upkeepId *big.Int) ([]byte, error)
 	UnpackGetUpkeepPrivilegeConfig(resp []byte) ([]byte, error)
-	DecodeStreamsLookupRequest(data []byte) (*StreamsLookupError, error)
+	DecodeStreamsLookupRequest(data []byte) (*mercury.StreamsLookupError, error)
 }

core/services/ocr2/plugins/ocr2keeper/evm21/encoding/packer.go

@@ -6,8 +6,11 @@ import (
 
 	"github.com/ethereum/go-ethereum/accounts/abi"
 	"github.com/ethereum/go-ethereum/common/hexutil"
+
 	ocr2keepers "github.com/smartcontractkit/ocr2keepers/pkg/v3/types"
 
+	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ocr2keeper/evm21/mercury"
+
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/generated/automation_utils_2_1"
 	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ocr2keeper/evm21/core"
 )
@@ -80,7 +83,7 @@ func (p *abiPacker) UnpackGetUpkeepPrivilegeConfig(resp []byte) ([]byte, error)
 	return bts, nil
 }
 
-func (p *abiPacker) UnpackCheckCallbackResult(callbackResp []byte) (PipelineExecutionState, bool, []byte, uint8, *big.Int, error) {
+func (p *abiPacker) UnpackCheckCallbackResult(callbackResp []byte) (uint8, bool, []byte, uint8, *big.Int, error) {
 	out, err := p.registryABI.Methods["checkCallback"].Outputs.UnpackValues(callbackResp)
 	if err != nil {
 		return PackUnpackDecodeFailed, false, nil, 0, nil, fmt.Errorf("%w: unpack checkUpkeep return: %s", err, hexutil.Encode(callbackResp))
@@ -94,7 +97,7 @@ func (p *abiPacker) UnpackCheckCallbackResult(callbackResp []byte) (PipelineExec
 	return NoPipelineError, upkeepNeeded, rawPerformData, failureReason, gasUsed, nil
 }
 
-func (p *abiPacker) UnpackPerformResult(raw string) (PipelineExecutionState, bool, error) {
+func (p *abiPacker) UnpackPerformResult(raw string) (uint8, bool, error) {
 	b, err := hexutil.Decode(raw)
 	if err != nil {
 		return PackUnpackDecodeFailed, false, err
@@ -161,24 +164,16 @@ func (p *abiPacker) UnpackReport(raw []byte) (automation_utils_2_1.KeeperRegistr
 	return report, nil
 }
 
-type StreamsLookupError struct {
-	FeedParamKey string
-	Feeds        []string
-	TimeParamKey string
-	Time         *big.Int
-	ExtraData    []byte
-}
-
 // DecodeStreamsLookupRequest decodes the revert error StreamsLookup(string feedParamKey, string[] feeds, string feedParamKey, uint256 time, byte[] extraData)
-func (p *abiPacker) DecodeStreamsLookupRequest(data []byte) (*StreamsLookupError, error) {
+func (p *abiPacker) DecodeStreamsLookupRequest(data []byte) (*mercury.StreamsLookupError, error) {
 	e := p.streamsABI.Errors["StreamsLookup"]
 	unpack, err := e.Unpack(data)
 	if err != nil {
 		return nil, fmt.Errorf("unpack error: %w", err)
 	}
 	errorParameters := unpack.([]interface{})
 
-	return &StreamsLookupError{
+	return &mercury.StreamsLookupError{
 		FeedParamKey: *abi.ConvertType(errorParameters[0], new(string)).(*string),
 		Feeds:        *abi.ConvertType(errorParameters[1], new([]string)).(*[]string),
 		TimeParamKey: *abi.ConvertType(errorParameters[2], new(string)).(*string),
@@ -188,10 +183,10 @@ func (p *abiPacker) DecodeStreamsLookupRequest(data []byte) (*StreamsLookupError
 }
 
 // GetIneligibleCheckResultWithoutPerformData returns an ineligible check result with ineligibility reason and pipeline execution state but without perform data
-func GetIneligibleCheckResultWithoutPerformData(p ocr2keepers.UpkeepPayload, reason UpkeepFailureReason, state PipelineExecutionState, retryable bool) ocr2keepers.CheckResult {
+func GetIneligibleCheckResultWithoutPerformData(p ocr2keepers.UpkeepPayload, reason uint8, state uint8, retryable bool) ocr2keepers.CheckResult {
 	return ocr2keepers.CheckResult{
-		IneligibilityReason:    uint8(reason),
-		PipelineExecutionState: uint8(state),
+		IneligibilityReason:    reason,
+		PipelineExecutionState: state,
 		Retryable:              retryable,
 		UpkeepID:               p.UpkeepID,
 		Trigger:                p.Trigger,

core/services/ocr2/plugins/ocr2keeper/evm21/encoding/packer_test.go

@@ -13,6 +13,8 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ocr2keeper/evm21/mercury"
+
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/generated/automation_utils_2_1"
 	automation21Utils "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/generated/automation_utils_2_1"
 	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/ocr2keeper/evm21/core"
@@ -210,7 +212,7 @@ func TestPacker_UnpackPerformResult(t *testing.T) {
 	tests := []struct {
 		Name    string
 		RawData string
-		State   PipelineExecutionState
+		State   uint8
 	}{
 		{
 			Name:    "unpack success",
@@ -238,7 +240,7 @@ func TestPacker_UnpackCheckCallbackResult(t *testing.T) {
 		FailureReason uint8
 		GasUsed       *big.Int
 		ErrorString   string
-		State         PipelineExecutionState
+		State         uint8
 	}{
 		{
 			Name:          "unpack upkeep needed",
@@ -441,14 +443,14 @@ func TestPacker_DecodeStreamsLookupRequest(t *testing.T) {
 	tests := []struct {
 		name     string
 		data     []byte
-		expected *StreamsLookupError
-		state    PipelineExecutionState
+		expected *mercury.StreamsLookupError
+		state    uint8
 		err      error
 	}{
 		{
 			name: "success - decode to streams lookup",
 			data: hexutil.MustDecode("0xf055e4a200000000000000000000000000000000000000000000000000000000000000a000000000000000000000000000000000000000000000000000000000000000e000000000000000000000000000000000000000000000000000000000000002400000000000000000000000000000000000000000000000000000000002435eb50000000000000000000000000000000000000000000000000000000000000280000000000000000000000000000000000000000000000000000000000000000966656564496448657800000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000002000000000000000000000000000000000000000000000000000000000000004000000000000000000000000000000000000000000000000000000000000000c000000000000000000000000000000000000000000000000000000000000000423078343535343438326435353533343432643431353234323439353435323535346432643534343535333534346534353534303030303030303030303030303030300000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000042307834323534343332643535353334343264343135323432343935343532353534643264353434353533353434653435353430303030303030303030303030303030000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000b626c6f636b4e756d62657200000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000140000000000000000000000000000000000000064000000000000000000000000"),
-			expected: &StreamsLookupError{
+			expected: &mercury.StreamsLookupError{
 				FeedParamKey: "feedIdHex",
 				Feeds:        []string{"0x4554482d5553442d415242495452554d2d544553544e45540000000000000000", "0x4254432d5553442d415242495452554d2d544553544e45540000000000000000"},
 				TimeParamKey: "blockNumber",