Merge pull request #6384 from oasisprotocol/peternose/trivial/simplify-pruner

go/worker/compute/executor: Simplify prune handler

Author: peternose

.changelog/6384.trivial.md

@@ -79,11 +79,6 @@ func (n *Node) Initialized() <-chan struct{} {
 	return n.initCh
 }
 
-// HandleNewBlockEarlyLocked is guarded by CrossNode.
-func (n *Node) HandleNewBlockEarlyLocked(*runtime.BlockInfo) {
-	// Nothing to do here.
-}
-
 // HandleNewBlockLocked is guarded by CrossNode.
 func (n *Node) HandleNewBlockLocked(*runtime.BlockInfo) {
 	// Nothing to do here.

go/worker/client/committee/node.go

@@ -128,8 +128,6 @@ var (
 // NodeHooks defines a worker's duties at common events.
 // These are called from the runtime's common node's worker.
 type NodeHooks interface {
-	// Guarded by CrossNode.
-	HandleNewBlockEarlyLocked(*runtime.BlockInfo)
 	// Guarded by CrossNode.
 	HandleNewBlockLocked(*runtime.BlockInfo)
 	// Guarded by CrossNode.
@@ -185,12 +183,11 @@ type Node struct {
 
 	// Mutable and shared between nodes' workers.
 	// Guarded by .CrossNode.
-	CrossNode             sync.Mutex
-	CurrentBlock          *block.Block
-	CurrentBlockHeight    int64
-	CurrentConsensusBlock *consensus.LightBlock
-	CurrentDescriptor     *registry.Runtime
-	CurrentEpoch          beacon.EpochTime
+	CrossNode          sync.Mutex
+	CurrentBlock       *block.Block
+	CurrentBlockHeight int64
+	CurrentDescriptor  *registry.Runtime
+	CurrentEpoch       beacon.EpochTime
 
 	logger *logging.Logger
 }
@@ -413,6 +410,13 @@ func (n *Node) handleSuspendLocked(int64) {
 	}
 }
 
+func (n *Node) updateHostedRuntimeVersion() {
+	n.CrossNode.Lock()
+	defer n.CrossNode.Unlock()
+
+	n.updateHostedRuntimeVersionLocked()
+}
+
 func (n *Node) updateHostedRuntimeVersionLocked() {
 	if n.CurrentDescriptor == nil {
 		return
@@ -465,8 +469,6 @@ func (n *Node) handleNewBlock(blk *block.Block, height int64) {
 
 	processedBlockCount.With(n.getMetricLabels()).Inc()
 
-	header := blk.Header
-
 	// The first received block will be treated an epoch transition (if valid).
 	// This will refresh the committee on the first block,
 	// instead of waiting for the next epoch transition to occur.
@@ -487,10 +489,9 @@ func (n *Node) handleNewBlock(blk *block.Block, height int64) {
 	// Update the current block.
 	n.CurrentBlock = blk
 	n.CurrentBlockHeight = height
-	n.CurrentConsensusBlock = consensusBlk
 
 	// Update active descriptor on epoch transitions.
-	if firstBlockReceived || header.HeaderType == block.EpochTransition || header.HeaderType == block.Suspended {
+	if firstBlockReceived || blk.Header.HeaderType == block.EpochTransition || blk.Header.HeaderType == block.Suspended {
 		var rs *roothash.RuntimeState
 		rs, err = n.Consensus.RootHash().GetRuntimeState(n.ctx, &roothash.RuntimeRequest{
 			RuntimeID: n.Runtime.ID(),
@@ -524,19 +525,8 @@ func (n *Node) handleNewBlock(blk *block.Block, height int64) {
 		n.KeyManagerClient.SetKeyManagerID(n.CurrentDescriptor.KeyManager)
 	}
 
-	bi := &runtime.BlockInfo{
-		RuntimeBlock:     n.CurrentBlock,
-		ConsensusBlock:   n.CurrentConsensusBlock,
-		Epoch:            n.CurrentEpoch,
-		ActiveDescriptor: n.CurrentDescriptor,
-	}
-
-	for _, hooks := range n.hooks {
-		hooks.HandleNewBlockEarlyLocked(bi)
-	}
-
 	// Perform actions based on block type.
-	switch header.HeaderType {
+	switch blk.Header.HeaderType {
 	case block.Normal:
 		if firstBlockReceived {
 			n.logger.Warn("forcing an epoch transition on first received block")
@@ -559,11 +549,18 @@ func (n *Node) handleNewBlock(blk *block.Block, height int64) {
 		n.handleSuspendLocked(height)
 	default:
 		n.logger.Error("invalid block type",
-			"block", bi.RuntimeBlock,
+			"block", blk,
 		)
 		return
 	}
 
+	bi := &runtime.BlockInfo{
+		RuntimeBlock:     n.CurrentBlock,
+		ConsensusBlock:   consensusBlk,
+		Epoch:            n.CurrentEpoch,
+		ActiveDescriptor: n.CurrentDescriptor,
+	}
+
 	n.TxPool.ProcessBlock(bi)
 
 	// Fetch incoming messages.
@@ -729,9 +726,7 @@ func (n *Node) worker() {
 
 	// Perform initial hosted runtime version update to ensure we have something even in cases where
 	// initial block processing fails for any reason.
-	n.CrossNode.Lock()
-	n.updateHostedRuntimeVersionLocked()
-	n.CrossNode.Unlock()
+	n.updateHostedRuntimeVersion()
 
 	// Start the runtime.
 	hrt := n.GetHostedRuntime()
@@ -779,11 +774,7 @@ func (n *Node) worker() {
 					return
 				}
 
-				func() {
-					n.CrossNode.Lock()
-					defer n.CrossNode.Unlock()
-					n.updateHostedRuntimeVersionLocked()
-				}()
+				n.updateHostedRuntimeVersion()
 			case compNotify.Removed != nil:
 				// Received removal of a component.
 				if err := n.RemoveHostedRuntimeComponent(*compNotify.Removed); err != nil {

go/worker/common/committee/node.go

@@ -1,26 +1,19 @@
 package committee
 
 import (
-	"github.com/oasisprotocol/oasis-core/go/common/crash"
 	runtime "github.com/oasisprotocol/oasis-core/go/runtime/api"
 	"github.com/oasisprotocol/oasis-core/go/worker/common/committee"
 )
 
 // Ensure Node implements NodeHooks.
 var _ committee.NodeHooks = (*Node)(nil)
 
-// HandleNewBlockEarlyLocked implements NodeHooks.
+// HandleNewBlockLocked implements NodeHooks.
 // Guarded by n.commonNode.CrossNode.
-func (n *Node) HandleNewBlockEarlyLocked(*runtime.BlockInfo) {
-	crash.Here(crashPointRoothashReceiveAfter)
-
+func (n *Node) HandleNewBlockLocked(bi *runtime.BlockInfo) {
 	// Update our availability.
 	n.nudgeAvailabilityLocked(false)
-}
 
-// HandleNewBlockLocked implements NodeHooks.
-// Guarded by n.commonNode.CrossNode.
-func (n *Node) HandleNewBlockLocked(bi *runtime.BlockInfo) {
 	// Drop blocks if the worker falls behind.
 	select {
 	case <-n.blockInfoCh:

go/worker/compute/executor/committee/hooks.go

@@ -11,7 +11,6 @@ const (
 	crashPointBatchProposeBefore       = "worker.executor.batch.propose.before"
 	crashPointBatchProposeAfter        = "worker.executor.batch.propose.after"
 	crashPointDiscrepancyDetectedAfter = "worker.executor.batch.discrepancy_detected.after"
-	crashPointRoothashReceiveAfter     = "worker.executor.batch.roothash.receive.after"
 	crashPointBatchPublishAfter        = "worker.executor.batch.schedule.publish.after"
 )
 
@@ -23,7 +22,6 @@ func init() {
 		crashPointBatchProposeBefore,
 		crashPointBatchProposeAfter,
 		crashPointDiscrepancyDetectedAfter,
-		crashPointRoothashReceiveAfter,
 		crashPointBatchPublishAfter,
 	)
 }

go/worker/compute/executor/committee/init.go

@@ -1568,9 +1568,6 @@ func NewNode(
 		logger:           logging.GetLogger("worker/executor/committee").With("runtime_id", commonNode.Runtime.ID()),
 	}
 
-	// Register prune handler.
-	commonNode.Runtime.History().Pruner().RegisterHandler(&pruneHandler{commonNode: commonNode})
-
 	// Register committee message handler.
 	commonNode.P2P.RegisterHandler(committeeTopic, &committeeMsgHandler{n})
 

go/worker/compute/executor/committee/node.go

@@ -4,29 +4,36 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/oasisprotocol/oasis-core/go/common"
+	consensus "github.com/oasisprotocol/oasis-core/go/consensus/api"
 	roothash "github.com/oasisprotocol/oasis-core/go/roothash/api"
-	"github.com/oasisprotocol/oasis-core/go/worker/common/committee"
 )
 
-// pruneHandler is a prune handler that prevents pruning of the last normal round.
-type pruneHandler struct {
-	commonNode *committee.Node
+// PruneHandler is a prune handler that prevents pruning of the last normal round.
+type PruneHandler struct {
+	runtimeID common.Namespace
+	consensus consensus.Service
 }
 
-func (p *pruneHandler) Prune(rounds []uint64) error {
-	p.commonNode.CrossNode.Lock()
-	height := p.commonNode.CurrentBlockHeight
-	p.commonNode.CrossNode.Unlock()
+// NewPruneHandler creates a new prune handler.
+func NewPruneHandler(runtimeID common.Namespace, consensus consensus.Service) *PruneHandler {
+	return &PruneHandler{
+		runtimeID: runtimeID,
+		consensus: consensus,
+	}
+}
 
+// Prune verifies that no rounds beyond the last normal round are pruned.
+func (p *PruneHandler) Prune(rounds []uint64) error {
 	// Make sure we never prune past the last normal round, as some runtimes will do historic queries
 	// for things that are not available in the last consensus state (e.g. delegation/undelegation
 	// events that happened while the runtime was suspended or not producing blocks).
-	state, err := p.commonNode.Consensus.RootHash().GetRuntimeState(context.Background(), &roothash.RuntimeRequest{
-		RuntimeID: p.commonNode.Runtime.ID(),
-		Height:    height,
+	state, err := p.consensus.RootHash().GetRuntimeState(context.TODO(), &roothash.RuntimeRequest{
+		RuntimeID: p.runtimeID,
+		Height:    consensus.HeightLatest,
 	})
 	if err != nil {
-		return fmt.Errorf("worker/executor: failed to fetch runtime state at %d: %w", height, err)
+		return fmt.Errorf("worker/executor: failed to fetch runtime state: %w", err)
 	}
 
 	for _, round := range rounds {

go/worker/compute/executor/committee/prune.go

@@ -155,6 +155,10 @@ func (w *Worker) registerRuntime(commonNode *committeeCommon.Node) error {
 		return err
 	}
 
+	// Register prune handler.
+	pruner := committee.NewPruneHandler(commonNode.Runtime.ID(), commonNode.Consensus)
+	commonNode.Runtime.History().Pruner().RegisterHandler(pruner)
+
 	commonNode.AddHooks(node)
 	w.runtimes[id] = node
 

go/worker/compute/executor/worker.go

@@ -330,11 +330,6 @@ func (w *Worker) GetLocalStorage() storageApi.LocalBackend {
 
 // NodeHooks implementation.
 
-// HandleNewBlockEarlyLocked is guarded by CrossNode.
-func (w *Worker) HandleNewBlockEarlyLocked(*runtime.BlockInfo) {
-	// Nothing to do here.
-}
-
 // HandleNewBlockLocked is guarded by CrossNode.
 func (w *Worker) HandleNewBlockLocked(bi *runtime.BlockInfo) {
 	// Notify the state syncer that there is a new block.