Stream fallback: error handler

contracts/src/v0.8/automation/dev/MercuryRegistry.sol

@@ -144,6 +144,14 @@ contract MercuryRegistry is ConfirmedOwner, AutomationCompatibleInterface, Strea
     return (filteredValues.length > 0, performData);
   }
 
+  function checkErrorHandler(
+    uint256 errCode,
+    bytes memory extraData
+  ) external view override returns (bool upkeepNeeded, bytes memory performData) {
+    // dummy function with default values
+    return (false, new bytes(0));
+  }
+
   // Use deviated off-chain values to update on-chain state.
   function performUpkeep(bytes calldata performData) external override {
     (bytes[] memory values /* bytes memory lookupData */, ) = abi.decode(performData, (bytes[], bytes));

contracts/src/v0.8/automation/dev/MercuryRegistryBatchUpkeep.sol

@@ -61,6 +61,14 @@ contract MercuryRegistryBatchUpkeep is ConfirmedOwner, AutomationCompatibleInter
     return i_registry.checkCallback(values, lookupData);
   }
 
+  function checkErrorHandler(
+    uint256 errCode,
+    bytes memory extraData
+  ) external view override returns (bool upkeepNeeded, bytes memory performData) {
+    // dummy function with default values
+    return (false, new bytes(0));
+  }
+
   // Use the master registry to update state.
   function performUpkeep(bytes calldata performData) external override {
     i_registry.performUpkeep(performData);

contracts/src/v0.8/automation/interfaces/StreamsLookupCompatibleInterface.sol

@@ -17,4 +17,17 @@ interface StreamsLookupCompatibleInterface {
     bytes[] memory values,
     bytes memory extraData
   ) external view returns (bool upkeepNeeded, bytes memory performData);
+
+  /**
+   * @notice this is a new, optional function in streams lookup. It is meant to surface streams lookup errors.
+   * @param errCode an uint value that represents the streams lookup error code.
+   * @param extraData context data from streams lookup process.
+   * @return upkeepNeeded boolean to indicate whether the keeper should call performUpkeep or not.
+   * @return performData bytes that the keeper should call performUpkeep with, if
+   * upkeep is needed. If you would like to encode data to decode later, try `abi.encode`.
+   */
+  function checkErrorHandler(
+    uint256 errCode,
+    bytes memory extraData
+  ) external view returns (bool upkeepNeeded, bytes memory performData);
 }

contracts/src/v0.8/automation/testhelpers/LogTriggeredStreamsLookup.sol

@@ -130,6 +130,14 @@ contract LogTriggeredStreamsLookup is ILogAutomation, StreamsLookupCompatibleInt
     return (true, performData);
   }
 
+  function checkErrorHandler(
+    uint256 errCode,
+    bytes memory extraData
+  ) external view override returns (bool upkeepNeeded, bytes memory performData) {
+    // dummy function with default values
+    return (false, new bytes(0));
+  }
+
   function getBlockNumber() internal view returns (uint256) {
     if (useArbitrumBlockNum) {
       return ARB_SYS.arbBlockNumber();

contracts/src/v0.8/tests/StreamsLookupUpkeep.sol

@@ -88,6 +88,14 @@ contract StreamsLookupUpkeep is AutomationCompatibleInterface, StreamsLookupComp
     return (callbackReturnBool, performData);
   }
 
+  function checkErrorHandler(
+    uint256 errCode,
+    bytes memory extraData
+  ) external view override returns (bool upkeepNeeded, bytes memory performData) {
+    // dummy function with default values
+    return (false, new bytes(0));
+  }
+
   function checkUpkeep(bytes calldata data) external view returns (bool, bytes memory) {
     if (!eligible()) {
       return (false, data);

contracts/src/v0.8/tests/VerifiableLoadLogTriggerUpkeep.sol

@@ -116,4 +116,12 @@ contract VerifiableLoadLogTriggerUpkeep is VerifiableLoadBase, StreamsLookupComp
     bytes memory performData = abi.encode(values, extraData);
     return (true, performData);
   }
+
+  function checkErrorHandler(
+    uint256 errCode,
+    bytes memory extraData
+  ) external view override returns (bool upkeepNeeded, bytes memory performData) {
+    // dummy function with default values
+    return (false, new bytes(0));
+  }
 }

contracts/src/v0.8/tests/VerifiableLoadStreamsLookupUpkeep.sol

@@ -16,6 +16,14 @@ contract VerifiableLoadStreamsLookupUpkeep is VerifiableLoadBase, StreamsLookupC
     return (true, performData);
   }
 
+  function checkErrorHandler(
+    uint256 errCode,
+    bytes memory extraData
+  ) external view override returns (bool upkeepNeeded, bytes memory performData) {
+    // dummy function with default values
+    return (false, new bytes(0));
+  }
+
   function checkUpkeep(bytes calldata checkData) external returns (bool, bytes memory) {
     uint256 startGas = gasleft();
     uint256 upkeepId = abi.decode(checkData, (uint256));

core/gethwrappers/generation/generated-wrapper-dependency-versions-do-not-edit.txt

@@ -46,7 +46,7 @@ keeper_registry_wrapper_2_1: ../../contracts/solc/v0.8.16/KeeperRegistry2_1/Keep
 keeper_registry_wrapper_2_2: ../../contracts/solc/v0.8.16/AutomationRegistry2_2/AutomationRegistry2_2.abi ../../contracts/solc/v0.8.16/AutomationRegistry2_2/AutomationRegistry2_2.bin 8f1c6d3eddf36201a556a101350abd6eb733d4c3177da2ddb5eca0d24d9a88ef
 keepers_vrf_consumer: ../../contracts/solc/v0.8.6/KeepersVRFConsumer/KeepersVRFConsumer.abi ../../contracts/solc/v0.8.6/KeepersVRFConsumer/KeepersVRFConsumer.bin fa75572e689c9e84705c63e8dbe1b7b8aa1a8fe82d66356c4873d024bb9166e8
 log_emitter: ../../contracts/solc/v0.8.19/LogEmitter/LogEmitter.abi ../../contracts/solc/v0.8.19/LogEmitter/LogEmitter.bin 4b129ab93432c95ff9143f0631323e189887668889e0b36ccccf18a571e41ccf
-log_triggered_streams_lookup_wrapper: ../../contracts/solc/v0.8.16/LogTriggeredStreamsLookup/LogTriggeredStreamsLookup.abi ../../contracts/solc/v0.8.16/LogTriggeredStreamsLookup/LogTriggeredStreamsLookup.bin f8da43a927c1a66238a9f4fd5d5dd7e280e361daa0444da1f7f79498ace901e1
+log_triggered_streams_lookup_wrapper: ../../contracts/solc/v0.8.16/LogTriggeredStreamsLookup/LogTriggeredStreamsLookup.abi ../../contracts/solc/v0.8.16/LogTriggeredStreamsLookup/LogTriggeredStreamsLookup.bin bb5f9f72c698fb2a79667cbf9206fa3c848aa4cb4866c5fa8a541627f0af48ec
 log_upkeep_counter_wrapper: ../../contracts/solc/v0.8.6/LogUpkeepCounter/LogUpkeepCounter.abi ../../contracts/solc/v0.8.6/LogUpkeepCounter/LogUpkeepCounter.bin 42426bbb83f96dfbe55fc576d6c65020eaeed690e2289cf99b0c4aa810a5f4ec
 mock_aggregator_proxy: ../../contracts/solc/v0.8.6/MockAggregatorProxy/MockAggregatorProxy.abi ../../contracts/solc/v0.8.6/MockAggregatorProxy/MockAggregatorProxy.bin b16c108f3dd384c342ddff5e94da7c0a8d39d1be5e3d8f2cf61ecc7f0e50ff42
 mock_ethlink_aggregator_wrapper: ../../contracts/solc/v0.6/MockETHLINKAggregator/MockETHLINKAggregator.abi ../../contracts/solc/v0.6/MockETHLINKAggregator/MockETHLINKAggregator.bin 1c52c24f797b8482aa12b8251dcea1c072827bd5b3426b822621261944b99ca0
@@ -66,16 +66,16 @@ solidity_vrf_request_id_v08: ../../contracts/solc/v0.8.6/VRFRequestIDBaseTestHel
 solidity_vrf_v08_verifier_wrapper: ../../contracts/solc/v0.8.6/VRFTestHelper/VRFTestHelper.abi ../../contracts/solc/v0.8.6/VRFTestHelper/VRFTestHelper.bin f37f8b21a81c113085c6137835a2246db6ebda07da455c4f2b5c7ec60c725c3b
 solidity_vrf_verifier_wrapper: ../../contracts/solc/v0.6/VRFTestHelper/VRFTestHelper.abi ../../contracts/solc/v0.6/VRFTestHelper/VRFTestHelper.bin 44c2b67d8d2990ab580453deb29d63508c6147a3dc49908a1db563bef06e6474
 solidity_vrf_wrapper: ../../contracts/solc/v0.6/VRF/VRF.abi ../../contracts/solc/v0.6/VRF/VRF.bin 04ede5b83c06ba5b76ef99c081c72928007d8a7aaefcf21449a46a07cbd4bfc2
-streams_lookup_compatible_interface: ../../contracts/solc/v0.8.16/StreamsLookupCompatibleInterface/StreamsLookupCompatibleInterface.abi ../../contracts/solc/v0.8.16/StreamsLookupCompatibleInterface/StreamsLookupCompatibleInterface.bin feb92cc666df21ea04ab9d7a588a513847b01b2f66fc167d06ab28ef2b17e015
-streams_lookup_upkeep_wrapper: ../../contracts/solc/v0.8.16/StreamsLookupUpkeep/StreamsLookupUpkeep.abi ../../contracts/solc/v0.8.16/StreamsLookupUpkeep/StreamsLookupUpkeep.bin b1a598963cacac51ed4706538d0f142bdc0d94b9a4b13e2d402131cdf05c9bcf
+streams_lookup_compatible_interface: ../../contracts/solc/v0.8.16/StreamsLookupCompatibleInterface/StreamsLookupCompatibleInterface.abi ../../contracts/solc/v0.8.16/StreamsLookupCompatibleInterface/StreamsLookupCompatibleInterface.bin 2861f553fb4731e89126b13319462df674727005a51982d1e617e2c2e44fa422
+streams_lookup_upkeep_wrapper: ../../contracts/solc/v0.8.16/StreamsLookupUpkeep/StreamsLookupUpkeep.abi ../../contracts/solc/v0.8.16/StreamsLookupUpkeep/StreamsLookupUpkeep.bin 37e3a61091cc2a156539dd4aaff987e07577118aa02e97931a647df55705465e
 test_api_consumer_wrapper: ../../contracts/solc/v0.6/TestAPIConsumer/TestAPIConsumer.abi ../../contracts/solc/v0.6/TestAPIConsumer/TestAPIConsumer.bin ed10893cb18894c18e275302329c955f14ea2de37ee044f84aa1e067ac5ea71e
 trusted_blockhash_store: ../../contracts/solc/v0.8.6/TrustedBlockhashStore/TrustedBlockhashStore.abi ../../contracts/solc/v0.8.6/TrustedBlockhashStore/TrustedBlockhashStore.bin 98cb0dc06c15af5dcd3b53bdfc98e7ed2489edc96a42203294ac2fc0efdda02b
 type_and_version_interface_wrapper: ../../contracts/solc/v0.8.6/KeeperRegistry1_2/TypeAndVersionInterface.abi ../../contracts/solc/v0.8.6/KeeperRegistry1_2/TypeAndVersionInterface.bin bc9c3a6e73e3ebd5b58754df0deeb3b33f4bb404d5709bb904aed51d32f4b45e
 upkeep_counter_wrapper: ../../contracts/solc/v0.8.16/UpkeepCounter/UpkeepCounter.abi ../../contracts/solc/v0.8.16/UpkeepCounter/UpkeepCounter.bin 77f000229a501f638dd2dc439859257f632894c728b31e68aea4f6d6c52f1b71
 upkeep_perform_counter_restrictive_wrapper: ../../contracts/solc/v0.8.16/UpkeepPerformCounterRestrictive/UpkeepPerformCounterRestrictive.abi ../../contracts/solc/v0.8.16/UpkeepPerformCounterRestrictive/UpkeepPerformCounterRestrictive.bin 20955b21acceb58355fa287b29194a73edf5937067ba7140667301017cb2b24c
 upkeep_transcoder: ../../contracts/solc/v0.8.6/UpkeepTranscoder/UpkeepTranscoder.abi ../../contracts/solc/v0.8.6/UpkeepTranscoder/UpkeepTranscoder.bin 336c92a981597be26508455f81a908a0784a817b129a59686c5b2c4afcba730a
-verifiable_load_log_trigger_upkeep_wrapper: ../../contracts/solc/v0.8.16/VerifiableLoadLogTriggerUpkeep/VerifiableLoadLogTriggerUpkeep.abi ../../contracts/solc/v0.8.16/VerifiableLoadLogTriggerUpkeep/VerifiableLoadLogTriggerUpkeep.bin fb674ba44c0e8f3b385cd10b2f7dea5cd07b5f38df08066747e8b1542e152557
-verifiable_load_streams_lookup_upkeep_wrapper: ../../contracts/solc/v0.8.16/VerifiableLoadStreamsLookupUpkeep/VerifiableLoadStreamsLookupUpkeep.abi ../../contracts/solc/v0.8.16/VerifiableLoadStreamsLookupUpkeep/VerifiableLoadStreamsLookupUpkeep.bin 785f68c44bfff070505eaa65e38a1af94046e5f9afc1189bcf2c8cfcd1102d66
+verifiable_load_log_trigger_upkeep_wrapper: ../../contracts/solc/v0.8.16/VerifiableLoadLogTriggerUpkeep/VerifiableLoadLogTriggerUpkeep.abi ../../contracts/solc/v0.8.16/VerifiableLoadLogTriggerUpkeep/VerifiableLoadLogTriggerUpkeep.bin ee5c608e4e84c80934e42b0c02a49624840adf10b50c91f688bf8f0c7c6994c2
+verifiable_load_streams_lookup_upkeep_wrapper: ../../contracts/solc/v0.8.16/VerifiableLoadStreamsLookupUpkeep/VerifiableLoadStreamsLookupUpkeep.abi ../../contracts/solc/v0.8.16/VerifiableLoadStreamsLookupUpkeep/VerifiableLoadStreamsLookupUpkeep.bin 58f1f6b31a313e04ceb3e0e0f0393bc195cc2f4784a3b0e14a80a86fc836f427
 verifiable_load_upkeep_wrapper: ../../contracts/solc/v0.8.16/VerifiableLoadUpkeep/VerifiableLoadUpkeep.abi ../../contracts/solc/v0.8.16/VerifiableLoadUpkeep/VerifiableLoadUpkeep.bin a3e02c43756ea91e7ce4b81e48c11648f1d12f6663c236780147e41dfa36ebee
 vrf_consumer_v2: ../../contracts/solc/v0.8.6/VRFConsumerV2/VRFConsumerV2.abi ../../contracts/solc/v0.8.6/VRFConsumerV2/VRFConsumerV2.bin 9ef258bf8e9f8d880fd229ceb145593d91e24fc89366baa0bf19169c5787d15f
 vrf_consumer_v2_plus_upgradeable_example: ../../contracts/solc/v0.8.6/VRFConsumerV2PlusUpgradeableExample/VRFConsumerV2PlusUpgradeableExample.abi ../../contracts/solc/v0.8.6/VRFConsumerV2PlusUpgradeableExample/VRFConsumerV2PlusUpgradeableExample.bin 3155c611e4d6882e9324b6e975033b31356776ea8b031ca63d63da37589d583b

core/services/ocr2/plugins/ocr2keeper/evmregistry/v21/encoding/interface.go

@@ -1,6 +1,8 @@
 package encoding
 
 import (
+	"net/http"
+
 	ocr2keepers "github.com/smartcontractkit/chainlink-common/pkg/types/automation"
 
 	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/generated/automation_utils_2_1"
@@ -43,6 +45,36 @@ const (
 	UpkeepNotAuthorized    PipelineExecutionState = 9
 )
 
+// ErrCode is used for invoking an error handler with a specific error code.
+type ErrCode uint32
+
+const (
+	ErrCodeNil                 ErrCode = 0
+	ErrCodePartialContent      ErrCode = 808206
+	ErrCodeDataStreamsError    ErrCode = 808500
+	ErrCodeBadRequest          ErrCode = 808400
+	ErrCodeUnauthorized        ErrCode = 808401
+	ErrCodeEncodingError       ErrCode = 808600
+	ErrCodeStreamLookupTimeout ErrCode = 808603
+)
+
+func HttpToErrCode(statusCode int) ErrCode {
+	switch statusCode {
+	case http.StatusOK:
+		return ErrCodeNil
+	case http.StatusPartialContent:
+		return ErrCodePartialContent
+	case http.StatusBadRequest:
+		return ErrCodeBadRequest
+	case http.StatusUnauthorized:
+		return ErrCodeUnauthorized
+	case http.StatusInternalServerError, http.StatusBadGateway, http.StatusServiceUnavailable, http.StatusGatewayTimeout:
+		return ErrCodeDataStreamsError
+	default:
+		return 0
+	}
+}
+
 type UpkeepInfo = iregistry21.KeeperRegistryBase21UpkeepInfo
 
 type Packer interface {

core/services/ocr2/plugins/ocr2keeper/evmregistry/v21/mercury/mercury.go

@@ -26,7 +26,8 @@ const (
 	BlockNumber              = "blockNumber" // valid for v0.2
 	Timestamp                = "timestamp"   // valid for v0.3
 	totalFastPluginRetries   = 5
-	totalMediumPluginRetries = 10
+	totalMediumPluginRetries = totalFastPluginRetries + 1
+	RetryIntervalTimeout     = time.Duration(-1)
 )
 
 var GenerateHMACFn = func(method string, path string, body []byte, clientId string, secret string, ts int64) string {
@@ -45,8 +46,7 @@ var GenerateHMACFn = func(method string, path string, body []byte, clientId stri
 }
 
 // CalculateRetryConfig returns plugin retry interval based on how many times plugin has retried this work
-var CalculateRetryConfigFn = func(prk string, mercuryConfig MercuryConfigProvider) time.Duration {
-	var retryInterval time.Duration
+var CalculateRetryConfigFn = func(prk string, mercuryConfig MercuryConfigProvider) (retryInterval time.Duration) {
 	var retries int
 	totalAttempts, ok := mercuryConfig.GetPluginRetry(prk)
 	if ok {
@@ -55,9 +55,9 @@ var CalculateRetryConfigFn = func(prk string, mercuryConfig MercuryConfigProvide
 			retryInterval = 1 * time.Second
 		} else if retries < totalMediumPluginRetries {
 			retryInterval = 5 * time.Second
+		} else {
+			retryInterval = RetryIntervalTimeout
 		}
-		// if the core node has retried totalMediumPluginRetries times, do not set retry interval and plugin will use
-		// the default interval
 	} else {
 		retryInterval = 1 * time.Second
 	}
@@ -68,6 +68,7 @@ var CalculateRetryConfigFn = func(prk string, mercuryConfig MercuryConfigProvide
 type MercuryData struct {
 	Index     int
 	Error     error
+	ErrCode   encoding.ErrCode
 	Retryable bool
 	Bytes     [][]byte
 	State     encoding.PipelineExecutionState
@@ -86,7 +87,17 @@ type HttpClient interface {
 }
 
 type MercuryClient interface {
-	DoRequest(ctx context.Context, streamsLookup *StreamsLookup, pluginRetryKey string) (encoding.PipelineExecutionState, encoding.UpkeepFailureReason, [][]byte, bool, time.Duration, error)
+	// DoRequest makes a data stream request, it manages retries and returns the following:
+	// state: the state of the pipeline execution, used by our components.
+	// upkeepFailureReason: the reason for the upkeep failure, used by our components.
+	// data: the data returned from the data stream.
+	// retryable: whether the request is retryable.
+	// retryInterval: the interval to wait before retrying the request, or RetryIntervalTimeout if no more retries are allowed.
+	// errCode: the error code of the request, to be passed to the user's error handler if applicable.
+	// error: the raw error that occurred during the request.
+	//
+	// Exploratory: consider to merge state/failureReason/errCode into a single object
+	DoRequest(ctx context.Context, streamsLookup *StreamsLookup, pluginRetryKey string) (encoding.PipelineExecutionState, encoding.UpkeepFailureReason, [][]byte, bool, time.Duration, encoding.ErrCode, error)
 }
 
 type StreamsLookupError struct {