Merge pull request #5406 from oasisprotocol/peternose/stable/23.0.x/b…

…ackport-5404

[BACKPORT/23.0.x] go/worker: Minor bug fixes

.changelog/5403.bugfix.md

@@ -0,0 +1 @@
+go/worker/client: Fix nil dereference on early Query

.changelog/5403.internal.md

@@ -0,0 +1 @@
+go: Bump golang.org/x/net to v0.17.0

.changelog/5404.bugfix.1.md

@@ -0,0 +1 @@
+go/worker/compute/executor: Use local time for batch scheduling

.changelog/5404.bugfix.2.md

@@ -0,0 +1 @@
+go/worker/compute/executor: Start processing once all txs are fetched

.changelog/5404.bugfix.3.md

@@ -0,0 +1 @@
+go/worker/compute/executor: Schedule only if higher ranks didn't propose

.changelog/5404.bugfix.4.md

@@ -0,0 +1 @@
+go/worker/compute/executor: Estimate pool rank from observed commitments

docs/oasis-node/metrics.md

@@ -127,7 +127,7 @@ oasis_worker_node_status_frozen | Gauge | Is oasis node frozen (binary). |  | [w
 oasis_worker_node_status_runtime_faults | Gauge | Number of runtime faults. | runtime | [worker/registration](https://github.com/oasisprotocol/oasis-core/tree/master/go/worker/registration/worker.go)
 oasis_worker_node_status_runtime_suspended | Gauge | Runtime node suspension status (binary). | runtime | [worker/registration](https://github.com/oasisprotocol/oasis-core/tree/master/go/worker/registration/worker.go)
 oasis_worker_processed_block_count | Counter | Number of processed roothash blocks. | runtime | [worker/common/committee](https://github.com/oasisprotocol/oasis-core/tree/master/go/worker/common/committee/node.go)
-oasis_worker_processed_event_count | Counter | Number of processed roothash events. | runtime | [worker/common/committee](https://github.com/oasisprotocol/oasis-core/tree/master/go/worker/common/committee/node.go)
+oasis_worker_processed_event_count | Counter | Number of processed roothash events. | runtime | [worker/compute/executor/committee](https://github.com/oasisprotocol/oasis-core/tree/master/go/worker/compute/executor/committee/metrics.go)
 oasis_worker_storage_commit_latency | Summary | Latency of storage commit calls (state + outputs) (seconds). | runtime | [worker/compute/executor/committee](https://github.com/oasisprotocol/oasis-core/tree/master/go/worker/compute/executor/committee/metrics.go)
 oasis_worker_storage_full_round | Gauge | The last round that was fully synced and finalized. | runtime | [worker/storage/committee](https://github.com/oasisprotocol/oasis-core/tree/master/go/worker/storage/committee/metrics.go)
 oasis_worker_storage_pending_round | Gauge | The last round that is in-flight for syncing. | runtime | [worker/storage/committee](https://github.com/oasisprotocol/oasis-core/tree/master/go/worker/storage/committee/metrics.go)

go/consensus/cometbft/apps/roothash/finalization.go

@@ -83,7 +83,7 @@ func (app *rootHashApplication) tryFinalizeRoundInsideTx( //nolint: gocyclo
 	case commitment.ErrDiscrepancyDetected:
 		ctx.Logger().Warn("executor discrepancy detected",
 			"round", round,
-			"priority", rtState.CommitmentPool.HighestRank,
+			"rank", rtState.CommitmentPool.HighestRank,
 			logging.LogEvent, roothash.LogEventExecutionDiscrepancyDetected,
 		)
 
@@ -291,6 +291,7 @@ func (app *rootHashApplication) finalizeBlock(ctx *tmapi.Context, rtState *rooth
 
 	// Emit event.
 	ctx.Logger().Debug("new runtime block",
+		"runtime_id", rtState.Runtime.ID,
 		"height", ctx.BlockHeight()+1, // Current height is ctx.BlockHeight() + 1
 		"round", blk.Header.Round,
 		"type", blk.Header.HeaderType,

go/go.mod

@@ -51,9 +51,9 @@ require (
 	github.com/tidwall/btree v1.6.0
 	github.com/tyler-smith/go-bip39 v1.1.0
 	go.uber.org/zap v1.25.0
-	golang.org/x/crypto v0.12.0
+	golang.org/x/crypto v0.14.0
 	golang.org/x/exp v0.0.0-20230817173708-d852ddb80c63
-	golang.org/x/net v0.14.0
+	golang.org/x/net v0.17.0
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19
 	google.golang.org/grpc v1.57.0
 	google.golang.org/grpc/security/advancedtls v0.0.0-20221004221323-12db695f1648
@@ -184,8 +184,8 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/mod v0.12.0 // indirect
 	golang.org/x/sync v0.3.0 // indirect
-	golang.org/x/sys v0.11.0 // indirect
-	golang.org/x/text v0.12.0 // indirect
+	golang.org/x/sys v0.13.0 // indirect
+	golang.org/x/text v0.13.0 // indirect
 	golang.org/x/tools v0.12.1-0.20230815132531-74c255bcf846 // indirect
 	gopkg.in/fsnotify.v1 v1.4.7 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect

go/go.sum

@@ -730,8 +730,8 @@ golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
 golang.org/x/crypto v0.0.0-20210421170649-83a5a9bb288b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
 golang.org/x/crypto v0.0.0-20211108221036-ceb1ce70b4fa/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
-golang.org/x/crypto v0.12.0 h1:tFM/ta59kqch6LlvYnPa0yx5a83cL2nHflFhYKvv9Yk=
-golang.org/x/crypto v0.12.0/go.mod h1:NF0Gs7EO5K4qLn+Ylc+fih8BSTeIjAP05siRnAh98yw=
+golang.org/x/crypto v0.14.0 h1:wBqGXzWJW6m1XrIKlAH0Hs1JJ7+9KBwnIO8v66Q9cHc=
+golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8=
@@ -811,8 +811,8 @@ golang.org/x/net v0.0.0-20201224014010-6772e930b67b/go.mod h1:m0MpNAwzfU5UDzcl9v
 golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
-golang.org/x/net v0.14.0 h1:BONx9s002vGdD9umnlX1Po8vOZmrgH34qlHcD1MfK14=
-golang.org/x/net v0.14.0/go.mod h1:PpSgVXXLK0OxS0F31C1/tv6XNguvCrnXIDrFMspZIUI=
+golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM=
+golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20181017192945-9dcd33a902f4/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20181203162652-d668ce993890/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
@@ -903,17 +903,17 @@ golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20221010170243-090e33056c14/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.11.0 h1:eG7RXZHdqOJ1i+0lgLgCpSXAp6M3LYlAo6osgSi0xOM=
-golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.13.0 h1:Af8nKPmuFypiUBjVoU9V20FiaFXOcuZI21p0ycVYYGE=
+golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.12.0 h1:k+n5B8goJNdU7hSvEtMUz3d1Q6D/XW4COJSJR6fN0mc=
-golang.org/x/text v0.12.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
+golang.org/x/text v0.13.0 h1:ablQoSUd0tRdKxZewP80B+BaqeKJuVhuRxj/dkrun3k=
+golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=

go/roothash/api/commitment/pool.go

@@ -228,13 +228,15 @@ func (p *Pool) AddVerifiedExecutorCommitment(c *scheduler.Committee, ec *Executo
 		// Discrepancy detection accepts commitments arriving in any order, e.g., a backup worker
 		// can submit a commitment even before there is a discrepancy.
 		logger.Debug("node is not in the committee",
+			"round", ec.Header.Header.Round,
 			"node_id", ec.NodeID,
 		)
 		return ErrNotInCommittee
 	case p.Discrepancy && !c.IsBackupWorker(ec.NodeID):
 		// Discrepancy resolution accepts commitments only from backup workers to prevent workers
 		// from improving their liveness statistics.
 		logger.Debug("node is not a backup worker",
+			"round", ec.Header.Header.Round,
 			"node_id", ec.NodeID,
 		)
 		return ErrBadExecutorCommitment
@@ -246,7 +248,9 @@ func (p *Pool) AddVerifiedExecutorCommitment(c *scheduler.Committee, ec *Executo
 		// Reject commitments with invalid schedulers.
 		logger.Debug("executor commitment's scheduler is not in the committee",
 			"round", ec.Header.Header.Round,
+			"node_id", ec.NodeID,
 			"scheduler_id", ec.Header.SchedulerID,
+			"rank", rank,
 		)
 		return ErrBadExecutorCommitment
 	}
@@ -255,18 +259,22 @@ func (p *Pool) AddVerifiedExecutorCommitment(c *scheduler.Committee, ec *Executo
 	switch {
 	case rank > p.HighestRank:
 		// Reject commitments with higher ranking.
-		logger.Debug("executor commitment's scheduler has too high ranking",
+		logger.Debug("executor commitment's scheduler has worse ranking",
 			"round", ec.Header.Header.Round,
-			"commitment_rank", rank,
-			"pool_rank", p.HighestRank,
+			"node_id", ec.NodeID,
+			"scheduler_id", ec.Header.SchedulerID,
+			"rank", rank,
+			"highest_rank", p.HighestRank,
 		)
 		return ErrBadExecutorCommitment
 	case rank != p.HighestRank && p.Discrepancy:
 		// Prevent placing commitments with different rank during discrepancy resolution.
 		logger.Debug("executor commitment's scheduler rank does not match",
 			"round", ec.Header.Header.Round,
-			"commitment_rank", rank,
-			"pool_rank", p.HighestRank,
+			"node_id", ec.NodeID,
+			"scheduler_id", ec.Header.SchedulerID,
+			"rank", rank,
+			"highest_rank", p.HighestRank,
 		)
 		return ErrBadExecutorCommitment
 	case rank < p.HighestRank:

go/roothash/tests/tester.go

@@ -32,7 +32,7 @@ import (
 )
 
 const (
-	recvTimeout = 5 * time.Second
+	recvTimeout = 10 * time.Second
 	nrRuntimes  = 3
 )
 

go/worker/client/committee/node.go

@@ -98,11 +98,6 @@ func (n *Node) HandleNewBlockLocked(*runtime.BlockInfo) {
 	// Nothing to do here.
 }
 
-// HandleNewEventLocked is guarded by CrossNode.
-func (n *Node) HandleNewEventLocked(*roothash.Event) {
-	// Nothing to do here.
-}
-
 // HandleRuntimeHostEventLocked is guarded by CrossNode.
 func (n *Node) HandleRuntimeHostEventLocked(*host.Event) {
 	// Nothing to do here.
@@ -149,13 +144,14 @@ func (n *Node) Query(ctx context.Context, round uint64, method string, args []by
 	// Fetch the active descriptor so we can get the current message limits.
 	n.commonNode.CrossNode.Lock()
 	dsc := n.commonNode.CurrentDescriptor
-	latestRound := n.commonNode.CurrentBlock.Header.Round
+	blk := n.commonNode.CurrentBlock
 	n.commonNode.CrossNode.Unlock()
 
-	if dsc == nil {
+	if dsc == nil || blk == nil {
 		return nil, api.ErrNoHostedRuntime
 	}
 	maxMessages := dsc.Executor.MaxMessages
+	latestRound := n.commonNode.CurrentBlock.Header.Round
 
 	rtInfo, err := hrt.GetInfo(ctx)
 	if err != nil {

go/worker/common/committee/node.go

@@ -43,13 +43,6 @@ var (
 		},
 		[]string{"runtime"},
 	)
-	processedEventCount = prometheus.NewCounterVec(
-		prometheus.CounterOpts{
-			Name: "oasis_worker_processed_event_count",
-			Help: "Number of processed roothash events.",
-		},
-		[]string{"runtime"},
-	)
 	failedRoundCount = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Name: "oasis_worker_failed_round_count",
@@ -116,7 +109,6 @@ var (
 
 	nodeCollectors = []prometheus.Collector{
 		processedBlockCount,
-		processedEventCount,
 		failedRoundCount,
 		epochTransitionCount,
 		epochNumber,
@@ -145,8 +137,6 @@ type NodeHooks interface {
 	// Guarded by CrossNode.
 	HandleNewBlockLocked(*runtime.BlockInfo)
 	// Guarded by CrossNode.
-	HandleNewEventLocked(*roothash.Event)
-	// Guarded by CrossNode.
 	HandleRuntimeHostEventLocked(*host.Event)
 
 	// Initialized returns a channel that will be closed when the worker is initialized and ready
@@ -614,15 +604,6 @@ func (n *Node) handleNewBlockLocked(blk *block.Block, height int64) {
 	}
 }
 
-// Guarded by n.CrossNode.
-func (n *Node) handleNewEventLocked(ev *roothash.Event) {
-	processedEventCount.With(n.getMetricLabels()).Inc()
-
-	for _, hooks := range n.hooks {
-		hooks.HandleNewEventLocked(ev)
-	}
-}
-
 // Guarded by n.CrossNode.
 func (n *Node) handleRuntimeHostEventLocked(ev *host.Event) {
 	n.logger.Debug("got runtime event", "ev", ev)
@@ -710,16 +691,6 @@ func (n *Node) worker() {
 	}
 	defer blocksSub.Close()
 
-	// Start watching roothash events.
-	events, eventsSub, err := n.Consensus.RootHash().WatchEvents(n.ctx, n.Runtime.ID())
-	if err != nil {
-		n.logger.Error("failed to subscribe to roothash events",
-			"err", err,
-		)
-		return
-	}
-	defer eventsSub.Close()
-
 	// Provision the hosted runtime.
 	hrt, hrtNotifier, err := n.ProvisionHostedRuntime(n.ctx)
 	if err != nil {
@@ -796,13 +767,6 @@ func (n *Node) worker() {
 				defer n.CrossNode.Unlock()
 				n.handleNewBlockLocked(blk.Block, blk.Height)
 			}()
-		case ev := <-events:
-			// Received an event.
-			func() {
-				n.CrossNode.Lock()
-				defer n.CrossNode.Unlock()
-				n.handleNewEventLocked(ev)
-			}()
 		case ev := <-hrtEventCh:
 			// Received a hosted runtime event.
 			func() {

go/worker/compute/executor/committee/discrepancy.go

@@ -13,17 +13,6 @@ type discrepancyEvent struct {
 	authoritative bool
 }
 
-func (n *Node) NotifyDiscrepancy(info *discrepancyEvent) {
-	// Drop discrepancies if the worker falls behind.
-	select {
-	case <-n.discrepancyCh:
-	default:
-	}
-
-	// Non-blocking send.
-	n.discrepancyCh <- info
-}
-
 func (n *Node) handleDiscrepancy(ctx context.Context, ev *discrepancyEvent) {
 	n.logger.Warn("execution discrepancy detected",
 		"rank", ev.rank,
@@ -55,17 +44,7 @@ func (n *Node) handleDiscrepancy(ctx context.Context, ev *discrepancyEvent) {
 	n.discrepancy = ev
 }
 
-func (n *Node) handleObservedExecutorCommitment(ctx context.Context, ec *commitment.ExecutorCommitment) {
-	// Don't do anything if we are not a backup worker or we are an executor worker.
-	id := n.commonNode.Identity.NodeSigner.Public()
-	if !n.committee.IsBackupWorker(id) || n.committee.IsWorker(id) {
-		return
-	}
-
-	n.logger.Debug("observed executor commitment",
-		"commitment", ec,
-	)
-
+func (n *Node) predictDiscrepancy(ctx context.Context, ec *commitment.ExecutorCommitment) {
 	// TODO: Handle equivocation detection.
 
 	// Don't do anything if the discrepancy has already been detected.
@@ -101,7 +80,7 @@ func (n *Node) handleObservedExecutorCommitment(ctx context.Context, ec *commitm
 
 	n.logger.Warn("observed commitments indicate discrepancy")
 
-	n.NotifyDiscrepancy(&discrepancyEvent{
+	n.handleDiscrepancy(ctx, &discrepancyEvent{
 		rank:          n.commitPool.HighestRank,
 		height:        uint64(n.blockInfo.ConsensusBlock.Height),
 		authoritative: false,

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

@@ -4,8 +4,6 @@ import (
 	"context"
 
 	"github.com/oasisprotocol/oasis-core/go/common/crash"
-	roothash "github.com/oasisprotocol/oasis-core/go/roothash/api"
-	"github.com/oasisprotocol/oasis-core/go/roothash/api/commitment"
 	runtime "github.com/oasisprotocol/oasis-core/go/runtime/api"
 	"github.com/oasisprotocol/oasis-core/go/worker/common/committee"
 )
@@ -46,34 +44,3 @@ func (n *Node) HandleNewBlockLocked(bi *runtime.BlockInfo) {
 	// Non-blocking send.
 	n.blockInfoCh <- bi
 }
-
-// HandleNewEventLocked implements NodeHooks.
-// Guarded by n.commonNode.CrossNode.
-func (n *Node) HandleNewEventLocked(ev *roothash.Event) {
-	switch {
-	case ev.ExecutionDiscrepancyDetected != nil:
-		n.NotifyDiscrepancy(&discrepancyEvent{
-			rank:          ev.ExecutionDiscrepancyDetected.Rank,
-			height:        uint64(ev.Height),
-			authoritative: true,
-		})
-	case ev.ExecutorCommitted != nil:
-		n.NotifySchedulerCommitment(&ev.ExecutorCommitted.Commit)
-	}
-}
-
-func (n *Node) NotifySchedulerCommitment(ec *commitment.ExecutorCommitment) {
-	// Filter scheduler commitments.
-	if ec.NodeID != ec.Header.SchedulerID {
-		return
-	}
-
-	// Drop commitments if the worker falls behind. The pool's rank can only improve.
-	select {
-	case <-n.schedulerCommitmentCh:
-	default:
-	}
-
-	// Non-blocking send.
-	n.schedulerCommitmentCh <- ec
-}

go/worker/compute/executor/committee/metrics.go

@@ -9,6 +9,13 @@ import (
 )
 
 var (
+	processedEventCount = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "oasis_worker_processed_event_count",
+			Help: "Number of processed roothash events.",
+		},
+		[]string{"runtime"},
+	)
 	discrepancyDetectedCount = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Name: "oasis_worker_execution_discrepancy_detected_count",
@@ -52,6 +59,7 @@ var (
 		[]string{"runtime"},
 	)
 	nodeCollectors = []prometheus.Collector{
+		processedEventCount,
 		discrepancyDetectedCount,
 		abortedBatchCount,
 		storageCommitLatency,