various fixes related to transaction building & submission via spamoor

pk910 · pk910 · commit 304fbf4f5697 · 2026-02-20T10:47:06.000+01:00
diff --git a/go.mod b/go.mod
@@ -7,7 +7,7 @@ require (
 	github.com/donovanhide/eventsource v0.0.0-20210830082556-c59027999da0
 	github.com/ethereum/go-ethereum v1.17.0
 	github.com/ethpandaops/ethwallclock v0.4.0
-	github.com/ethpandaops/spamoor v1.1.17-0.20260210191103-c3c5caa2a6b6
+	github.com/ethpandaops/spamoor v1.1.18-0.20260220084150-d48b77369e81
 	github.com/glebarez/go-sqlite v1.22.0
 	github.com/golang-jwt/jwt/v5 v5.3.1
 	github.com/google/uuid v1.6.0
diff --git a/go.sum b/go.sum
@@ -75,8 +75,8 @@ github.com/ethereum/go-ethereum v1.17.0 h1:2D+1Fe23CwZ5tQoAS5DfwKFNI1HGcTwi65/kR
 github.com/ethereum/go-ethereum v1.17.0/go.mod h1:2W3msvdosS/MCWytpqTcqgFiRYbTH59FxDJzqah120o=
 github.com/ethpandaops/ethwallclock v0.4.0 h1:+sgnhf4pk6hLPukP076VxkiLloE4L0Yk1yat+ZyHh1g=
 github.com/ethpandaops/ethwallclock v0.4.0/go.mod h1:y0Cu+mhGLlem19vnAV2x0hpFS5KZ7oOi2SWYayv9l24=
-github.com/ethpandaops/spamoor v1.1.17-0.20260210191103-c3c5caa2a6b6 h1:udcP+wIKp8yfhrbJruuApazjEjZ8wRgRcvSInYzOO0M=
-github.com/ethpandaops/spamoor v1.1.17-0.20260210191103-c3c5caa2a6b6/go.mod h1:+Z810GiNvmc+MxNj675Wo59Md6Ip8jbJvih6/7oMNc0=
+github.com/ethpandaops/spamoor v1.1.18-0.20260220084150-d48b77369e81 h1:0RtYv8dF1tAAank8R+rs/ii4QyqNdEc+dMF7knzrmZQ=
+github.com/ethpandaops/spamoor v1.1.18-0.20260220084150-d48b77369e81/go.mod h1:GW6Xjym8fliGk+Si2/dV+7M0RXIJTwW2w/ZHYXnri0A=
 github.com/fatih/color v1.16.0 h1:zmkK9Ngbjj+K0yRhTVONQh1p/HknKYSlNT+vZCzyokM=
 github.com/fatih/color v1.16.0/go.mod h1:fL2Sau1YI5c0pdGEVCbKQbLXB6edEj1ZgiY4NijnWvE=
 github.com/ferranbt/fastssz v0.1.4 h1:OCDB+dYDEQDvAgtAGnTSidK1Pe2tW3nFV40XyMkTeDY=
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
@@ -92,7 +92,7 @@ func (ts *TaskScheduler) RunTasks(testRunCtx context.Context, timeout time.Durat
 	defer ts.cancelTaskCtx()
 
 	for _, task := range ts.rootTasks {
-		err := ts.ExecuteTask(tasksCtx, task, ts.WatchTaskPass)
+		err := ts.ExecuteTask(tasksCtx, task, nil)
 		if err != nil {
 			return err
 		}
@@ -111,7 +111,7 @@ func (ts *TaskScheduler) runCleanupTasks(ctx context.Context) {
 			return
 		}
 
-		err := ts.ExecuteTask(ctx, taskIndex, ts.WatchTaskPass)
+		err := ts.ExecuteTask(ctx, taskIndex, nil)
 		if err != nil {
 			taskState := ts.getTaskState(taskIndex)
 			taskState.logger.GetLogger().Errorf("cleanup task failed: %v", err)
diff --git a/pkg/scheduler/task_execution.go b/pkg/scheduler/task_execution.go
@@ -273,25 +273,3 @@ func (ts *TaskScheduler) emitTaskProgress(taskState *taskState, percent float64,
 		message,
 	)
 }
-
-func (ts *TaskScheduler) WatchTaskPass(ctx context.Context, cancelFn context.CancelFunc, taskIndex types.TaskIndex) {
-	taskState := ts.GetTaskState(taskIndex)
-
-	// poll task result and cancel context when task result is passed or failed
-	for {
-		updateChan := taskState.GetTaskResultUpdateChan(types.TaskResultNone)
-		if updateChan != nil {
-			select {
-			case <-ctx.Done():
-				return
-			case <-updateChan:
-			}
-		}
-
-		taskStatus := taskState.GetTaskStatus()
-		if taskStatus.Result != types.TaskResultNone {
-			cancelFn()
-			return
-		}
-	}
-}
diff --git a/pkg/tasks/generate_blob_transactions/task.go b/pkg/tasks/generate_blob_transactions/task.go
@@ -219,6 +219,48 @@ func (t *Task) Execute(ctx context.Context) error {
 	return nil
 }
 
+// isFuluActive checks whether the Fulu fork is active by comparing the
+// current wallclock epoch against FULU_FORK_EPOCH from the consensus specs.
+func (t *Task) isFuluActive() bool {
+	blockCache := t.ctx.Scheduler.GetServices().ClientPool().
+		GetConsensusPool().GetBlockCache()
+
+	specValues := blockCache.GetSpecValues()
+	if specValues == nil {
+		return false
+	}
+
+	fuluEpochVal, ok := specValues["FULU_FORK_EPOCH"]
+	if !ok {
+		return false
+	}
+
+	var fuluEpoch uint64
+
+	switch v := fuluEpochVal.(type) {
+	case uint64:
+		fuluEpoch = v
+	case string:
+		if _, err := fmt.Sscanf(v, "%d", &fuluEpoch); err != nil {
+			return false
+		}
+	default:
+		return false
+	}
+
+	wallclock := blockCache.GetWallclock()
+	if wallclock == nil {
+		return false
+	}
+
+	_, currentEpoch, err := wallclock.Now()
+	if err != nil {
+		return false
+	}
+
+	return currentEpoch.Number() >= fuluEpoch
+}
+
 func (t *Task) generateTransaction(ctx context.Context, transactionIdx uint64, completeFn spamoor.TxCompleteFn) error {
 	txWallet := t.wallet
 	if t.wallet == nil {
@@ -307,6 +349,13 @@ func (t *Task) generateTransaction(ctx context.Context, transactionIdx uint64, c
 		}
 	}
 
+	walletMgr := t.ctx.Scheduler.GetServices().WalletManager()
+	spamoorClients := make([]*spamoor.Client, len(clients))
+
+	for i, c := range clients {
+		spamoorClients[i] = walletMgr.GetClient(c)
+	}
+
 	blobTx, err := txbuilder.BuildBlobTx(&txbuilder.TxMetadata{
 		GasFeeCap:  uint256.MustFromBig(t.config.FeeCap),
 		GasTipCap:  uint256.MustFromBig(t.config.TipCap),
@@ -325,11 +374,13 @@ func (t *Task) generateTransaction(ctx context.Context, transactionIdx uint64, c
 		return fmt.Errorf("cannot build blob transaction: %w", err)
 	}
 
-	walletMgr := t.ctx.Scheduler.GetServices().WalletManager()
-	client := walletMgr.GetClient(clients[transactionIdx%uint64(len(clients))])
+	// Check if Fulu is active to determine blob sidecar version.
+	// After Fulu, blob sidecars must use v1 format (cell proofs).
+	useBlobV1 := t.isFuluActive()
 
-	err = walletMgr.GetTxPool().SendTransaction(ctx, txWallet, tx, &spamoor.SendTransactionOptions{
-		Client:      client,
+	sendOpts := &spamoor.SendTransactionOptions{
+		Client:      spamoorClients[transactionIdx%uint64(len(spamoorClients))],
+		ClientList:  spamoorClients,
 		Rebroadcast: true,
 		OnComplete:  completeFn,
 		LogFn: func(client *spamoor.Client, retry int, rebroadcast int, err error) {
@@ -351,7 +402,25 @@ func (t *Task) generateTransaction(ctx context.Context, transactionIdx uint64, c
 
 			logEntry.Infof("submitted blob transaction %v (nonce: %v, attempt: %v)", transactionIdx, tx.Nonce(), retry)
 		},
-	})
+	}
+
+	if useBlobV1 {
+		convertedToV1 := false
+
+		sendOpts.OnEncode = func(tx *ethtypes.Transaction) ([]byte, error) {
+			if !convertedToV1 {
+				if convErr := tx.BlobTxSidecar().ToV1(); convErr != nil {
+					return nil, fmt.Errorf("cannot convert blob sidecar to v1: %w", convErr)
+				}
+
+				convertedToV1 = true
+			}
+
+			return nil, nil
+		}
+	}
+
+	err = walletMgr.GetTxPool().SendTransaction(ctx, txWallet, tx, sendOpts)
 	if err != nil {
 		txWallet.MarkSkippedNonce(tx.Nonce())
 		return err
diff --git a/pkg/tasks/generate_consolidations/task.go b/pkg/tasks/generate_consolidations/task.go
@@ -193,8 +193,6 @@ func (t *Task) Execute(ctx context.Context) error {
 			// Note: onComplete callback is still called by spamoor even on error,
 			// so we don't drain pendingChan or call pendingWg.Done() here
 		} else {
-			t.ctx.SetResult(types.TaskResultSuccess)
-
 			perSlotCount++
 			totalCount++
 
@@ -377,6 +375,11 @@ func (t *Task) generateConsolidation(ctx context.Context, accountIdx uint64, onC
 
 	walletMgr := t.ctx.Scheduler.GetServices().WalletManager()
 
+	spamoorClients := make([]*spamoor.Client, len(clients))
+	for i, c := range clients {
+		spamoorClients[i] = walletMgr.GetClient(c)
+	}
+
 	txWallet, err := walletMgr.GetWalletByPrivkey(ctx, t.walletPrivKey)
 	if err != nil {
 		return nil, fmt.Errorf("cannot initialize wallet: %w", err)
@@ -405,10 +408,9 @@ func (t *Task) generateConsolidation(ctx context.Context, accountIdx uint64, onC
 		return nil, fmt.Errorf("cannot build consolidation transaction: %w", err)
 	}
 
-	client := walletMgr.GetClient(clients[0])
-
 	err = walletMgr.GetTxPool().SendTransaction(ctx, txWallet, tx, &spamoor.SendTransactionOptions{
-		Client:      client,
+		Client:      spamoorClients[0],
+		ClientList:  spamoorClients,
 		Rebroadcast: true,
 		OnComplete:  onComplete,
 		LogFn: func(client *spamoor.Client, retry int, rebroadcast int, err error) {
diff --git a/pkg/tasks/generate_deposits/task.go b/pkg/tasks/generate_deposits/task.go
@@ -220,8 +220,6 @@ func (t *Task) Execute(ctx context.Context) error {
 			// Note: onComplete callback is still called by spamoor even on error,
 			// so we don't call pendingWg.Done() here
 		} else {
-			t.ctx.SetResult(types.TaskResultSuccess)
-
 			perSlotCount++
 			totalCount++
 
@@ -527,6 +525,11 @@ func (t *Task) generateDeposit(ctx context.Context, accountIdx uint64, onComplet
 
 	walletMgr := t.ctx.Scheduler.GetServices().WalletManager()
 
+	spamoorClients := make([]*spamoor.Client, len(clients))
+	for i, c := range clients {
+		spamoorClients[i] = walletMgr.GetClient(c)
+	}
+
 	txWallet, err := walletMgr.GetWalletByPrivkey(t.ctx.Scheduler.GetTestRunCtx(), t.walletPrivKey)
 	if err != nil {
 		return nil, nil, fmt.Errorf("cannot initialize wallet: %w", err)
@@ -551,10 +554,9 @@ func (t *Task) generateDeposit(ctx context.Context, accountIdx uint64, onComplet
 		return nil, nil, fmt.Errorf("cannot build deposit transaction: %w", err)
 	}
 
-	client := walletMgr.GetClient(clients[0])
-
 	err = walletMgr.GetTxPool().SendTransaction(ctx, txWallet, tx, &spamoor.SendTransactionOptions{
-		Client:      client,
+		Client:      spamoorClients[0],
+		ClientList:  spamoorClients,
 		Rebroadcast: true,
 		OnComplete:  onComplete,
 		LogFn: func(client *spamoor.Client, retry int, rebroadcast int, err error) {
diff --git a/pkg/tasks/generate_eoa_transactions/task.go b/pkg/tasks/generate_eoa_transactions/task.go
@@ -352,10 +352,15 @@ func (t *Task) generateTransaction(ctx context.Context, transactionIdx uint64, c
 	}
 
 	walletMgr := t.ctx.Scheduler.GetServices().WalletManager()
-	client := walletMgr.GetClient(clients[transactionIdx%uint64(len(clients))])
+
+	spamoorClients := make([]*spamoor.Client, len(clients))
+	for i, c := range clients {
+		spamoorClients[i] = walletMgr.GetClient(c)
+	}
 
 	err = walletMgr.GetTxPool().SendTransaction(ctx, txWallet, tx, &spamoor.SendTransactionOptions{
-		Client:      client,
+		Client:      spamoorClients[transactionIdx%uint64(len(spamoorClients))],
+		ClientList:  spamoorClients,
 		Rebroadcast: true,
 		OnComplete:  completeFn,
 		LogFn: func(client *spamoor.Client, retry int, rebroadcast int, err error) {
diff --git a/pkg/tasks/generate_transaction/task.go b/pkg/tasks/generate_transaction/task.go
@@ -207,16 +207,21 @@ func (t *Task) Execute(ctx context.Context) error {
 	}
 
 	walletMgr := t.ctx.Scheduler.GetServices().WalletManager()
+	spamoorClients := make([]*spamoor.Client, len(clients))
+	for i, c := range clients {
+		spamoorClients[i] = walletMgr.GetClient(c)
+	}
 
-	for i := 0; i < len(clients); i++ {
-		client := clients[i%len(clients)]
+	for i := 0; i < len(spamoorClients); i++ {
+		clientIdx := i % len(spamoorClients)
 
 		t.logger.WithFields(logrus.Fields{
-			"client": client.GetName(),
+			"client": spamoorClients[clientIdx].GetName(),
 		}).Infof("sending tx: %v", tx.Hash().Hex())
 
 		err = walletMgr.GetTxPool().SendTransaction(ctx, t.wallet, tx, &spamoor.SendTransactionOptions{
-			Client:             walletMgr.GetClient(client),
+			Client:             spamoorClients[clientIdx],
+			ClientList:         spamoorClients,
 			ClientsStartOffset: i,
 		})
 		if err == nil {
diff --git a/pkg/tasks/generate_withdrawal_requests/task.go b/pkg/tasks/generate_withdrawal_requests/task.go
@@ -194,8 +194,6 @@ func (t *Task) Execute(ctx context.Context) error {
 			// Note: onComplete callback is still called by spamoor even on error,
 			// so we don't drain pendingChan or call pendingWg.Done() here
 		} else {
-			t.ctx.SetResult(types.TaskResultSuccess)
-
 			perSlotCount++
 			totalCount++
 
@@ -366,6 +364,11 @@ func (t *Task) generateWithdrawal(ctx context.Context, accountIdx uint64, onComp
 
 	walletMgr := t.ctx.Scheduler.GetServices().WalletManager()
 
+	spamoorClients := make([]*spamoor.Client, len(clients))
+	for i, c := range clients {
+		spamoorClients[i] = walletMgr.GetClient(c)
+	}
+
 	txWallet, err := walletMgr.GetWalletByPrivkey(ctx, t.walletPrivKey)
 	if err != nil {
 		return nil, fmt.Errorf("cannot initialize wallet: %w", err)
@@ -397,10 +400,9 @@ func (t *Task) generateWithdrawal(ctx context.Context, accountIdx uint64, onComp
 		return nil, fmt.Errorf("cannot build withdrawal transaction: %w", err)
 	}
 
-	client := walletMgr.GetClient(clients[0])
-
 	err = walletMgr.GetTxPool().SendTransaction(ctx, txWallet, tx, &spamoor.SendTransactionOptions{
-		Client:      client,
+		Client:      spamoorClients[0],
+		ClientList:  spamoorClients,
 		Rebroadcast: true,
 		OnComplete:  onComplete,
 		LogFn: func(client *spamoor.Client, _ int, rebroadcast int, err error) {
diff --git a/pkg/tasks/run_external_tasks/task.go b/pkg/tasks/run_external_tasks/task.go
@@ -167,7 +167,7 @@ taskLoop:
 		progress := float64(i) / float64(totalTasks) * 100
 		t.ctx.ReportProgress(progress, fmt.Sprintf("Executing task %d/%d...", i+1, totalTasks))
 
-		err := t.ctx.Scheduler.ExecuteTask(ctx, task, t.ctx.Scheduler.WatchTaskPass)
+		err := t.ctx.Scheduler.ExecuteTask(ctx, task, nil)
 
 		switch {
 		case t.config.IgnoreFailure:
@@ -188,7 +188,7 @@ taskLoop:
 			return ctx.Err()
 		}
 
-		err := t.ctx.Scheduler.ExecuteTask(ctx, task, t.ctx.Scheduler.WatchTaskPass)
+		err := t.ctx.Scheduler.ExecuteTask(ctx, task, nil)
 		if err != nil {
 			t.logger.Warnf("cleanup task #%v failed: %w", i+1, err)
 		}
diff --git a/pkg/tasks/run_task_options/config.go b/pkg/tasks/run_task_options/config.go
@@ -17,6 +17,7 @@ type Config struct {
 	// Result transformation
 	InvertResult  bool `yaml:"invertResult" json:"invertResult" desc:"If true, swap success and failure results."`
 	IgnoreResult  bool `yaml:"ignoreResult" json:"ignoreResult" desc:"If true, always report success regardless of child task result."`
+	IgnoreFailure bool `yaml:"ignoreFailure" json:"ignoreFailure" desc:"If true, ignore the failure of the child task."`
 	ExpectFailure bool `yaml:"expectFailure" json:"expectFailure" desc:"Alias for invertResult - expect the task to fail."`
 }
 
diff --git a/pkg/tasks/run_task_options/task.go b/pkg/tasks/run_task_options/task.go
@@ -121,6 +121,10 @@ func (t *Task) Execute(ctx context.Context) error {
 		return nil
 	}
 
+	if t.config.IgnoreFailure && taskErr != nil {
+		return nil
+	}
+
 	// ExpectFailure is an alias for InvertResult
 	if t.config.ExpectFailure || t.config.InvertResult {
 		if taskErr != nil {
diff --git a/pkg/types/scheduler.go b/pkg/types/scheduler.go
@@ -18,7 +18,6 @@ type TaskSchedulerRunner interface {
 	GetTestRunCtx() context.Context
 	ParseTaskOptions(rawtask helper.IRawMessage) (*TaskOptions, error)
 	ExecuteTask(ctx context.Context, taskIndex TaskIndex, taskWatchFn func(ctx context.Context, cancelFn context.CancelFunc, taskIndex TaskIndex)) error
-	WatchTaskPass(ctx context.Context, cancelFn context.CancelFunc, taskIndex TaskIndex)
 }
 
 type TaskScheduler interface {
diff --git a/playbooks/pectra-dev/eip7002-all.yaml b/playbooks/pectra-dev/eip7002-all.yaml
diff --git a/playbooks/stable/kurtosis/validator-exit-test.yaml b/playbooks/stable/kurtosis/validator-exit-test.yaml
diff --git a/playbooks/stable/kurtosis/validator-slashing-test.yaml b/playbooks/stable/kurtosis/validator-slashing-test.yaml
diff --git a/playbooks/stable/kurtosis/validator-withdrawal-test.yaml b/playbooks/stable/kurtosis/validator-withdrawal-test.yaml
