added generate_eoa_transactions task

pkg/coordinator/clients/execution/block.go

@@ -1,6 +1,7 @@
 package execution
 
 import (
+	"context"
 	"sort"
 	"sync"
 	"time"
@@ -45,10 +46,15 @@ func (block *Block) GetBlock() *types.Block {
 	return block.block
 }
 
-func (block *Block) AwaitBlock(timeout time.Duration) *types.Block {
+func (block *Block) AwaitBlock(ctx context.Context, timeout time.Duration) *types.Block {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+
 	select {
 	case <-block.blockChan:
 	case <-time.After(timeout):
+	case <-ctx.Done():
 	}
 
 	return block.block

pkg/coordinator/clients/execution/pool.go

@@ -3,8 +3,6 @@ package execution
 import (
 	"fmt"
 	"sync"
-
-	"github.com/ethereum/go-ethereum/common"
 )
 
 type SchedulerMode uint8
@@ -30,9 +28,6 @@ type Pool struct {
 	schedulerMode  SchedulerMode
 	schedulerMutex sync.Mutex
 	rrLastIndexes  map[ClientType]uint16
-
-	walletsMutex sync.Mutex
-	walletsMap   map[common.Address]*Wallet
 }
 
 func NewPool(config *PoolConfig) (*Pool, error) {
@@ -41,7 +36,6 @@ func NewPool(config *PoolConfig) (*Pool, error) {
 		clients:       make([]*Client, 0),
 		forkCache:     map[int64][]*HeadFork{},
 		rrLastIndexes: map[ClientType]uint16{},
-		walletsMap:    map[common.Address]*Wallet{},
 	}
 
 	var err error
@@ -84,6 +78,13 @@ func (pool *Pool) GetAllEndpoints() []*Client {
 }
 
 func (pool *Pool) GetReadyEndpoint(clientType ClientType) *Client {
+	readyClients := pool.GetReadyEndpoints()
+	selectedClient := pool.runClientScheduler(readyClients, clientType)
+
+	return selectedClient
+}
+
+func (pool *Pool) GetReadyEndpoints() []*Client {
 	canonicalFork := pool.GetCanonicalFork(-1)
 	if canonicalFork == nil {
 		return nil
@@ -94,9 +95,7 @@ func (pool *Pool) GetReadyEndpoint(clientType ClientType) *Client {
 		return nil
 	}
 
-	selectedClient := pool.runClientScheduler(readyClients, clientType)
-
-	return selectedClient
+	return readyClients
 }
 
 func (pool *Pool) IsClientReady(client *Client) bool {

pkg/coordinator/coordinator.go

@@ -12,6 +12,7 @@ import (
 	"github.com/ethpandaops/assertoor/pkg/coordinator/test"
 	"github.com/ethpandaops/assertoor/pkg/coordinator/types"
 	"github.com/ethpandaops/assertoor/pkg/coordinator/vars"
+	"github.com/ethpandaops/assertoor/pkg/coordinator/wallet"
 	"github.com/ethpandaops/assertoor/pkg/coordinator/web/server"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/sirupsen/logrus"
@@ -22,6 +23,7 @@ type Coordinator struct {
 	Config          *Config
 	log             logrus.FieldLogger
 	clientPool      *clients.ClientPool
+	walletManager   *wallet.Manager
 	webserver       *server.WebServer
 	validatorNames  *names.ValidatorNames
 	tests           []types.Test
@@ -54,6 +56,7 @@ func (c *Coordinator) Run(ctx context.Context) error {
 	}
 
 	c.clientPool = clientPool
+	c.walletManager = wallet.NewManager(clientPool.GetExecutionPool(), c.log.WithField("module", "wallet"))
 
 	for idx := range c.Config.Endpoints {
 		err = clientPool.AddClient(&c.Config.Endpoints[idx])
@@ -128,6 +131,10 @@ func (c *Coordinator) ClientPool() *clients.ClientPool {
 	return c.clientPool
 }
 
+func (c *Coordinator) WalletManager() *wallet.Manager {
+	return c.walletManager
+}
+
 func (c *Coordinator) ValidatorNames() *names.ValidatorNames {
 	return c.validatorNames
 }

pkg/coordinator/tasks/generate_deposits/task.go

@@ -271,7 +271,7 @@ func (t *Task) generateDeposit(ctx context.Context, accountIdx uint64, validator
 		return fmt.Errorf("cannot create bound instance of DepositContract: %w", err)
 	}
 
-	wallet, err := clientPool.GetExecutionPool().GetWalletByPrivkey(t.walletPrivKey)
+	wallet, err := t.ctx.Scheduler.GetCoordinator().WalletManager().GetWalletByPrivkey(t.walletPrivKey)
 	if err != nil {
 		return fmt.Errorf("cannot initialize wallet: %w", err)
 	}