go/worker/compute/executor: Propose promptly upon detecting discrepancy

.changelog/5447.bugfix.2.md

@@ -0,0 +1,6 @@
+go/runtime: Fix zombie channel pipe leak on runtime restarts
+
+Pipes created by a call to channels.Unwrap spawned new goroutines
+that were not terminated during runtime restarts. These zombie
+pipes also intercepted one value from the newly created pipes,
+causing them to block indefinitely.

.changelog/5447.bugfix.md

@@ -0,0 +1 @@
+go/worker/compute/executor: Propose promptly upon detecting discrepancy

go/consensus/cometbft/apps/roothash/transactions.go

@@ -97,6 +97,7 @@
 		if err = commitment.VerifyExecutorCommitment(ctx, rtState.LastBlock, rtState.Runtime, rtState.Committee.ValidFor, &commit, msgGasAccountant, nl); err != nil { // nolint: gosec
 			ctx.Logger().Debug("failed to verify executor commitment",
 				"err", err,
+				"runtime_id", cc.ID,
 				"round", commit.Header.Header.Round,
 			)
 			return err
@@ -105,12 +106,14 @@
 		if err := rtState.CommitmentPool.AddVerifiedExecutorCommitment(rtState.Committee, &commit); err != nil { // nolint: gosec
 			ctx.Logger().Debug("failed to add executor commitment",
 				"err", err,
+				"runtime_id", cc.ID,
 				"round", commit.Header.Header.Round,
 			)
 			return err
 		}
 
 		ctx.Logger().Debug("executor commitment added to pool",
+			"runtime_id", cc.ID,
 			"round", commit.Header.Header.Round,
 			"node_id", commit.NodeID,
 			"scheduler_id", commit.Header.SchedulerID,

go/runtime/host/host.go

@@ -64,7 +64,7 @@ type Runtime interface {
 	Call(ctx context.Context, body *protocol.Body) (*protocol.Body, error)
 
 	// UpdateCapabilityTEE asks the runtime to update its CapabilityTEE with latest data.
-	UpdateCapabilityTEE(ctx context.Context) error
+	UpdateCapabilityTEE()
 
 	// WatchEvents subscribes to runtime status events.
 	WatchEvents(ctx context.Context) (<-chan *Event, pubsub.ClosableSubscription, error)

go/runtime/host/mock/mock.go

@@ -185,8 +185,7 @@
 }
 
 // Implements host.Runtime.
-func (r *runtime) UpdateCapabilityTEE(context.Context) error {
-	return nil
+func (r *runtime) UpdateCapabilityTEE() {
 }
 
 // Implements host.Runtime.

go/runtime/host/multi/multi.go

@@ -207,18 +207,16 @@
 }
 
 // UpdateCapabilityTEE implements host.Runtime.
-func (agg *Aggregate) UpdateCapabilityTEE(ctx context.Context) error {
+func (agg *Aggregate) UpdateCapabilityTEE() {
 	agg.l.RLock()
 	defer agg.l.RUnlock()
 
-	var err error
 	if agg.active != nil {
-		err = errors.Join(err, agg.active.host.UpdateCapabilityTEE(ctx))
+		agg.active.host.UpdateCapabilityTEE()
 	}
 	if agg.next != nil {
-		err = errors.Join(err, agg.next.host.UpdateCapabilityTEE(ctx))
+		agg.next.host.UpdateCapabilityTEE()
 	}
-	return err
 }
 
 // WatchEvents implements host.Runtime.

go/runtime/host/sandbox/sandbox.go

@@ -12,7 +12,6 @@
 	"time"
 
 	"github.com/cenkalti/backoff/v4"
-	"github.com/eapache/channels"
 
 	"github.com/oasisprotocol/oasis-core/go/common"
 	cmnBackoff "github.com/oasisprotocol/oasis-core/go/common/backoff"
@@ -82,15 +81,15 @@
 	id := cfg.Bundle.Manifest.ID
 
 	r := &sandboxedRuntime{
-		cfg:                       p.cfg,
-		rtCfg:                     cfg,
-		id:                        id,
-		stopCh:                    make(chan struct{}),
-		quitCh:                    make(chan struct{}),
-		ctrlCh:                    make(chan interface{}, ctrlChannelBufferSize),
-		notifier:                  pubsub.NewBroker(false),
-		notifyUpdateCapabilityTEE: channels.NewRingChannel(1),
-		logger:                    p.cfg.Logger.With("runtime_id", id),
+		cfg:                         p.cfg,
+		rtCfg:                       cfg,
+		id:                          id,
+		stopCh:                      make(chan struct{}),
+		quitCh:                      make(chan struct{}),
+		ctrlCh:                      make(chan interface{}, ctrlChannelBufferSize),
+		notifier:                    pubsub.NewBroker(false),
+		notifyUpdateCapabilityTEECh: make(chan struct{}, 1),
+		logger:                      p.cfg.Logger.With("runtime_id", id),
 	}
 
 	return r, nil
@@ -124,8 +123,8 @@
 	conn     protocol.Connection
 	notifier *pubsub.Broker
 
-	notifyUpdateCapabilityTEE *channels.RingChannel
-	capabilityTEE             *node.CapabilityTEE
+	notifyUpdateCapabilityTEECh chan struct{}
+	capabilityTEE               *node.CapabilityTEE
 
 	logger *logging.Logger
 }
@@ -193,13 +192,11 @@
 }
 
 // Implements host.Runtime.
-func (r *sandboxedRuntime) UpdateCapabilityTEE(ctx context.Context) error {
+func (r *sandboxedRuntime) UpdateCapabilityTEE() {
 	select {
-	case <-ctx.Done():
-		return ctx.Err()
-	case r.notifyUpdateCapabilityTEE.In() <- struct{}{}:
+	case r.notifyUpdateCapabilityTEECh <- struct{}{}:
+	default:
 	}
-	return nil
 }
 
 // Implements host.Runtime.
@@ -407,15 +404,12 @@
 		return fmt.Errorf("version mismatch (runtime reported: %s bundle: %s)", *rtVersion, bndVersion)
 	}
 
-	notifyUpdateCapabilityTEECh := make(chan struct{})
-	channels.Unwrap(r.notifyUpdateCapabilityTEE, notifyUpdateCapabilityTEECh)
-
 	hp := &HostInitializerParams{
 		Runtime:                   r,
 		Version:                   *rtVersion,
 		Process:                   p,
 		Connection:                pc,
-		NotifyUpdateCapabilityTEE: notifyUpdateCapabilityTEECh,
+		NotifyUpdateCapabilityTEE: r.notifyUpdateCapabilityTEECh,
 	}
 
 	// Perform configuration-specific host initialization.

go/runtime/registry/host.go

@@ -926,11 +926,7 @@
 
 				n.logger.Debug("requesting the runtime to update CapabilityTEE")
 
-				if err = n.host.UpdateCapabilityTEE(n.ctx); err != nil {
-					n.logger.Error("failed to update runtime CapabilityTEE",
-						"err", err,
-					)
-				}
+				n.host.UpdateCapabilityTEE()
 				lastAttestationUpdateHeight = height
 				lastAttestationUpdate = time.Now()
 			}

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

@@ -1668,42 +1668,37 @@ func (n *Node) roundWorker(ctx context.Context) {
 			}
 		}
 
-		for {
-			select {
-			case <-ctx.Done():
-				n.logger.Debug("exiting round, context canceled")
-				return
-			case ev := <-n.evCh:
-				// Handle an event.
-				n.handleEvent(ctx, ev)
-			case txs := <-n.txCh:
-				// Check any queued transactions.
-				n.handleNewCheckedTransactions(txs)
-			case txs := <-n.missingTxCh:
-				// Missing transactions fetched.
-				n.handleMissingTransactions(txs)
-			case ec := <-n.ecCh:
-				// Process observed executor commitments.
-				n.handleObservedExecutorCommitment(ctx, ec)
-				continue
-			case batch := <-n.processedBatchCh:
-				// Batch processing has finished.
-				n.handleProcessedBatch(ctx, batch)
-			case <-schedulerRankTicker.C:
-				// Change scheduler rank and try again.
-				schedulerRank++
-				n.logger.Debug("scheduler rank has changed",
-					"rank", schedulerRank,
-				)
-			case <-flushTimer.C:
-				// Force scheduling for primary transaction scheduler.
-				n.logger.Debug("scheduling is now forced")
-				flush = true
-			case <-n.reselectCh:
-				// Try again.
-			}
-
-			break
+		select {
+		case <-ctx.Done():
+			n.logger.Debug("exiting round, context canceled")
+			return
+		case ev := <-n.evCh:
+			// Handle an event.
+			n.handleEvent(ctx, ev)
+		case txs := <-n.txCh:
+			// Check any queued transactions.
+			n.handleNewCheckedTransactions(txs)
+		case txs := <-n.missingTxCh:
+			// Missing transactions fetched.
+			n.handleMissingTransactions(txs)
+		case ec := <-n.ecCh:
+			// Process observed executor commitments.
+			n.handleObservedExecutorCommitment(ctx, ec)
+		case batch := <-n.processedBatchCh:
+			// Batch processing has finished.
+			n.handleProcessedBatch(ctx, batch)
+		case <-schedulerRankTicker.C:
+			// Change scheduler rank and try again.
+			schedulerRank++
+			n.logger.Debug("scheduler rank has changed",
+				"rank", schedulerRank,
+			)
+		case <-flushTimer.C:
+			// Force scheduling for primary transaction scheduler.
+			n.logger.Debug("scheduling is now forced")
+			flush = true
+		case <-n.reselectCh:
+			// Try again.
 		}
 	}
 }

go/worker/registration/worker.go

@@ -162,7 +162,11 @@ func (rp *roleProvider) SetAvailableWithCallback(hook RegisterNodeHook, cb Regis
 	rp.cb = cb
 	rp.Unlock()
 
-	rp.w.registerCh <- struct{}{}
+	// Notify worker that role provider has been updated.
+	select {
+	case rp.w.registerCh <- struct{}{}:
+	default:
+	}
 }
 
 func (rp *roleProvider) SetUnavailable() {
@@ -1079,7 +1083,7 @@ func New(
 		logger:             logger,
 		consensus:          consensus,
 		p2p:                p2p,
-		registerCh:         make(chan struct{}, 64),
+		registerCh:         make(chan struct{}, 1),
 	}
 
 	w.storedDeregister = storedDeregister