Merge pull request #721 from forta-network/ali/wait-init-to-process

Wait for agent initalization to start processing
forta-network · May 18, 2023 · 3f766fb · 3f766fb
2 parents 600462b + 0abe85f
commit 3f766fb
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 3 deletions.
diff --git a/services/scanner/agentpool/agent_pool.go b/services/scanner/agentpool/agent_pool.go
@@ -148,7 +148,7 @@ func (ap *AgentPool) SendEvaluateTxRequest(req *protocol.EvaluateTxRequest) {
 	}
 	var metricsList []*protocol.AgentMetric
 	for _, agent := range agents {
-		if !agent.IsInitialized() || !agent.ShouldProcessBlock(req.Event.Block.BlockNumber) {
+		if !agent.IsReady() || !agent.ShouldProcessBlock(req.Event.Block.BlockNumber) {
 			continue
 		}
 		lg.WithFields(log.Fields{
@@ -212,7 +212,7 @@ func (ap *AgentPool) SendEvaluateBlockRequest(req *protocol.EvaluateBlockRequest
 
 	var metricsList []*protocol.AgentMetric
 	for _, agent := range agents {
-		if !agent.IsInitialized() || !agent.ShouldProcessBlock(req.Event.BlockNumber) {
+		if !agent.IsReady() || !agent.ShouldProcessBlock(req.Event.BlockNumber) {
 			continue
 		}
 
@@ -293,7 +293,7 @@ func (ap *AgentPool) SendEvaluateAlertRequest(req *protocol.EvaluateAlertRequest
 			continue
 		}
 
-		if !agent.IsInitialized() {
+		if !agent.IsReady() {
 			continue
 		}
 		target = agent

diff --git a/services/scanner/agentpool/poolagent/agent.go b/services/scanner/agentpool/poolagent/agent.go
@@ -357,6 +357,8 @@ func (agent *Agent) processTransactions() {
 			"evaluate":  "transaction",
 		},
 	)
+	// wait for initialization to start processing
+	<-agent.Initialized()
 
 	for request := range agent.txRequests {
 		if exit := agent.processTransaction(lg, request); exit {
@@ -447,6 +449,8 @@ func (agent *Agent) processBlocks() {
 			"evaluate":  "block",
 		},
 	)
+	// wait for initialization to start processing
+	<-agent.Initialized()
 
 	for request := range agent.blockRequests {
 		if exit := agent.processBlock(lg, request); exit {
@@ -528,6 +532,9 @@ func (agent *Agent) processCombinationAlerts() {
 		},
 	)
 
+	// wait for initialization to start processing
+	<-agent.Initialized()
+
 	for request := range agent.combinationRequests {
 		if exit := agent.processCombinationAlert(lg, request); exit {
 			return