Merge pull request #56 from mangata-finance/feature/fix-event-receipt…

…-sync

fix OP sync for created event

avs-aggregator/core/chainio/avs_writer.go

@@ -66,7 +66,7 @@ func (w *AvsWriter) SendNewTaskVerifyBlock(ctx context.Context, blockNumber *big
 	if err != nil {
 		return taskmanager.IFinalizerTaskManagerTask{}, 0, errors.New("failed to send tx with err: " + err.Error())
 	}
-	w.logger.Infof("tx hash: %s", tx.Hash().String())
+	w.logger.Infof("tx hash: %s", receipt.TxHash.String())
 	w.logger.Info("sent new task with the AVS's task manager")
 	newTaskCreatedEvent, err := w.AvsContractBindings.TaskManager.ContractFinalizerTaskManagerFilterer.ParseNewTaskCreated(*receipt.Logs[0])
 	if err != nil {
@@ -92,7 +92,7 @@ func (w *AvsWriter) SendAggregatedResponse(ctx context.Context, task taskmanager
 	if err != nil {
 		return nil, errors.New("failed to send tx with err: " + err.Error())
 	}
-	w.logger.Infof("tx hash: %s", tx.Hash().String())
+	w.logger.Infof("tx hash: %s", receipt.TxHash.String())
 	w.logger.Info("sent aggregated response with the AVS's task manager")
 	return receipt, nil
 }

avs-finalizer/src/operator.rs

@@ -9,11 +9,17 @@ use bindings::{
     finalizer_task_manager::NewTaskCreatedFilter,
     shared_types::{G1Point, G2Point, TaskResponse},
 };
-use ethers::prelude::*;
+use ethers::providers::{Middleware, PendingTransaction};
+use ethers::{
+    contract::{stream, LogMeta},
+    providers::StreamExt,
+    types::Address,
+};
 use node_executor::ExecutorDispatch;
 use node_primitives::BlockNumber;
 
 use serde::Serialize;
+use sp_core::H256;
 use sp_runtime::traits::BlakeTwo256;
 use sp_runtime::{generic, OpaqueExtrinsic};
 use std::sync::Arc;
@@ -73,13 +79,14 @@ impl Operator {
     #[instrument(skip_all)]
     pub async fn watch_new_tasks(&self) -> eyre::Result<()> {
         let evs = self.avs_contracts.new_task_stream();
-        let mut stream: stream::EventStream<'_, _, NewTaskCreatedFilter, _> =
-            evs.subscribe().await?;
+        let mut stream: stream::EventStream<'_, _, (NewTaskCreatedFilter, LogMeta), _> =
+            evs.subscribe_with_meta().await?;
 
         while let Some(res) = stream.next().await {
             match res {
-                Ok(event) => {
-                    // tokio::time::sleep(std::time::Duration::new(5, 0)).await;
+                Ok((event, log)) => {
+                    debug!("Got new task at: {:?}", log);
+                    PendingTransaction::new(log.transaction_hash, self.client.provider()).await?;
                     info!("Executing a Block for task: {:?}", event);
                     let proofs = self.execute_block(event.task.block_number.as_u32()).await?;
                     debug!("Block executed successfully {:?}", proofs);