fix streaming in error case (#4861)

* save * save
2024-12-23 04:03:49 +00:00 · 2022-07-29 11:08:04 +07:00 · 2022-07-29 11:08:04 +07:00 · deccbf973a
commit deccbf973a
parent 2681ee392a
4 changed files with 7 additions and 28 deletions
--- a/cmd/rpcdaemon/commands/trace_filtering.go
+++ b/cmd/rpcdaemon/commands/trace_filtering.go
@ -300,6 +300,7 @@ func (api *TraceAPIImpl) Filter(ctx context.Context, req TraceFilterRequest, str

 	var json = jsoniter.ConfigCompatibleWithStandardLibrary
 	stream.WriteArrayStart()
+	defer stream.WriteArrayEnd()
 	first := true
 	// Execute all transactions in picked blocks

@ -320,17 +321,14 @@ func (api *TraceAPIImpl) Filter(ctx context.Context, req TraceFilterRequest, str
 		// Extract transactions from block
 		hash, hashErr := rawdb.ReadCanonicalHash(dbtx, b)
 		if hashErr != nil {
-			stream.WriteNil()
 			return hashErr
 		}

 		block, bErr := api.blockWithSenders(dbtx, hash, b)
 		if bErr != nil {
-			stream.WriteNil()
 			return bErr
 		}
 		if block == nil {
-			stream.WriteNil()
 			return fmt.Errorf("could not find block %x %d", hash, b)
 		}

@ -339,7 +337,6 @@ func (api *TraceAPIImpl) Filter(ctx context.Context, req TraceFilterRequest, str
 		txs := block.Transactions()
 		t, tErr := api.callManyTransactions(ctx, dbtx, txs, []string{TraceTypeTrace}, block.ParentHash(), rpc.BlockNumber(block.NumberU64()-1), block.Header(), -1 /* all tx indices */, types.MakeSigner(chainConfig, b), chainConfig.Rules(b))
 		if tErr != nil {
-			stream.WriteNil()
 			return tErr
 		}
 		includeAll := len(fromAddresses) == 0 && len(toAddresses) == 0
@ -356,7 +353,6 @@ func (api *TraceAPIImpl) Filter(ctx context.Context, req TraceFilterRequest, str
 					pt.TransactionPosition = &txPosition
 					b, err := json.Marshal(pt)
 					if err != nil {
-						stream.WriteNil()
 						return err
 					}
 					if nSeen > after && nExported < count {
@ -388,7 +384,6 @@ func (api *TraceAPIImpl) Filter(ctx context.Context, req TraceFilterRequest, str
 			tr.TraceAddress = []int{}
 			b, err := json.Marshal(tr)
 			if err != nil {
-				stream.WriteNil()
 				return err
 			}
 			if nSeen > after && nExported < count {
@ -419,7 +414,6 @@ func (api *TraceAPIImpl) Filter(ctx context.Context, req TraceFilterRequest, str
 					tr.TraceAddress = []int{}
 					b, err := json.Marshal(tr)
 					if err != nil {
-						stream.WriteNil()
 						return err
 					}
 					if nSeen > after && nExported < count {
@ -435,8 +429,7 @@ func (api *TraceAPIImpl) Filter(ctx context.Context, req TraceFilterRequest, str
 			}
 		}
 	}
-	stream.WriteArrayEnd()
-	return stream.Flush()
+	return nil
 }

 func filter_trace(pt *ParityTrace, fromAddresses map[common.Address]struct{}, toAddresses map[common.Address]struct{}) bool {
--- a/cmd/rpcdaemon/commands/tracing.go
+++ b/cmd/rpcdaemon/commands/tracing.go
@ -85,6 +85,7 @@ func (api *PrivateDebugAPIImpl) traceBlock(ctx context.Context, blockNrOrHash rp
 	signer := types.MakeSigner(chainConfig, block.NumberU64())
 	rules := chainConfig.Rules(block.NumberU64())
 	stream.WriteArrayStart()
+	defer stream.WriteArrayEnd()
 	for idx, tx := range block.Transactions() {
 		select {
 		default:
@ -107,8 +108,6 @@ func (api *PrivateDebugAPIImpl) traceBlock(ctx context.Context, blockNrOrHash rp
 		}
 		stream.Flush()
 	}
-	stream.WriteArrayEnd()
-	stream.Flush()
 	return nil
 }

@ -403,6 +402,7 @@ func (api *PrivateDebugAPIImpl) TraceCallMany(ctx context.Context, bundles []Bun
 	}

 	stream.WriteArrayStart()
+	defer stream.WriteArrayEnd()
 	for bundle_index, bundle := range bundles {
 		stream.WriteArrayStart()
 		// first change blockContext
@ -420,9 +420,7 @@ func (api *PrivateDebugAPIImpl) TraceCallMany(ctx context.Context, bundles []Bun
 			ibs := evm.IntraBlockState().(*state.IntraBlockState)
 			ibs.Prepare(common.Hash{}, parent.Hash(), txn_index)
 			err = transactions.TraceTx(ctx, msg, blockCtx, txCtx, evm.IntraBlockState(), config, chainConfig, stream)
-
 			if err != nil {
-				stream.WriteNil()
 				return err
 			}

@ -438,6 +436,5 @@ func (api *PrivateDebugAPIImpl) TraceCallMany(ctx context.Context, bundles []Bun
 		blockCtx.BlockNumber++
 		blockCtx.Time++
 	}
-	stream.WriteArrayEnd()
 	return nil
 }
--- a/cmd/rpcdaemon22/commands/trace_filtering.go
+++ b/cmd/rpcdaemon22/commands/trace_filtering.go
@ -300,6 +300,7 @@ func (api *TraceAPIImpl) Filter(ctx context.Context, req TraceFilterRequest, str

 	var json = jsoniter.ConfigCompatibleWithStandardLibrary
 	stream.WriteArrayStart()
+	defer stream.WriteArrayEnd()
 	first := true
 	// Execute all transactions in picked blocks

@ -330,7 +331,6 @@ func (api *TraceAPIImpl) Filter(ctx context.Context, req TraceFilterRequest, str
 		}))
 		if blockNum > lastBlockNum {
 			if lastHeader, err = api._blockReader.HeaderByNumber(ctx, nil, blockNum); err != nil {
-				stream.WriteNil()
 				return err
 			}
 			lastBlockNum = blockNum
@ -341,7 +341,6 @@ func (api *TraceAPIImpl) Filter(ctx context.Context, req TraceFilterRequest, str
 		if txNum+1 == api._txNums[blockNum] {
 			body, err := api._blockReader.Body(ctx, nil, lastBlockHash, blockNum)
 			if err != nil {
-				stream.WriteNil()
 				return err
 			}
 			// Block reward section, handle specially
@ -362,7 +361,6 @@ func (api *TraceAPIImpl) Filter(ctx context.Context, req TraceFilterRequest, str
 				tr.TraceAddress = []int{}
 				b, err := json.Marshal(tr)
 				if err != nil {
-					stream.WriteNil()
 					return err
 				}
 				if nSeen > after && nExported < count {
@ -393,7 +391,6 @@ func (api *TraceAPIImpl) Filter(ctx context.Context, req TraceFilterRequest, str
 						tr.TraceAddress = []int{}
 						b, err := json.Marshal(tr)
 						if err != nil {
-							stream.WriteNil()
 							return err
 						}
 						if nSeen > after && nExported < count {
@ -418,13 +415,11 @@ func (api *TraceAPIImpl) Filter(ctx context.Context, req TraceFilterRequest, str
 		//fmt.Printf("txNum=%d, blockNum=%d, txIndex=%d\n", txNum, blockNum, txIndex)
 		txn, err := api._txnReader.TxnByIdxInBlock(ctx, nil, blockNum, int(txIndex))
 		if err != nil {
-			stream.WriteNil()
 			return err
 		}
 		txHash := txn.Hash()
 		msg, err := txn.AsMessage(*lastSigner, lastHeader.BaseFee, lastRules)
 		if err != nil {
-			stream.WriteNil()
 			return err
 		}
 		contractHasTEVM := func(contractHash common.Hash) (bool, error) { return false, nil }
@ -451,16 +446,13 @@ func (api *TraceAPIImpl) Filter(ctx context.Context, req TraceFilterRequest, str
 		var execResult *core.ExecutionResult
 		execResult, err = core.ApplyMessage(evm, msg, gp, true /* refunds */, false /* gasBailout */)
 		if err != nil {
-			stream.WriteNil()
 			return err
 		}
 		traceResult.Output = common.CopyBytes(execResult.ReturnData)
 		if err = ibs.FinalizeTx(evm.ChainRules(), noop); err != nil {
-			stream.WriteNil()
 			return err
 		}
 		if err = ibs.CommitBlock(evm.ChainRules(), cachedWriter); err != nil {
-			stream.WriteNil()
 			return err
 		}
 		for _, pt := range traceResult.Trace {
@ -472,7 +464,6 @@ func (api *TraceAPIImpl) Filter(ctx context.Context, req TraceFilterRequest, str
 				pt.TransactionPosition = &txIndex
 				b, err := json.Marshal(pt)
 				if err != nil {
-					stream.WriteNil()
 					return err
 				}
 				if nSeen > after && nExported < count {
@ -487,8 +478,7 @@ func (api *TraceAPIImpl) Filter(ctx context.Context, req TraceFilterRequest, str
 			}
 		}
 	}
-	stream.WriteArrayEnd()
-	return stream.Flush()
+	return nil
 }

 func filter_trace(pt *ParityTrace, fromAddresses map[common.Address]struct{}, toAddresses map[common.Address]struct{}) bool {
--- a/cmd/rpcdaemon22/commands/tracing.go
+++ b/cmd/rpcdaemon22/commands/tracing.go
@ -80,6 +80,7 @@ func (api *PrivateDebugAPIImpl) traceBlock(ctx context.Context, blockNrOrHash rp
 	signer := types.MakeSigner(chainConfig, block.NumberU64())
 	rules := chainConfig.Rules(block.NumberU64())
 	stream.WriteArrayStart()
+	defer stream.WriteArrayEnd()
 	for idx, tx := range block.Transactions() {
 		select {
 		default:
@ -102,8 +103,6 @@ func (api *PrivateDebugAPIImpl) traceBlock(ctx context.Context, blockNrOrHash rp
 		}
 		stream.Flush()
 	}
-	stream.WriteArrayEnd()
-	stream.Flush()
 	return nil
 }