From 3bababb5f3fece218773b927ddc1af4e87e9c9ed Mon Sep 17 00:00:00 2001 From: cavemanloverboy <93507302+cavemanloverboy@users.noreply.github.com> Date: Sun, 18 Oct 2026 04:31:35 +0000 Subject: [PATCH] reject blocks with unprocessable transactions --- pkg/replay/block.go | 28 +++++++++++------ pkg/replay/block_execution.go | 32 +++++++++++++------ pkg/replay/block_execution_test.go | 15 +++++++++ pkg/replay/streaming_lifecycle_test.go | 43 ++++++++++++++++++++++++++ 4 files changed, 100 insertions(+), 18 deletions(-) diff --git a/pkg/replay/block.go b/pkg/replay/block.go index 9b7bf651..8a430d84 100644 --- a/pkg/replay/block.go +++ b/pkg/replay/block.go @@ -3977,7 +3977,7 @@ func validatePreConsensusTransactionStatuses( return statuses.validateBlockWithPlan(&candidate, plan) } -func sequentialTxLoop(slotCtx *sealevel.SlotCtx, sigverifyWg *sync.WaitGroup, block *b.Block, executionPlan blockTransactionExecutionPlan, dbgOpts *DebugOptions, shouldVerifySignatures bool) (fees.TxFeeInfoAccumulator, uint64) { +func sequentialTxLoop(slotCtx *sealevel.SlotCtx, sigverifyWg *sync.WaitGroup, block *b.Block, executionPlan blockTransactionExecutionPlan, dbgOpts *DebugOptions, shouldVerifySignatures bool) (fees.TxFeeInfoAccumulator, uint64, error) { var txFeeAccumulator fees.TxFeeInfoAccumulator var totalComputeUnitsConsumed uint64 // process & execute each transaction in turn @@ -3994,7 +3994,7 @@ func sequentialTxLoop(slotCtx *sealevel.SlotCtx, sigverifyWg *sync.WaitGroup, bl if txMeta == nil { if txFeeInfo == nil { - panic(fmt.Sprintf("txFeeInfo is nil for tx %s in slot %d", tx.Signatures[0], block.Slot)) + return fees.TxFeeInfoAccumulator{}, 0, unprocessableTransactionError(idx, txErr) } txFeeAccumulator.Add(txFeeInfo) continue @@ -4024,7 +4024,7 @@ func sequentialTxLoop(slotCtx *sealevel.SlotCtx, sigverifyWg *sync.WaitGroup, bl } txFeeAccumulator.Add(txFeeInfo) } - return txFeeAccumulator, totalComputeUnitsConsumed + return txFeeAccumulator, totalComputeUnitsConsumed, nil } func lightbringerEntryExecutionBatches(transactions []*solana.Transaction, entry *b.TxEntry, relaxIntraBatchAccountLocks bool) [][]uint64 { @@ -4113,7 +4113,7 @@ func parallelTxLoop( txParallelism int, dbgOpts *DebugOptions, shouldVerifySignatures bool, -) (fees.TxFeeInfoAccumulator, uint64) { +) (fees.TxFeeInfoAccumulator, uint64, error) { var txFeeAccumulator fees.TxFeeInfoAccumulator txFeeInfos := make([]*fees.TxFeeInfo, len(block.Transactions)) txComputeUnitsConsumed := make([]uint64, len(block.Transactions)) @@ -4270,12 +4270,17 @@ func parallelTxLoop( } else { mlog.Log.Errorf(" RecentBlockhashes: nil or empty!") } - panic(fmt.Sprintf("txFeeInfo is nil - blockhash validation failed for tx %s", tx.Signatures[0])) + // With RPC metadata the transaction was processed onchain, so a + // missing fee result is a local divergence. + if idx < len(block.TxMetas) && block.TxMetas[idx] != nil { + panic(fmt.Sprintf("txFeeInfo is nil - blockhash validation failed for tx %s", tx.Signatures[0])) + } + return fees.TxFeeInfoAccumulator{}, 0, unprocessableTransactionError(idx, errs[idx]) } txFeeAccumulator.Add(txFeeInfo) } - return txFeeAccumulator, totalComputeUnitsConsumed + return txFeeAccumulator, totalComputeUnitsConsumed, nil } func ProcessBlock( @@ -4376,14 +4381,19 @@ func ProcessBlock( exec.setReplayStage("tx_loop") txLoopRegion := trace.StartRegion(exec.ctx, "TxLoop") shouldVerifySignatures := !block.TransactionSignaturesVerified() + var txLoopErr error if txParallelism > 0 { - txFeeAccumulator, totalComputeUnitsConsumed = parallelTxLoop(slotCtx, &exec.sigverifyWg, planner, block, executionPlan, txParallelism, dbgOpts, shouldVerifySignatures) + txFeeAccumulator, totalComputeUnitsConsumed, txLoopErr = parallelTxLoop(slotCtx, &exec.sigverifyWg, planner, block, executionPlan, txParallelism, dbgOpts, shouldVerifySignatures) } else { - txFeeAccumulator, totalComputeUnitsConsumed = sequentialTxLoop(slotCtx, &exec.sigverifyWg, block, executionPlan, dbgOpts, shouldVerifySignatures) + txFeeAccumulator, totalComputeUnitsConsumed, txLoopErr = sequentialTxLoop(slotCtx, &exec.sigverifyWg, block, executionPlan, dbgOpts, shouldVerifySignatures) } - slotCtx.TotalComputeUnitsConsumed = totalComputeUnitsConsumed txLoopRegion.End() metrics.GlobalBlockReplay.TxLoop.AddTimingSince(start) + if txLoopErr != nil { + // The partially executed SlotCtx is dropped unfinalized. + return nil, fmt.Errorf("execute transactions for slot %d: %w", block.Slot, txLoopErr) + } + slotCtx.TotalComputeUnitsConsumed = totalComputeUnitsConsumed exec.txFeeAccumulator = txFeeAccumulator exec.totalCU = totalComputeUnitsConsumed diff --git a/pkg/replay/block_execution.go b/pkg/replay/block_execution.go index da6484b4..fa0f21af 100644 --- a/pkg/replay/block_execution.go +++ b/pkg/replay/block_execution.go @@ -347,7 +347,7 @@ func (exec *blockExecution) executeTransactionGroup(txs []*solana.Transaction, i exec.setReplayStage("tx_loop") start := time.Now() txLoopRegion := trace.StartRegion(exec.ctx, "TxLoop") - feeInfos, computeUnits, err := exec.runTransactionGroup(txs, execute, shouldVerifySignatures) + feeInfos, computeUnits, err := exec.runTransactionGroup(txs, execute, shouldVerifySignatures, base) txLoopRegion.End() metrics.GlobalBlockReplay.TxLoop.AddTimingSince(start) if err != nil { @@ -440,10 +440,11 @@ func (exec *blockExecution) loadTransactionAccounts(view *b.Block) error { } // runTransactionGroup is parallelTxLoop over a transaction slice with a plan -// built for the group alone (indices are group-local). Without the planner +// built for the group alone (indices are group-local; base is the group's +// block index, used only to report errors). Without the planner // (txParallelism <= 1) it runs sequentially, which // is always correct because groups are consumed in block order. -func (exec *blockExecution) runTransactionGroup(txs []*solana.Transaction, execute []bool, shouldVerifySignatures bool) ([]*fees.TxFeeInfo, []uint64, error) { +func (exec *blockExecution) runTransactionGroup(txs []*solana.Transaction, execute []bool, shouldVerifySignatures bool, base int) ([]*fees.TxFeeInfo, []uint64, error) { slotCtx := exec.slotCtx feeInfos := make([]*fees.TxFeeInfo, len(txs)) computeUnits := make([]uint64, len(txs)) @@ -472,7 +473,7 @@ func (exec *blockExecution) runTransactionGroup(txs []*solana.Transaction, execu var txErr error feeInfos[idx], computeUnits[idx], txErr = ProcessTransaction(slotCtx, &exec.sigverifyWg, tx, nil, dbgOpts, nil, shouldVerifySignatures) if feeInfos[idx] == nil { - return nil, nil, streamingTransactionError(idx, txErr) + return nil, nil, unprocessableTransactionError(base+idx, txErr) } } return feeInfos, computeUnits, nil @@ -514,19 +515,32 @@ func (exec *blockExecution) runTransactionGroup(txs []*solana.Transaction, execu <-plannerDone for idx := range txs { if execute[idx] && feeInfos[idx] == nil { - return nil, nil, streamingTransactionError(idx, txErrors[idx]) + return nil, nil, unprocessableTransactionError(base+idx, txErrors[idx]) } } return feeInfos, computeUnits, nil } -// Instruction failures still carry charged fees and remain valid block entries. -// A missing fee result means transaction admission failed: discard the bank. -func streamingTransactionError(index int, err error) error { +// UnprocessableTransactionError reports the first transaction, by block index, +// that produced no fee result. Instruction failures still carry charged fees +// and remain valid block entries. A missing fee result means transaction +// admission failed: discard the bank. +type UnprocessableTransactionError struct { + Index int + Err error +} + +func (e *UnprocessableTransactionError) Error() string { + return fmt.Sprintf("unprocessable transaction %d: %v", e.Index, e.Err) +} + +func (e *UnprocessableTransactionError) Unwrap() error { return e.Err } + +func unprocessableTransactionError(index int, err error) error { if err == nil { err = errors.New("missing fee result") } - return fmt.Errorf("unprocessable streaming transaction %d: %w", index, err) + return &UnprocessableTransactionError{Index: index, Err: err} } // finalize runs the end-of-block phases over the open SlotCtx: fees to the diff --git a/pkg/replay/block_execution_test.go b/pkg/replay/block_execution_test.go index 31556fdf..97fc51ca 100644 --- a/pkg/replay/block_execution_test.go +++ b/pkg/replay/block_execution_test.go @@ -321,6 +321,21 @@ func TestStreamingGroupRejectsUnprocessableTransactions(t *testing.T) { } } +func TestStreamingGroupReportsBlockIndex(t *testing.T) { + for _, workers := range []int{0, 4} { + env := newGroupExecutionEnv(t, workers, 10_000_000) + defer env.cleanup() + txs := transferTransactions(t, 4, 1) + require.NoError(t, env.exec.executeTransactionGroup(txs[:2], nil, false)) + txs[3].Message.RecentBlockhash = solana.Hash{0xFA} + err := env.exec.executeTransactionGroup(txs[2:], nil, false) + require.ErrorIs(t, err, TxErrInvalidBlockhash) + var unprocessable *UnprocessableTransactionError + require.ErrorAs(t, err, &unprocessable) + require.Equal(t, 3, unprocessable.Index, "workers %d", workers) + } +} + func TestStreamingGroupRejectsUnresolvedLookupsBeforeExecution(t *testing.T) { for _, workers := range []int{0, 4} { env := newGroupExecutionEnv(t, workers, 10_000_000) diff --git a/pkg/replay/streaming_lifecycle_test.go b/pkg/replay/streaming_lifecycle_test.go index 6ee885ef..c3430033 100644 --- a/pkg/replay/streaming_lifecycle_test.go +++ b/pkg/replay/streaming_lifecycle_test.go @@ -403,6 +403,49 @@ func TestStreamingLifecycleDiscardThenWholeBlock(t *testing.T) { } } +// A live block (no RPC metadata) with a transaction whose recent blockhash is +// unknown is rejected with the streaming path's error, never panics, and is +// neither finalized nor committed. +func TestProcessBlockRejectsUnprocessableTransaction(t *testing.T) { + for _, txParallelism := range []int{0, 4} { + env := newLifecycleEnv(t) + txs := transferTransactions(t, 3, 100) + txs[1].Message.RecentBlockhash = solana.Hash{0xFA} + tail := &lifecycleTail{durable: env.durable} + block := env.block(txs) + block.MarkTransactionSignaturesVerified() + txCountBefore := global.TransactionCount() + + slotCtx, err := ProcessBlock(env.acctsDb, block, env.epochSchedule, txParallelism, nil, &persistedTracker{}, tail, NewTransactionStatusCache(), false, env.parent) + require.Nil(t, slotCtx, "txpar %d", txParallelism) + require.ErrorIs(t, err, TxErrInvalidBlockhash, "txpar %d", txParallelism) + var unprocessable *UnprocessableTransactionError + require.ErrorAs(t, err, &unprocessable, "txpar %d", txParallelism) + require.Equal(t, 1, unprocessable.Index, "txpar %d", txParallelism) + require.Empty(t, tail.added, "txpar %d: a rejected bank commits nothing", txParallelism) + require.Equal(t, txCountBefore, global.TransactionCount(), "txpar %d: a rejected bank is not finalized", txParallelism) + } +} + +// A transaction that fails at the instruction level is still processed: its +// fee is charged and the block is accepted. +func TestProcessBlockChargesInstructionFailures(t *testing.T) { + for _, txParallelism := range []int{0, 4} { + env := newLifecycleEnv(t) + // 900,000 lamports stays rent-exempt (890,880) after the 5,000 fee, + // so the fee payer is valid, but the 999,001-lamport transfer fails + // in the system program for insufficient funds. + _ = env.durable.SetAccountWithoutLock(txfixture.PayerPubkey(), &accounts.Account{ + Key: txfixture.PayerPubkey(), Lamports: 900_000, Owner: addresses.SystemProgramAddr, RentEpoch: ^uint64(0), + }) + outcome := lifecycleWholeBlock(t, env, transferTransactions(t, 1, 999_000), txParallelism) + require.Equal(t, uint64(895_000), outcome.delta[txfixture.PayerPubkey()].Lamports, "txpar %d", txParallelism) + if dest, ok := outcome.delta[txfixture.DestPubkey()]; ok { + require.Equal(t, uint64(10_000_000), dest.Lamports, "txpar %d", txParallelism) + } + } +} + // V0 / address-lookup-table coverage. Resolving lookups mutates the message // object (SetAddressTables refuses a second call; ResolveLookups appends to // AccountKeys), so a stream must execute its own copies and leave the block's