Return pending block instead of latest in eth_ namespace and attach pending block hash to receipts

Author: jdowning100

core/core.go

@@ -1171,6 +1171,10 @@ func (c *Core) PendingBlock() *types.WorkObject {
 	return c.sl.miner.PendingBlock()
 }
 
+func (c *Core) PendingBlockByHash(blockHash common.Hash) *types.WorkObject {
+	return c.sl.miner.worker.PendingBlockByHash(blockHash)
+}
+
 // PendingBlockAndReceipts returns the currently pending block and corresponding receipts.
 func (c *Core) PendingBlockAndReceipts() (*types.WorkObject, types.Receipts) {
 	return c.sl.miner.PendingBlockAndReceipts()

core/types/wo.go

@@ -407,6 +407,16 @@ func (wo *WorkObject) RegionStateRoot() common.Hash {
 	return wo.Header().RegionStateRoot()
 }
 
+func (wo *WorkObject) QuaiTransactions() []*Transaction {
+	quaiTxs := make([]*Transaction, 0)
+	for _, t := range wo.Transactions() {
+		if t.Type() == QuaiTxType {
+			quaiTxs = append(quaiTxs, t)
+		}
+	}
+	return quaiTxs
+}
+
 func (wo *WorkObject) QiTransactions() []*Transaction {
 	qiTxs := make([]*Transaction, 0)
 	for _, t := range wo.Transactions() {

core/worker.go

@@ -193,9 +193,10 @@ type worker struct {
 
 	logger *log.Logger
 
-	newPendingCh chan *PendingState // newPendingCh is used to send new pending state, block and receipts to the worker
-	pendingMu    sync.RWMutex
-	pendingState *PendingState
+	newPendingCh      chan *PendingState // newPendingCh is used to send new pending state, block and receipts to the worker
+	pendingBlockCache *lru.Cache[common.Hash, *types.WorkObject]
+	pendingMu         sync.RWMutex
+	pendingState      *PendingState
 }
 
 type RollingAverage struct {
@@ -282,6 +283,7 @@ func newWorker(config *Config, chainConfig *params.ChainConfig, db ethdb.Databas
 	if headerchain.ProcessingState() && nodeCtx == common.ZONE_CTX {
 		worker.chainSideSub = worker.hc.SubscribeChainSideEvent(worker.chainSideCh)
 		worker.wg.Add(1)
+		worker.pendingBlockCache, _ = lru.New[common.Hash, *types.WorkObject](20)
 		go worker.asyncStateLoop()
 		go worker.PendingStateCache()
 	}
@@ -378,6 +380,11 @@ func (w *worker) enablePreseal() {
 	atomic.StoreUint32(&w.noempty, 0)
 }
 
+func (w *worker) PendingBlockByHash(blockHash common.Hash) *types.WorkObject {
+	pendingBlock, _ := w.pendingBlockCache.Get(blockHash)
+	return pendingBlock
+}
+
 // pending returns the pending state and corresponding block.
 func (w *worker) pending() *types.WorkObject {
 	w.pendingMu.RLock()
@@ -452,6 +459,16 @@ func (w *worker) PendingStateCache() {
 			w.pendingMu.Lock()
 			w.pendingState = newPendingState
 			w.pendingMu.Unlock()
+			w.pendingBlockCache.Add(newPendingState.pendingBlock.Hash(), newPendingState.pendingBlock)
+			for _, r := range newPendingState.pendingReceipts {
+				if r.Type == types.QuaiTxType {
+					w.logger.WithFields(log.Fields{
+						"txHash":      r.TxHash,
+						"blockHash":   newPendingState.pendingBlock.Hash(),
+						"blockNumber": newPendingState.pendingBlock.NumberU64(common.ZONE_CTX),
+					}).Info("New receipt in pending block")
+				}
+			}
 		case <-w.exitCh:
 			return
 		}
@@ -528,7 +545,7 @@ func (w *worker) asyncStateLoop() {
 	}()
 	defer w.wg.Done() // decrement the wait group after the close of the loop
 
-	ticker := time.NewTicker(2 * time.Second)
+	ticker := time.NewTicker(minRecommitInterval)
 	var prevHeader *types.WorkObject = nil
 	defer ticker.Stop()
 	for {

internal/quaiapi/api.go

@@ -365,15 +365,18 @@ func (s *PublicBlockChainAPI) GetHeaderByHash(ctx context.Context, hash common.H
 //   - When fullTx is true all transactions in the block are returned, otherwise
 //     only the transaction hash is returned.
 func (s *PublicBlockChainAPI) GetBlockByNumber(ctx context.Context, number rpc.BlockNumber, fullTx bool) (map[string]interface{}, error) {
+	if number == rpc.LatestBlockNumber && s.b.UsePendingState() {
+		number = rpc.PendingBlockNumber // Use pending state for API queries if latest is requested and flag is set
+	}
 	block, err := s.b.BlockByNumber(ctx, number)
 	if block != nil && err == nil {
 		response, err := RPCMarshalETHBlock(block, true, fullTx, s.b.NodeLocation())
-		if err == nil && number == rpc.PendingBlockNumber {
+		/*if err == nil && number == rpc.PendingBlockNumber {
 			// Pending blocks need to nil out a few fields
 			for _, field := range []string{"hash", "nonce", "miner"} {
 				response[field] = nil
 			}
-		}
+		}*/
 		return response, err
 	}
 	return nil, err
@@ -383,6 +386,9 @@ func (s *PublicBlockChainAPI) GetBlockByNumber(ctx context.Context, number rpc.B
 // detail, otherwise only the transaction hash is returned.
 func (s *PublicBlockChainAPI) GetBlockByHash(ctx context.Context, hash common.Hash, fullTx bool) (map[string]interface{}, error) {
 	block, err := s.b.BlockByHash(ctx, hash)
+	if block == nil && s.b.UsePendingState() {
+		block = s.b.PendingBlockByHash(hash)
+	}
 	if block != nil {
 		return RPCMarshalETHBlock(block, true, fullTx, s.b.NodeLocation())
 	}
@@ -963,7 +969,7 @@ func RPCMarshalETHBlock(block *types.WorkObject, inclTx bool, fullTx bool, nodeL
 				return newRPCTransactionFromBlockHash(block, tx.Hash(), false, nodeLocation), nil
 			}
 		}
-		txs := block.Transactions()
+		txs := block.QuaiTransactions()
 		transactions := make([]interface{}, len(txs))
 		var err error
 		for i, tx := range txs {
@@ -1303,14 +1309,13 @@ func (s *PublicBlockChainAPI) GetTransactionReceipt(ctx context.Context, hash co
 			}
 			usePending = true
 			blockNumber = s.b.CurrentHeader().NumberU64(s.b.NodeCtx()) + 1
-			blockHash = common.Hash{}
 		} else {
 			return nil, nil
 		}
 	}
 	receipt := &types.Receipt{}
 	if usePending {
-		receipt = s.b.GetPendingReceipt(hash)
+		receipt, blockHash = s.b.GetPendingReceipt(hash)
 		if receipt == nil {
 			return nil, nil
 		}

internal/quaiapi/backend.go

@@ -69,7 +69,8 @@ type Backend interface {
 	GetOutpointsByAddressAndRange(ctx context.Context, address common.Address, start, end uint32) ([]*types.OutpointAndDenomination, error)
 	UTXOsByAddress(ctx context.Context, address common.Address) ([]*types.UtxoEntry, error)
 	GetReceipts(ctx context.Context, hash common.Hash) (types.Receipts, error)
-	GetPendingReceipt(txHash common.Hash) *types.Receipt
+	GetPendingReceipt(txHash common.Hash) (*types.Receipt, common.Hash)
+	PendingBlockByHash(blockHash common.Hash) *types.WorkObject
 	GetEVM(ctx context.Context, msg core.Message, state *state.StateDB, header *types.WorkObject, parent *types.WorkObject, vmConfig *vm.Config) (*vm.EVM, func() error, error)
 	SetCurrentExpansionNumber(expansionNumber uint8)
 	SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription

internal/quaiapi/quai_api.go

@@ -1507,14 +1507,13 @@ func (s *PublicBlockChainQuaiAPI) GetTransactionReceipt(ctx context.Context, has
 			}
 			usePending = true
 			blockNumber = s.b.CurrentHeader().NumberU64(s.b.NodeCtx()) + 1
-			blockHash = common.Hash{}
 		} else {
 			return nil, nil
 		}
 	}
 	receipt := &types.Receipt{}
 	if usePending {
-		receipt = s.b.GetPendingReceipt(hash)
+		receipt, blockHash = s.b.GetPendingReceipt(hash)
 		if receipt == nil {
 			return nil, nil
 		}

quai/api_backend.go

@@ -238,14 +238,18 @@ func (b *QuaiAPIBackend) GetReceipts(ctx context.Context, hash common.Hash) (typ
 	return b.quai.core.GetReceiptsByHash(hash), nil
 }
 
-func (b *QuaiAPIBackend) GetPendingReceipt(txHash common.Hash) *types.Receipt {
-	pendingReceipts := b.quai.core.PendingReceipts()
+func (b *QuaiAPIBackend) GetPendingReceipt(txHash common.Hash) (*types.Receipt, common.Hash) {
+	pendingBlock, pendingReceipts := b.quai.core.PendingBlockAndReceipts()
 	for _, receipt := range pendingReceipts {
 		if receipt.TxHash == txHash {
-			return receipt
+			return receipt, pendingBlock.Hash()
 		}
 	}
-	return nil
+	return nil, common.Hash{}
+}
+
+func (b *QuaiAPIBackend) PendingBlockByHash(blockHash common.Hash) *types.WorkObject {
+	return b.quai.core.PendingBlockByHash(blockHash)
 }
 
 func (b *QuaiAPIBackend) GetPrimeBlock(blockHash common.Hash) *types.WorkObject {