feat: eth prewitnessing

engine/src/witness/eth.rs

@@ -34,10 +34,18 @@ use anyhow::{Context, Result};
 
 const SAFETY_MARGIN: usize = 7;
 
-pub async fn start<StateChainClient, StateChainStream, ProcessCall, ProcessingFut>(
+pub async fn start<
+	StateChainClient,
+	StateChainStream,
+	ProcessCall,
+	ProcessingFut,
+	PrewitnessCall,
+	PrewitnessFut,
+>(
 	scope: &Scope<'_, anyhow::Error>,
 	eth_client: EthersRetryRpcClient,
 	process_call: ProcessCall,
+	prewitness_call: PrewitnessCall,
 	state_chain_client: Arc<StateChainClient>,
 	state_chain_stream: StateChainStream,
 	epoch_source: EpochSourceBuilder<'_, '_, StateChainClient, (), ()>,
@@ -52,6 +60,12 @@ where
 		+ Clone
 		+ 'static,
 	ProcessingFut: Future<Output = ()> + Send + 'static,
+	PrewitnessCall: Fn(state_chain_runtime::RuntimeCall, EpochIndex) -> PrewitnessFut
+		+ Send
+		+ Sync
+		+ Clone
+		+ 'static,
+	PrewitnessFut: Future<Output = ()> + Send + 'static,
 {
 	let state_chain_gateway_address = state_chain_client
         .storage_value::<pallet_cf_environment::EthereumStateChainGatewayAddress<state_chain_runtime::Runtime>>(
@@ -110,12 +124,74 @@ where
 		.logging("chain tracking")
 		.spawn(scope);
 
+	let vaults = epoch_source.vaults().await;
+
+	// ===== Prewitnessing stream =====
+	let pre_witness_source = eth_source
+		.clone()
+		.strictly_monotonic()
+		.shared(scope)
+		.chunk_by_vault(vaults.clone());
+
+	pre_witness_source
+		.clone()
+		.deposit_addresses(scope, state_chain_stream.clone(), state_chain_client.clone())
+		.await
+		.erc20_deposits::<_, _, _, UsdcEvents>(
+			prewitness_call.clone(),
+			eth_client.clone(),
+			cf_primitives::chains::assets::eth::Asset::Usdc,
+			usdc_contract_address,
+		)
+		.await?
+		.spawn(scope);
+
+	pre_witness_source
+		.clone()
+		.deposit_addresses(scope, state_chain_stream.clone(), state_chain_client.clone())
+		.await
+		.erc20_deposits::<_, _, _, FlipEvents>(
+			prewitness_call.clone(),
+			eth_client.clone(),
+			cf_primitives::chains::assets::eth::Asset::Flip,
+			flip_contract_address,
+		)
+		.await?
+		.spawn(scope);
+
+	pre_witness_source
+		.clone()
+		.deposit_addresses(scope, state_chain_stream.clone(), state_chain_client.clone())
+		.await
+		.ethereum_deposits(
+			prewitness_call.clone(),
+			eth_client.clone(),
+			eth::Asset::Eth,
+			address_checker_address,
+			vault_address,
+		)
+		.await
+		.spawn(scope);
+
+	pre_witness_source
+		.vault_witnessing(
+			prewitness_call,
+			eth_client.clone(),
+			vault_address,
+			cf_primitives::Asset::Eth,
+			cf_primitives::ForeignChain::Ethereum,
+			supported_erc20_tokens.clone(),
+		)
+		.spawn(scope);
+
+	// ===== Full witnessing stream =====
+
 	let eth_safe_vault_source = eth_source
 		.strictly_monotonic()
 		.lag_safety(SAFETY_MARGIN)
 		.logging("safe block produced")
 		.shared(scope)
-		.chunk_by_vault(epoch_source.vaults().await);
+		.chunk_by_vault(vaults);
 
 	eth_safe_vault_source
 		.clone()

engine/src/witness/start.rs

@@ -59,12 +59,15 @@ where
 
 	let prewitness_call = {
 		let state_chain_client = state_chain_client.clone();
-		move |call, _epoch_index| {
+		move |call, epoch_index| {
 			let state_chain_client = state_chain_client.clone();
 			async move {
 				let _ = state_chain_client
-					.finalize_signed_extrinsic(pallet_cf_witnesser::Call::pre_witness {
-						call: Box::new(call),
+					.finalize_signed_extrinsic(pallet_cf_witnesser::Call::witness_at_epoch {
+						call: Box::new(
+							pallet_cf_witnesser::Call::pre_witness { call: Box::new(call) }.into(),
+						),
+						epoch_index,
 					})
 					.await;
 			}
@@ -75,6 +78,7 @@ where
 		scope,
 		eth_client,
 		witness_call.clone(),
+		prewitness_call.clone(),
 		state_chain_client.clone(),
 		state_chain_stream.clone(),
 		epoch_source.clone(),