Return an error for unprocessable transactions in block replay

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

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

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)

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