Support v0.3 schema report format in LLO (#13780)

* Support v0.3 legacy reports in streams

- Add 'stream' to allowed job type list
- Improve streams logging; add "Verbose" option
- Bump chainlink-data-streams
- Fix some LLO bugs
- Close MERC-3525
- Close MERC-4184
- Close MERC-5934
- Handle case where LINK or NATIVE price query fails
- Close MERC-5952

* Fix rebase

* Prettier

* Bump migration number

.changeset/nice-turtles-begin.md

@@ -0,0 +1,5 @@
+---
+"chainlink": patch
+---
+
+Further development of LLO plugin (parallel composition) #wip

contracts/.changeset/eight-timers-sip.md

@@ -0,0 +1,5 @@
+---
+'@chainlink/contracts': patch
+---
+
+Add new channel definitions config store contract for parallel compositions #added

contracts/src/v0.8/llo-feeds/dev/ChannelConfigStore.sol

@@ -6,97 +6,21 @@ import {IChannelConfigStore} from "./interfaces/IChannelConfigStore.sol";
 import {TypeAndVersionInterface} from "../../interfaces/TypeAndVersionInterface.sol";
 
 contract ChannelConfigStore is ConfirmedOwner, IChannelConfigStore, TypeAndVersionInterface {
-  mapping(uint32 => ChannelDefinition) private s_channelDefinitions;
-
-  // mapping(bytes32 => ChannelConfiguration) private s_channelProductionConfigurations;
-  // mapping(bytes32 => ChannelConfiguration) private s_channelStagingConfigurations;
-
-  event NewChannelDefinition(uint32 channelId, ChannelDefinition channelDefinition);
-  event ChannelDefinitionRemoved(uint32 channelId);
-  // event NewProductionConfig(ChannelConfiguration channelConfig);
-  // event NewStagingConfig(ChannelConfiguration channelConfig);
-  event PromoteStagingConfig(uint32 channelId);
-
-  error OnlyCallableByEOA();
-  error StagingConfigAlreadyPromoted();
-  error EmptyStreamIDs();
-  error ZeroReportFormat();
-  error ZeroChainSelector();
-  error ChannelDefinitionNotFound();
+  event NewChannelDefinition(uint256 indexed donId, uint32 version, string url, bytes32 sha);
 
   constructor() ConfirmedOwner(msg.sender) {}
 
-  // function setStagingConfig(bytes32 configDigest, ChannelConfiguration calldata channelConfig) external onlyOwner {
-  //     s_channelStagingConfigurations[channelId] = channelConfig;
-
-  //     emit NewStagingConfig(channelConfig);
-  // }
-
-  //// this will trigger the following:
-  //// - offchain ShouldRetireCache will start returning true for the old (production)
-  ////   protocol instance
-  //// - once the old production instance retires it will generate a handover
-  ////   retirement report
-  //// - the staging instance will become the new production instance once
-  ////   any honest oracle that is on both instances forward the retirement
-  ////   report from the old instance to the new instace via the
-  ////   PredecessorRetirementReportCache
-  ////
-  //// Note: the promotion flow only works if the previous production instance
-  //// is working correctly & generating reports. If that's not the case, the
-  //// owner is expected to "setProductionConfig" directly instead. This will
-  //// cause "gaps" to be created, but that seems unavoidable in such a scenario.
-  // function promoteStagingConfig(bytes32 configDigest) external onlyOwner {
-  //     ChannelConfiguration memory stagingConfig = s_channelStagingConfigurations[channelId];
-
-  //     if(stagingConfig.channelConfigId.length == 0) {
-  //         revert StagingConfigAlreadyPromoted();
-  //     }
-
-  //     s_channelProductionConfigurations[channelId] = s_channelStagingConfigurations[channelId];
-
-  //     emit PromoteStagingConfig(channelId);
-  // }
-
-  function addChannel(uint32 channelId, ChannelDefinition calldata channelDefinition) external onlyOwner {
-    if (channelDefinition.streamIDs.length == 0) {
-      revert EmptyStreamIDs();
-    }
-
-    if (channelDefinition.chainSelector == 0) {
-      revert ZeroChainSelector();
-    }
-
-    if (channelDefinition.reportFormat == 0) {
-      revert ZeroReportFormat();
-    }
-
-    s_channelDefinitions[channelId] = channelDefinition;
-
-    emit NewChannelDefinition(channelId, channelDefinition);
-  }
-
-  function removeChannel(uint32 channelId) external onlyOwner {
-    if (s_channelDefinitions[channelId].streamIDs.length == 0) {
-      revert ChannelDefinitionNotFound();
-    }
-
-    delete s_channelDefinitions[channelId];
-
-    emit ChannelDefinitionRemoved(channelId);
-  }
-
-  function getChannelDefinitions(uint32 channelId) external view returns (ChannelDefinition memory) {
-    // solhint-disable-next-line avoid-tx-origin
-    if (msg.sender != tx.origin) {
-      revert OnlyCallableByEOA();
-    }
+  /// @notice The version of a channel definition keyed by DON ID
+  // Increments by 1 on every update
+  mapping(uint256 => uint256) internal s_channelDefinitionVersions;
 
-    return s_channelDefinitions[channelId];
+  function setChannelDefinitions(uint32 donId, string calldata url, bytes32 sha) external onlyOwner {
+    uint32 newVersion = uint32(++s_channelDefinitionVersions[uint256(donId)]);
+    emit NewChannelDefinition(donId, newVersion, url, sha);
   }
 
   function typeAndVersion() external pure override returns (string memory) {
-    return "ChannelConfigStore 0.0.0";
+    return "ChannelConfigStore 0.0.1";
   }
 
   function supportsInterface(bytes4 interfaceId) external pure returns (bool) {

contracts/src/v0.8/llo-feeds/dev/interfaces/IChannelConfigStore.sol

@@ -4,29 +4,5 @@ pragma solidity 0.8.19;
 import {IERC165} from "../../../vendor/openzeppelin-solidity/v4.8.3/contracts/interfaces/IERC165.sol";
 
 interface IChannelConfigStore is IERC165 {
-  // function setStagingConfig(bytes32 configDigest, ChannelConfiguration calldata channelConfig) external;
-
-  // function promoteStagingConfig(bytes32 configDigest) external;
-
-  function addChannel(uint32 channelId, ChannelDefinition calldata channelDefinition) external;
-
-  function removeChannel(uint32 channelId) external;
-
-  function getChannelDefinitions(uint32 channelId) external view returns (ChannelDefinition memory);
-
-  // struct ChannelConfiguration {
-  //     bytes32 configDigest;
-  // }
-
-  struct ChannelDefinition {
-    // e.g. evm, solana, CosmWasm, kalechain, etc...
-    uint32 reportFormat;
-    // Specifies the chain on which this channel can be verified. Currently uses
-    // CCIP chain selectors, but lots of other schemes are possible as well.
-    uint64 chainSelector;
-    // We assume that StreamIDs is always non-empty and that the 0-th stream
-    // contains the verification price in LINK and the 1-st stream contains the
-    // verification price in the native coin.
-    uint32[] streamIDs;
-  }
+  function setChannelDefinitions(uint32 donId, string calldata url, bytes32 sha) external;
 }

contracts/src/v0.8/llo-feeds/dev/test/ChannelConfigStore.t.sol

@@ -0,0 +1,41 @@
+// SPDX-License-Identifier: UNLICENSED
+pragma solidity 0.8.19;
+
+import {IChannelConfigStore} from "../interfaces/IChannelConfigStore.sol";
+import {Test} from "forge-std/Test.sol";
+import {ChannelConfigStore} from "../ChannelConfigStore.sol";
+import {ExposedChannelConfigStore} from "./mocks/ExposedChannelConfigStore.sol";
+
+/**
+ * @title ChannelConfigStoreTest
+ * @author samsondav
+ * @notice Base class for ChannelConfigStore tests
+ */
+contract ChannelConfigStoreTest is Test {
+  ExposedChannelConfigStore public channelConfigStore;
+  event NewChannelDefinition(uint256 indexed donId, uint32 version, string url, bytes32 sha);
+
+  function setUp() public virtual {
+    channelConfigStore = new ExposedChannelConfigStore();
+  }
+
+  function testTypeAndVersion() public view {
+    assertEq(channelConfigStore.typeAndVersion(), "ChannelConfigStore 0.0.1");
+  }
+
+  function testSupportsInterface() public view {
+    assertTrue(channelConfigStore.supportsInterface(type(IChannelConfigStore).interfaceId));
+  }
+
+  function testSetChannelDefinitions() public {
+    vm.expectEmit();
+    emit NewChannelDefinition(42, 1, "url", keccak256("sha"));
+    channelConfigStore.setChannelDefinitions(42, "url", keccak256("sha"));
+
+    vm.expectEmit();
+    emit NewChannelDefinition(42, 2, "url2", keccak256("sha2"));
+    channelConfigStore.setChannelDefinitions(42, "url2", keccak256("sha2"));
+
+    assertEq(channelConfigStore.exposedReadChannelDefinitionStates(42), uint32(2));
+  }
+}

contracts/src/v0.8/llo-feeds/dev/test/mocks/ExposedChannelConfigStore.sol

@@ -0,0 +1,15 @@
+// SPDX-License-Identifier: MIT
+pragma solidity 0.8.19;
+
+import {ChannelConfigStore} from "../../ChannelConfigStore.sol";
+
+// Exposed ChannelConfigStore exposes certain internal ChannelConfigStore
+// methods/structures so that golang code can access them, and we get
+// reliable type checking on their usage
+contract ExposedChannelConfigStore is ChannelConfigStore {
+  constructor() {}
+
+  function exposedReadChannelDefinitionStates(uint256 donId) public view returns (uint256) {
+    return s_channelDefinitionVersions[donId];
+  }
+}

core/chains/evm/utils/big/big.go

@@ -188,3 +188,8 @@ func (b *Big) Sub(c *Big) *Big {
 func (b *Big) Mod(c *Big) *Big {
 	return New(bigmath.Mod(b.ToInt(), c.ToInt()))
 }
+
+// IsZero returns true if b is zero
+func (b *Big) IsZero() bool {
+	return b.ToInt().Sign() == 0
+}

core/cmd/shell.go

@@ -168,13 +168,15 @@ func (n ChainlinkAppFactory) NewApplication(ctx context.Context, cfg chainlink.G
 
 	capabilitiesRegistry := capabilities.NewRegistry(appLggr)
 
+	unrestrictedClient := clhttp.NewUnrestrictedHTTPClient()
 	// create the relayer-chain interoperators from application configuration
 	relayerFactory := chainlink.RelayerFactory{
 		Logger:               appLggr,
 		LoopRegistry:         loopRegistry,
 		GRPCOpts:             grpcOpts,
 		MercuryPool:          mercuryPool,
 		CapabilitiesRegistry: capabilitiesRegistry,
+		HTTPClient:           unrestrictedClient,
 	}
 
 	evmFactoryCfg := chainlink.EVMFactoryConfig{
@@ -228,7 +230,6 @@ func (n ChainlinkAppFactory) NewApplication(ctx context.Context, cfg chainlink.G
 	}
 
 	restrictedClient := clhttp.NewRestrictedHTTPClient(cfg.Database(), appLggr)
-	unrestrictedClient := clhttp.NewUnrestrictedHTTPClient()
 	externalInitiatorManager := webhook.NewExternalInitiatorManager(ds, unrestrictedClient)
 	return chainlink.NewApplication(chainlink.ApplicationOpts{
 		Config:                     cfg,