fixes

ethpandaops · Dec 5, 2023 · a522896 · a522896
1 parent 7c60796
commit a522896
Show file tree

Hide file tree

Showing 4 changed files with 45 additions and 25 deletions.
diff --git a/pkg/coordinator/tasks/check_consensus_sync_status/task.go b/pkg/coordinator/tasks/check_consensus_sync_status/task.go
@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/ethpandaops/minccino/pkg/coordinator/clients"
+	"github.com/ethpandaops/minccino/pkg/coordinator/clients/consensus/rpc"
 	"github.com/ethpandaops/minccino/pkg/coordinator/types"
 	"github.com/imdario/mergo"
 	"github.com/sirupsen/logrus"
@@ -102,7 +103,23 @@ func (t *Task) processCheck(ctx context.Context) {
 	failedClients := []string{}
 
 	for _, client := range t.ctx.Scheduler.GetCoordinator().ClientPool().GetClientsByNamePatterns(t.config.ClientNamePatterns) {
-		checkResult := t.processClientCheck(ctx, client)
+		var checkResult bool
+
+		checkLogger := t.logger.WithField("client", client.Config.Name)
+		syncStatus, err := client.ConsensusClient.GetRPCClient().GetNodeSyncStatus(ctx)
+
+		if ctx.Err() != nil {
+			return
+		}
+
+		if err != nil {
+			checkLogger.Warnf("errof fetching sync status: %v", err)
+
+			checkResult = false
+		} else {
+			checkResult = t.processClientCheck(client, syncStatus, checkLogger)
+		}
+
 		if !checkResult {
 			allResultsPass = false
 
@@ -119,15 +136,7 @@ func (t *Task) processCheck(ctx context.Context) {
 	}
 }
 
-func (t *Task) processClientCheck(ctx context.Context, client *clients.PoolClient) bool {
-	checkLogger := t.logger.WithField("client", client.Config.Name)
-
-	syncStatus, err := client.ConsensusClient.GetRPCClient().GetNodeSyncStatus(ctx)
-	if err != nil {
-		checkLogger.Warnf("errof fetching sync status: %v", err)
-		return false
-	}
-
+func (t *Task) processClientCheck(client *clients.PoolClient, syncStatus *rpc.SyncStatus, checkLogger logrus.FieldLogger) bool {
 	clientIdx := client.ExecutionClient.GetIndex()
 	if t.firstHeight[clientIdx] == 0 {
 		t.firstHeight[clientIdx] = syncStatus.HeadSlot

diff --git a/pkg/coordinator/tasks/check_execution_sync_status/task.go b/pkg/coordinator/tasks/check_execution_sync_status/task.go
@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/ethpandaops/minccino/pkg/coordinator/clients"
+	"github.com/ethpandaops/minccino/pkg/coordinator/clients/execution/rpc"
 	"github.com/ethpandaops/minccino/pkg/coordinator/types"
 	"github.com/imdario/mergo"
 	"github.com/sirupsen/logrus"
@@ -102,7 +103,23 @@ func (t *Task) processCheck(ctx context.Context) {
 	failedClients := []string{}
 
 	for _, client := range t.ctx.Scheduler.GetCoordinator().ClientPool().GetClientsByNamePatterns(t.config.ClientNamePatterns) {
-		checkResult := t.processClientCheck(ctx, client)
+		var checkResult bool
+
+		checkLogger := t.logger.WithField("client", client.Config.Name)
+		syncStatus, err := client.ExecutionClient.GetRPCClient().GetNodeSyncing(ctx)
+
+		if ctx.Err() != nil {
+			return
+		}
+
+		if err != nil {
+			checkLogger.Warnf("errof fetching sync status: %v", err)
+
+			checkResult = false
+		} else {
+			checkResult = t.processClientCheck(client, syncStatus, checkLogger)
+		}
+
 		if !checkResult {
 			allResultsPass = false
 
@@ -119,15 +136,7 @@ func (t *Task) processCheck(ctx context.Context) {
 	}
 }
 
-func (t *Task) processClientCheck(ctx context.Context, client *clients.PoolClient) bool {
-	checkLogger := t.logger.WithField("client", client.Config.Name)
-
-	syncStatus, err := client.ExecutionClient.GetRPCClient().GetNodeSyncing(ctx)
-	if err != nil {
-		checkLogger.Warnf("errof fetching sync status: %v", err)
-		return false
-	}
-
+func (t *Task) processClientCheck(client *clients.PoolClient, syncStatus *rpc.SyncStatus, checkLogger logrus.FieldLogger) bool {
 	currentBlock, _ := client.ExecutionClient.GetLastHead()
 
 	clientIdx := client.ExecutionClient.GetIndex()

diff --git a/pkg/coordinator/tasks/run_tasks_concurrent/task.go b/pkg/coordinator/tasks/run_tasks_concurrent/task.go
@@ -163,7 +163,8 @@ func (t *Task) Execute(ctx context.Context) error {
 			failureCount = 0
 			pendingCount = 0
 
-			for _, result := range resultMap {
+			for _, task := range t.tasks {
+				result := resultMap[task]
 				switch result {
 				case types.TaskResultSuccess:
 					successCount++
@@ -181,14 +182,14 @@ func (t *Task) Execute(ctx context.Context) error {
 				taskComplete = true
 			}
 
-			if failureCount >= failureLimit {
+			if !taskComplete && failureCount >= failureLimit {
 				t.logger.Infof("failure limit reached (%v success, %v failure)", successCount, failureCount)
 				t.ctx.SetResult(types.TaskResultFailure)
 
 				taskComplete = true
 			}
 
-			if pendingCount == 0 {
+			if !taskComplete && pendingCount == 0 {
 				t.logger.Infof("all child tasks completed (%v success, %v failure)", successCount, failureCount)
 				t.ctx.SetResult(types.TaskResultFailure)
 
@@ -228,12 +229,12 @@ func (t *Task) watchChildTask(_ context.Context, _ context.CancelFunc, task type
 			taskActive = false
 		}
 
-		t.logger.Debugf("result update notification for task %v (%v -> %v)", t.taskIdxMap[task], oldStatus, taskStatus.Result)
-
 		if taskStatus.Result == oldStatus {
 			continue
 		}
 
+		t.logger.Debugf("result update notification for task %v (%v -> %v)", t.taskIdxMap[task], oldStatus, taskStatus.Result)
+
 		t.resultNotifyChan <- taskResultUpdate{
 			task:   task,
 			result: taskStatus.Result,

diff --git a/pkg/coordinator/test/test.go b/pkg/coordinator/test/test.go
@@ -28,6 +28,7 @@ func CreateTest(coordinator types.Coordinator, config *Config) (types.Test, erro
 		log:     coordinator.Logger().WithField("component", "test").WithField("test", config.Name),
 		config:  config,
 		metrics: NewMetrics("sync_test_coordinator", config.Name),
+		status:  types.TestStatusPending,
 	}
 	if test.config.Timeout.Duration > 0 {
 		test.timeout = test.config.Timeout.Duration