core, plugins: add exex reorg support and access to receipts

core/blockchain.go

@@ -1565,8 +1565,8 @@ func (bc *BlockChain) writeBlockAndSetHead(block *types.Block, receipts []*types
 	// canonical blocks. Avoid firing too many ChainHeadEvents,
 	// we will fire an accumulated ChainHeadEvent and disable fire
 	// event here.
+	exex.TriggerHeadHook(block.Header())
 	if emitHeadEvent {
-		exex.TriggerHeadHook(block.Header())
 		bc.chainHeadFeed.Send(ChainHeadEvent{Header: block.Header()})
 	}
 	return CanonStatTy, nil
@@ -1632,7 +1632,6 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool, makeWitness
 	// Fire a single chain head event if we've progressed the chain
 	defer func() {
 		if lastCanon != nil && bc.CurrentBlock().Hash() == lastCanon.Hash() {
-			exex.TriggerHeadHook(lastCanon.Header())
 			bc.chainHeadFeed.Send(ChainHeadEvent{Header: lastCanon.Header()})
 		}
 	}()
@@ -2262,6 +2261,10 @@ func (bc *BlockChain) reorg(oldHead *types.Header, newHead *types.Header) error
 		// rewind the canonical chain to a lower point.
 		log.Error("Impossible reorg, please file an issue", "oldnum", oldHead.Number, "oldhash", oldHead.Hash(), "oldblocks", len(oldChain), "newnum", newHead.Number, "newhash", newHead.Hash(), "newblocks", len(newChain))
 	}
+	// Trigger revertal reorgs
+	if len(oldChain) > 0 {
+		exex.TriggerReorgHook(oldChain, true)
+	}
 	// Acquire the tx-lookup lock before mutation. This step is essential
 	// as the txlookups should be changed atomically, and all subsequent
 	// reads should be blocked until the mutation is complete.
@@ -2369,6 +2372,12 @@ func (bc *BlockChain) reorg(oldHead *types.Header, newHead *types.Header) error
 	// Release the tx-lookup lock after mutation.
 	bc.txLookupLock.Unlock()
 
+	// Trigger application reorgs
+	if len(newChain) > 1 {
+		slices.Reverse(newChain)
+		exex.TriggerReorgHook(newChain[:len(newChain)-1], false)
+		slices.Reverse(newChain)
+	}
 	return nil
 }
 

core/blockchain_reader.go

@@ -209,6 +209,27 @@ func (bc *BlockChain) GetBlocksFromHash(hash common.Hash, n int) (blocks []*type
 	return
 }
 
+// GetReceiptsByNumber retrieves the receipts for all transactions in a given block.
+func (bc *BlockChain) GetReceiptsByNumber(number uint64) types.Receipts {
+	hash := rawdb.ReadCanonicalHash(bc.db, number)
+	if hash == (common.Hash{}) {
+		return nil
+	}
+	if receipts, ok := bc.receiptsCache.Get(hash); ok {
+		return receipts
+	}
+	header := bc.GetHeader(hash, number)
+	if header == nil {
+		return nil
+	}
+	receipts := rawdb.ReadReceipts(bc.db, hash, number, header.Time, bc.chainConfig)
+	if receipts == nil {
+		return nil
+	}
+	bc.receiptsCache.Add(hash, receipts)
+	return receipts
+}
+
 // GetReceiptsByHash retrieves the receipts for all transactions in a given block.
 func (bc *BlockChain) GetReceiptsByHash(hash common.Hash) types.Receipts {
 	if receipts, ok := bc.receiptsCache.Get(hash); ok {

core/exex/exex.go

@@ -47,4 +47,5 @@ type PluginV1 struct {
 	OnInit  InitHook  // Called when the chain gets initialized within Geth
 	OnClose CloseHook // Called when the chain gets torn down within Geth
 	OnHead  HeadHook  // Called when the chain head block is updated in Geth
+	OnReorg ReorgHook // Called wnen the chain reorgs to a sidechain within Geth
 }

core/exex/exex/adapter_chain.go

@@ -32,6 +32,7 @@ type gethChain interface {
 	GetHeaderByNumber(number uint64) *types.Header
 	GetBlockByNumber(number uint64) *types.Block
 	StateAt(root common.Hash) (*state.StateDB, error)
+	GetReceiptsByNumber(number uint64) types.Receipts
 }
 
 // chainAdapter is an adapter to convert Geth's internal blockchain (unstable
@@ -76,3 +77,17 @@ func (a *chainAdapter) State(root common.Hash) exex.State {
 	}
 	return wrapState(state)
 }
+
+// Receipts retrieves a set of receits belonging to all transactions within
+// a block from the canonical chain. Receipts on side-chains are not exposed
+// by the Chain interface.
+func (a *chainAdapter) Receipts(number uint64) []*types.Receipt {
+	// Receipts have public fields, copy to prevent modification
+	receipts := a.chain.GetReceiptsByNumber(number)
+
+	copies := make([]*types.Receipt, 0, len(receipts))
+	for _, receipt := range receipts {
+		copies = append(copies, types.CopyReceipt(receipt))
+	}
+	return copies
+}

core/exex/exex/registry.go

@@ -38,6 +38,7 @@ type registry interface {
 	TriggerInitHook(chain exex.Chain)
 	TriggerCloseHook()
 	TriggerHeadHook(head *types.Header)
+	TriggerReorgHook(headers []*types.Header, revert bool)
 }
 
 // Plugins returns a list of all registered plugins to generate CLI flags.
@@ -64,3 +65,8 @@ func TriggerCloseHook() {
 func TriggerHeadHook(head *types.Header) {
 	globalRegistry.TriggerHeadHook(head)
 }
+
+// TriggerReorgHook triggers the OnReorg hook in exex plugins.
+func TriggerReorgHook(headers []*types.Header, revert bool) {
+	globalRegistry.TriggerReorgHook(headers, revert)
+}

core/exex/hooks.go

@@ -31,8 +31,23 @@ type CloseHook = func()
 //   - In sync, this will be called on fork-choice updates
 type HeadHook = func(head *types.Header)
 
-// TODO(karalabe): This is interesting. We need to keep events in sync with the
-// user's chain access capabilities. We either need to provide side-chain access,
-// which gets nasty fast; or we need to reorg in lockstep; or we need two events
-// one to start a reorg (going back) and one having finished (going forward).
-// type ReorgHook = func(old, new *types.Header) error
+// ReorgHook is called when the chain head is updated to a different parent than
+// the previous head. In this case previously applied state changes need to be
+// rolled back, and state changes from a sidechain need to be applied.
+//
+// This method is called with a set of header being operated on and the direction
+// of the operation, usually both directions being called one after the other:
+//
+//   - If revert == true, the given headers are being rolled back, they are in
+//     reverse order, headers[0] being the previous chain head, and the last item
+//     being the olders block getting undone.
+//   - If revert == false, the given headers are being applied after the rollback,
+//     they are in forward order, headers[0] being the oldest block being applied
+//     and the last item being the newest getting applied. Note, the chain head
+//     that triggered the reorg will arrive in the HeadHook.
+//
+// The reason the reorg event it "emitted" in two parts is for both operations to
+// have access to a unified singletone view of the chain. An alternative would be
+// to pass in both the reverted and applied headers at the same time, but that
+// would require chain accessorts to support sidechains, which complicate APIs.
+type ReorgHook = func(headers []*types.Header, revert bool)

core/exex/interface_chain.go

@@ -39,4 +39,9 @@ type Chain interface {
 
 	// State retrieves a state accessor at a given root hash.
 	State(root common.Hash) State
+
+	// Receipts retrieves a set of receits belonging to all transactions within
+	// a block from the canonical chain. Receipts on side-chains are not exposed
+	// by the Chain interface.
+	Receipts(number uint64) []*types.Receipt
 }

core/exex/registry_internal.go

@@ -78,3 +78,12 @@ func (reg *registry) TriggerHeadHook(head *types.Header) {
 		}
 	}
 }
+
+// TriggerReorgHook triggers the OnReorg hook in exex plugins.
+func (reg *registry) TriggerReorgHook(headers []*types.Header, revert bool) {
+	for _, plugin := range globalRegistry.pluginsV1 {
+		if plugin.OnReorg != nil {
+			plugin.OnReorg(headers, revert)
+		}
+	}
+}

core/types/log.go

@@ -53,6 +53,20 @@ type Log struct {
 	Removed bool `json:"removed" rlp:"-"`
 }
 
+// CopyLog creates a deep copy of a log.
+func CopyLog(l *Log) *Log {
+	cpy := *l
+	if len(l.Topics) > 0 {
+		cpy.Topics = make([]common.Hash, len(l.Topics))
+		copy(cpy.Topics, l.Topics)
+	}
+	if len(l.Data) > 0 {
+		cpy.Data = make([]byte, len(l.Data))
+		copy(cpy.Data, l.Data)
+	}
+	return &cpy
+}
+
 type logMarshaling struct {
 	Data        hexutil.Bytes
 	BlockNumber hexutil.Uint64

core/types/receipt.go

@@ -117,6 +117,31 @@ func NewReceipt(root []byte, failed bool, cumulativeGasUsed uint64) *Receipt {
 	return r
 }
 
+// CopyReceipt creates a deep copy of a receipt.
+func CopyReceipt(r *Receipt) *Receipt {
+	cpy := *r
+	if len(r.PostState) > 0 {
+		cpy.PostState = make([]byte, len(r.PostState))
+		copy(cpy.PostState, r.PostState)
+	}
+	if len(r.Logs) > 0 {
+		cpy.Logs = make([]*Log, len(r.Logs))
+		for i, log := range r.Logs {
+			cpy.Logs[i] = CopyLog(log)
+		}
+	}
+	if r.EffectiveGasPrice != nil {
+		cpy.EffectiveGasPrice = new(big.Int).Set(r.EffectiveGasPrice)
+	}
+	if r.BlobGasPrice != nil {
+		cpy.BlobGasPrice = new(big.Int).Set(r.BlobGasPrice)
+	}
+	if r.BlockNumber != nil {
+		cpy.BlockNumber = new(big.Int).Set(r.BlockNumber)
+	}
+	return &cpy
+}
+
 // EncodeRLP implements rlp.Encoder, and flattens the consensus fields of a receipt
 // into an RLP stream. If no post state is present, byzantium fork is assumed.
 func (r *Receipt) EncodeRLP(w io.Writer) error {

plugins/minimal.go

@@ -33,5 +33,12 @@ func newMinimalPlugin(config *exex.ConfigV1) (*exex.PluginV1, error) {
 		OnHead: func(head *types.Header) {
 			config.Logger.Info("Chain head updated", "number", head.Number, "hash", head.Hash())
 		},
+		OnReorg: func(headers []*types.Header, revert bool) {
+			if revert {
+				config.Logger.Warn("Reorging blocks out", "count", len(headers))
+			} else {
+				config.Logger.Warn("Reorging blocks in", "count", len(headers))
+			}
+		},
 	}, nil
 }