cleanup code & move task scheduler into its own package

pkg/coordinator/test/scheduler.go → pkg/coordinator/scheduler/scheduler.go

@@ -1,4 +1,4 @@
-package test
+package scheduler
 
 import (
 	"context"
@@ -16,7 +16,7 @@ import (
 )
 
 type TaskScheduler struct {
-	coordinator      types.Coordinator
+	services         types.TaskServices
 	logger           logrus.FieldLogger
 	rootVars         types.Variables
 	taskCount        uint64
@@ -47,15 +47,15 @@ type taskExecutionState struct {
 	resultMutex      sync.RWMutex
 }
 
-func NewTaskScheduler(log logrus.FieldLogger, coordinator types.Coordinator, variables types.Variables) *TaskScheduler {
+func NewTaskScheduler(log logrus.FieldLogger, services types.TaskServices, variables types.Variables) *TaskScheduler {
 	return &TaskScheduler{
 		logger:       log,
 		rootVars:     variables,
 		taskCount:    1,
 		rootTasks:    make([]types.Task, 0),
 		allTasks:     make([]types.Task, 0),
 		taskStateMap: make(map[types.Task]*taskExecutionState),
-		coordinator:  coordinator,
+		services:     services,
 	}
 }
 
@@ -67,8 +67,8 @@ func (ts *TaskScheduler) GetTaskCount() int {
 	return len(ts.allTasks)
 }
 
-func (ts *TaskScheduler) GetCoordinator() types.Coordinator {
-	return ts.coordinator
+func (ts *TaskScheduler) GetServices() types.TaskServices {
+	return ts.services
 }
 
 func (ts *TaskScheduler) ParseTaskOptions(rawtask *helper.RawMessage) (*types.TaskOptions, error) {

pkg/coordinator/tasks/check_clients_are_healthy/task.go

@@ -106,7 +106,7 @@ func (t *Task) processCheck() {
 	totalClientCount := 0
 	failedClients := []string{}
 
-	for _, client := range t.ctx.Scheduler.GetCoordinator().ClientPool().GetClientsByNamePatterns(t.config.ClientPattern, "") {
+	for _, client := range t.ctx.Scheduler.GetServices().ClientPool().GetClientsByNamePatterns(t.config.ClientPattern, "") {
 		totalClientCount++
 
 		checkResult := t.processClientCheck(client)

pkg/coordinator/tasks/check_consensus_attestation_stats/task.go

@@ -101,7 +101,7 @@ func (t *Task) LoadConfig() error {
 }
 
 func (t *Task) Execute(ctx context.Context) error {
-	consensusPool := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool()
+	consensusPool := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool()
 
 	blockSubscription := consensusPool.GetBlockCache().SubscribeBlockEvent(10)
 	defer blockSubscription.Unsubscribe()
@@ -152,7 +152,7 @@ func (t *Task) Execute(ctx context.Context) error {
 }
 
 func (t *Task) processBlock(ctx context.Context, block *consensus.Block) {
-	consensusPool := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool()
+	consensusPool := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool()
 	specs := consensusPool.GetBlockCache().GetSpecs()
 
 	blockBody := block.AwaitBlock(ctx, 500*time.Millisecond)
@@ -226,7 +226,7 @@ func (t *Task) processBlock(ctx context.Context, block *consensus.Block) {
 }
 
 func (t *Task) runAttestationStatsCheck(ctx context.Context, epoch uint64) {
-	consensusPool := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool()
+	consensusPool := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool()
 	canonicalFork := consensusPool.GetCanonicalFork(1)
 
 	epochVotes := t.aggregateEpochVotes(ctx, epoch)
@@ -351,7 +351,7 @@ func (t *Task) newEpochVotes(base *epochVotes) *epochVotes {
 func (t *Task) aggregateEpochVotes(ctx context.Context, epoch uint64) []*epochVotes {
 	t1 := time.Now()
 
-	consensusBlockCache := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool().GetBlockCache()
+	consensusBlockCache := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool().GetBlockCache()
 	specs := consensusBlockCache.GetSpecs()
 
 	firstSlot := epoch * specs.SlotsPerEpoch

pkg/coordinator/tasks/check_consensus_block_proposals/task.go

@@ -94,7 +94,7 @@ func (t *Task) LoadConfig() error {
 }
 
 func (t *Task) Execute(ctx context.Context) error {
-	consensusPool := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool()
+	consensusPool := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool()
 
 	blockSubscription := consensusPool.GetBlockCache().SubscribeBlockEvent(10)
 	defer blockSubscription.Unsubscribe()
@@ -138,7 +138,7 @@ func (t *Task) Execute(ctx context.Context) error {
 }
 
 func (t *Task) loadValidatorSet(ctx context.Context) {
-	client := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool().GetReadyEndpoint(consensus.UnspecifiedClient)
+	client := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool().GetReadyEndpoint(consensus.UnspecifiedClient)
 	validatorSet, err := client.GetRPCClient().GetStateValidators(ctx, "head")
 
 	if err != nil {
@@ -251,7 +251,7 @@ func (t *Task) checkBlockValidatorName(block *consensus.Block, blockData *spec.V
 		return false
 	}
 
-	validatorName := t.ctx.Scheduler.GetCoordinator().ValidatorNames().GetValidatorName(uint64(proposerIndex))
+	validatorName := t.ctx.Scheduler.GetServices().ValidatorNames().GetValidatorName(uint64(proposerIndex))
 
 	matched, err := regexp.MatchString(t.config.ValidatorNamePattern, validatorName)
 	if err != nil {

pkg/coordinator/tasks/check_consensus_finality/task.go

@@ -87,7 +87,7 @@ func (t *Task) LoadConfig() error {
 }
 
 func (t *Task) Execute(ctx context.Context) error {
-	consensusPool := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool()
+	consensusPool := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool()
 
 	wallclockSubscription := consensusPool.GetBlockCache().SubscribeWallclockEpochEvent(10)
 	defer wallclockSubscription.Unsubscribe()
@@ -119,7 +119,7 @@ func (t *Task) Execute(ctx context.Context) error {
 }
 
 func (t *Task) runFinalityCheck() bool {
-	consensusPool := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool()
+	consensusPool := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool()
 
 	_, currentEpoch, err := consensusPool.GetBlockCache().GetWallclock().Now()
 	if err != nil {

pkg/coordinator/tasks/check_consensus_forks/task.go

@@ -86,7 +86,7 @@ func (t *Task) LoadConfig() error {
 }
 
 func (t *Task) Execute(ctx context.Context) error {
-	consensusPool := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool()
+	consensusPool := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool()
 	blockSubscription := consensusPool.GetBlockCache().SubscribeBlockEvent(10)
 
 	defer blockSubscription.Unsubscribe()
@@ -109,7 +109,7 @@ func (t *Task) Execute(ctx context.Context) error {
 }
 
 func (t *Task) runCheck() types.TaskResult {
-	consensusPool := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool()
+	consensusPool := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool()
 
 	headForks := consensusPool.GetHeadForks(int64(t.config.MaxForkDistance))
 	if len(headForks)-1 > int(t.config.MaxForkCount) {

pkg/coordinator/tasks/check_consensus_proposer_duty/task.go

@@ -89,7 +89,7 @@ func (t *Task) LoadConfig() error {
 }
 
 func (t *Task) Execute(ctx context.Context) error {
-	consensusPool := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool()
+	consensusPool := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool()
 
 	wallclockEpochSubscription := consensusPool.GetBlockCache().SubscribeWallclockEpochEvent(10)
 	defer wallclockEpochSubscription.Unsubscribe()
@@ -128,7 +128,7 @@ func (t *Task) Execute(ctx context.Context) error {
 }
 
 func (t *Task) loadEpochDuties(ctx context.Context, epoch uint64) {
-	client := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool().GetReadyEndpoint(consensus.UnspecifiedClient)
+	client := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool().GetReadyEndpoint(consensus.UnspecifiedClient)
 	proposerDuties, err := client.GetRPCClient().GetProposerDuties(ctx, epoch)
 
 	if err != nil {
@@ -169,7 +169,7 @@ func (t *Task) runProposerDutyCheck(slot uint64) bool {
 		}
 
 		if t.config.ValidatorNamePattern != "" {
-			validatorName := t.ctx.Scheduler.GetCoordinator().ValidatorNames().GetValidatorName(uint64(duty.ValidatorIndex))
+			validatorName := t.ctx.Scheduler.GetServices().ValidatorNames().GetValidatorName(uint64(duty.ValidatorIndex))
 			matched, err := regexp.MatchString(t.config.ValidatorNamePattern, validatorName)
 
 			if err != nil {

pkg/coordinator/tasks/check_consensus_reorgs/task.go

@@ -90,7 +90,7 @@ func (t *Task) LoadConfig() error {
 }
 
 func (t *Task) Execute(ctx context.Context) error {
-	consensusPool := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool()
+	consensusPool := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool()
 	blockSubscription := consensusPool.GetBlockCache().SubscribeBlockEvent(10)
 
 	defer blockSubscription.Unsubscribe()
@@ -134,7 +134,7 @@ func (t *Task) Execute(ctx context.Context) error {
 }
 
 func (t *Task) runCheck() types.TaskResult {
-	consensusPool := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool()
+	consensusPool := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool()
 
 	_, currentEpoch, err := consensusPool.GetBlockCache().GetWallclock().Now()
 	if err != nil {
@@ -158,7 +158,7 @@ func (t *Task) runCheck() types.TaskResult {
 }
 
 func (t *Task) processChainReorg(oldHead, newHead *consensus.Block) {
-	consensusPool := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool()
+	consensusPool := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool()
 	parentHead := oldHead
 	newHeadDistance := uint64(0)
 	oldHeadDistance := uint64(0)

pkg/coordinator/tasks/check_consensus_slot_range/task.go

@@ -87,7 +87,7 @@ func (t *Task) LoadConfig() error {
 }
 
 func (t *Task) Execute(ctx context.Context) error {
-	consensusPool := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool()
+	consensusPool := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool()
 
 	wallclockSubscription := consensusPool.GetBlockCache().SubscribeWallclockSlotEvent(10)
 	defer wallclockSubscription.Unsubscribe()
@@ -114,7 +114,7 @@ func (t *Task) Execute(ctx context.Context) error {
 }
 
 func (t *Task) runRangeCheck() (checkResult, isLower bool) {
-	consensusPool := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool()
+	consensusPool := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool()
 
 	currentSlot, currentEpoch, err := consensusPool.GetBlockCache().GetWallclock().Now()
 	if err != nil {

pkg/coordinator/tasks/check_consensus_sync_status/task.go

@@ -105,7 +105,7 @@ func (t *Task) processCheck(ctx context.Context) {
 	allResultsPass := true
 	failedClients := []string{}
 
-	for _, client := range t.ctx.Scheduler.GetCoordinator().ClientPool().GetClientsByNamePatterns(t.config.ClientPattern, "") {
+	for _, client := range t.ctx.Scheduler.GetServices().ClientPool().GetClientsByNamePatterns(t.config.ClientPattern, "") {
 		var checkResult bool
 
 		checkLogger := t.logger.WithField("client", client.Config.Name)

pkg/coordinator/tasks/check_consensus_validator_status/task.go

@@ -91,7 +91,7 @@ func (t *Task) LoadConfig() error {
 }
 
 func (t *Task) Execute(ctx context.Context) error {
-	consensusPool := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool()
+	consensusPool := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool()
 
 	wallclockEpochSubscription := consensusPool.GetBlockCache().SubscribeWallclockEpochEvent(10)
 	defer wallclockEpochSubscription.Unsubscribe()
@@ -121,7 +121,7 @@ func (t *Task) Execute(ctx context.Context) error {
 }
 
 func (t *Task) loadValidatorSet(ctx context.Context) {
-	client := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool().GetReadyEndpoint(consensus.UnspecifiedClient)
+	client := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool().GetReadyEndpoint(consensus.UnspecifiedClient)
 	validatorSet, err := client.GetRPCClient().GetStateValidators(ctx, "head")
 
 	if err != nil {
@@ -157,7 +157,7 @@ func (t *Task) runValidatorStatusCheck() bool {
 		}
 
 		if t.config.ValidatorNamePattern != "" {
-			validatorName := t.ctx.Scheduler.GetCoordinator().ValidatorNames().GetValidatorName(uint64(validator.Index))
+			validatorName := t.ctx.Scheduler.GetServices().ValidatorNames().GetValidatorName(uint64(validator.Index))
 			matched, err := regexp.MatchString(t.config.ValidatorNamePattern, validatorName)
 
 			if err != nil {

pkg/coordinator/tasks/check_execution_sync_status/task.go

@@ -105,7 +105,7 @@ func (t *Task) processCheck(ctx context.Context) {
 	allResultsPass := true
 	failedClients := []string{}
 
-	for _, client := range t.ctx.Scheduler.GetCoordinator().ClientPool().GetClientsByNamePatterns(t.config.ClientPattern, "") {
+	for _, client := range t.ctx.Scheduler.GetServices().ClientPool().GetClientsByNamePatterns(t.config.ClientPattern, "") {
 		var checkResult bool
 
 		checkLogger := t.logger.WithField("client", client.Config.Name)

pkg/coordinator/tasks/generate_blob_transactions/task.go

@@ -102,12 +102,12 @@ func (t *Task) LoadConfig() error {
 	}
 
 	if config.ChildWallets == 0 {
-		t.wallet, err = t.ctx.Scheduler.GetCoordinator().WalletManager().GetWalletByPrivkey(privKey)
+		t.wallet, err = t.ctx.Scheduler.GetServices().WalletManager().GetWalletByPrivkey(privKey)
 		if err != nil {
 			return fmt.Errorf("cannot initialize wallet: %w", err)
 		}
 	} else {
-		t.walletPool, err = t.ctx.Scheduler.GetCoordinator().WalletManager().GetWalletPoolByPrivkey(privKey, config.ChildWallets, config.WalletSeed)
+		t.walletPool, err = t.ctx.Scheduler.GetServices().WalletManager().GetWalletPoolByPrivkey(privKey, config.ChildWallets, config.WalletSeed)
 		if err != nil {
 			return fmt.Errorf("cannot initialize wallet pool: %w", err)
 		}
@@ -162,7 +162,7 @@ func (t *Task) Execute(ctx context.Context) error {
 
 	var subscription *execution.Subscription[*execution.Block]
 	if t.config.LimitPerBlock > 0 {
-		subscription = t.ctx.Scheduler.GetCoordinator().ClientPool().GetExecutionPool().GetBlockCache().SubscribeBlockEvent(10)
+		subscription = t.ctx.Scheduler.GetServices().ClientPool().GetExecutionPool().GetBlockCache().SubscribeBlockEvent(10)
 		defer subscription.Unsubscribe()
 	}
 
@@ -297,7 +297,7 @@ func (t *Task) generateTransaction(ctx context.Context, transactionIdx uint64, c
 		}
 
 		txObj := &ethtypes.BlobTx{
-			ChainID:    uint256.MustFromBig(t.ctx.Scheduler.GetCoordinator().ClientPool().GetExecutionPool().GetBlockCache().GetChainID()),
+			ChainID:    uint256.MustFromBig(t.ctx.Scheduler.GetServices().ClientPool().GetExecutionPool().GetBlockCache().GetChainID()),
 			Nonce:      nonce,
 			BlobFeeCap: uint256.MustFromBig(t.config.BlobFeeCap),
 			GasTipCap:  uint256.MustFromBig(t.config.TipCap),
@@ -318,7 +318,7 @@ func (t *Task) generateTransaction(ctx context.Context, transactionIdx uint64, c
 
 	var clients []*execution.Client
 
-	clientPool := t.ctx.Scheduler.GetCoordinator().ClientPool()
+	clientPool := t.ctx.Scheduler.GetServices().ClientPool()
 
 	if t.config.ClientPattern == "" && t.config.ExcludeClientPattern == "" {
 		clients = clientPool.GetExecutionPool().GetReadyEndpoints()

pkg/coordinator/tasks/generate_bls_changes/task.go

@@ -122,7 +122,7 @@ func (t *Task) Execute(ctx context.Context) error {
 
 	var subscription *consensus.Subscription[*consensus.Block]
 	if t.config.LimitPerSlot > 0 {
-		subscription = t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool().GetBlockCache().SubscribeBlockEvent(10)
+		subscription = t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool().GetBlockCache().SubscribeBlockEvent(10)
 		defer subscription.Unsubscribe()
 	}
 
@@ -173,7 +173,7 @@ func (t *Task) Execute(ctx context.Context) error {
 }
 
 func (t *Task) loadChainState(ctx context.Context) (map[phase0.ValidatorIndex]*v1.Validator, error) {
-	client := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool().GetReadyEndpoint(consensus.UnspecifiedClient)
+	client := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool().GetReadyEndpoint(consensus.UnspecifiedClient)
 
 	validators, err := client.GetRPCClient().GetStateValidators(ctx, "head")
 	if err != nil {
@@ -184,7 +184,7 @@ func (t *Task) loadChainState(ctx context.Context) (map[phase0.ValidatorIndex]*v
 }
 
 func (t *Task) generateBlsChange(ctx context.Context, accountIdx uint64, validators map[phase0.ValidatorIndex]*v1.Validator) error {
-	clientPool := t.ctx.Scheduler.GetCoordinator().ClientPool()
+	clientPool := t.ctx.Scheduler.GetServices().ClientPool()
 	validatorKeyPath := fmt.Sprintf("m/12381/3600/%d/0/0", accountIdx)
 
 	validatorPrivkey, err := util.PrivateKeyFromSeedAndPath(t.withdrSeed, validatorKeyPath)

pkg/coordinator/tasks/generate_child_wallet/task.go

@@ -89,7 +89,7 @@ func (t *Task) LoadConfig() error {
 		return err
 	}
 
-	t.wallet, err = t.ctx.Scheduler.GetCoordinator().WalletManager().GetWalletByPrivkey(privKey)
+	t.wallet, err = t.ctx.Scheduler.GetServices().WalletManager().GetWalletByPrivkey(privKey)
 	if err != nil {
 		return fmt.Errorf("cannot initialize wallet: %w", err)
 	}
@@ -112,7 +112,7 @@ func (t *Task) Execute(ctx context.Context) error {
 		walletSeed = t.randStringBytes(20)
 	}
 
-	walletPool, err := t.ctx.Scheduler.GetCoordinator().WalletManager().GetWalletPoolByPrivkey(t.wallet.GetPrivateKey(), 1, walletSeed)
+	walletPool, err := t.ctx.Scheduler.GetServices().WalletManager().GetWalletPoolByPrivkey(t.wallet.GetPrivateKey(), 1, walletSeed)
 	if err != nil {
 		return err
 	}

pkg/coordinator/tasks/generate_deposits/task.go

@@ -136,7 +136,7 @@ func (t *Task) Execute(ctx context.Context) error {
 
 	var subscription *consensus.Subscription[*consensus.Block]
 	if t.config.LimitPerSlot > 0 {
-		subscription = t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool().GetBlockCache().SubscribeBlockEvent(10)
+		subscription = t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool().GetBlockCache().SubscribeBlockEvent(10)
 		defer subscription.Unsubscribe()
 	}
 
@@ -285,7 +285,7 @@ func (t *Task) Execute(ctx context.Context) error {
 }
 
 func (t *Task) loadChainState(ctx context.Context) (map[phase0.ValidatorIndex]*v1.Validator, error) {
-	client := t.ctx.Scheduler.GetCoordinator().ClientPool().GetConsensusPool().GetReadyEndpoint(consensus.UnspecifiedClient)
+	client := t.ctx.Scheduler.GetServices().ClientPool().GetConsensusPool().GetReadyEndpoint(consensus.UnspecifiedClient)
 
 	validators, err := client.GetRPCClient().GetStateValidators(ctx, "head")
 	if err != nil {
@@ -296,7 +296,7 @@ func (t *Task) loadChainState(ctx context.Context) (map[phase0.ValidatorIndex]*v
 }
 
 func (t *Task) generateDeposit(ctx context.Context, accountIdx uint64, validators map[phase0.ValidatorIndex]*v1.Validator, onConfirm func(tx *ethtypes.Transaction, receipt *ethtypes.Receipt)) (*common.BLSPubkey, *ethtypes.Transaction, error) {
-	clientPool := t.ctx.Scheduler.GetCoordinator().ClientPool()
+	clientPool := t.ctx.Scheduler.GetServices().ClientPool()
 	validatorKeyPath := fmt.Sprintf("m/12381/3600/%d/0/0", accountIdx)
 	withdrAccPath := fmt.Sprintf("m/12381/3600/%d/0", accountIdx)
 
@@ -375,7 +375,7 @@ func (t *Task) generateDeposit(ctx context.Context, accountIdx uint64, validator
 		return nil, nil, fmt.Errorf("cannot create bound instance of DepositContract: %w", err)
 	}
 
-	wallet, err := t.ctx.Scheduler.GetCoordinator().WalletManager().GetWalletByPrivkey(t.walletPrivKey)
+	wallet, err := t.ctx.Scheduler.GetServices().WalletManager().GetWalletByPrivkey(t.walletPrivKey)
 	if err != nil {
 		return nil, nil, fmt.Errorf("cannot initialize wallet: %w", err)
 	}