go/roothash: Verify MaxMessages against global MaxRuntimeMessages

go/consensus/tendermint/api/app.go

@@ -29,6 +29,18 @@ type MessageDispatcher interface {
 	Publish(ctx *Context, kind, msg interface{}) error
 }
 
+// NoopMessageDispatcher is a no-op message dispatcher that performs no dispatch.
+type NoopMessageDispatcher struct{}
+
+// Implements MessageDispatcher.
+func (nd *NoopMessageDispatcher) Subscribe(interface{}, MessageSubscriber) {
+}
+
+// Implements MessageDispatcher.
+func (nd *NoopMessageDispatcher) Publish(*Context, interface{}, interface{}) error {
+	return nil
+}
+
 // Application is the interface implemented by multiplexed Oasis-specific
 // ABCI applications.
 type Application interface {

go/consensus/tendermint/apps/registry/api/api.go

@@ -3,8 +3,16 @@ package api
 
 type messageKind uint8
 
-// MessageNewRuntimeRegistered is the message kind for new runtime registrations. The message is
-// the runtime descriptor of the runtime that has been registered.
-//
-// The message is not emitted for runtime descriptor updates.
-var MessageNewRuntimeRegistered = messageKind(0)
+var (
+	// MessageNewRuntimeRegistered is the message kind for new runtime registrations. The message is
+	// the runtime descriptor of the runtime that has been registered.
+	//
+	// The message is not emitted for runtime descriptor updates.
+	MessageNewRuntimeRegistered = messageKind(0)
+
+	// MessageRuntimeUpdated is the message kind for runtime registration updates. The message is
+	// the runtime descriptor of the runtime that has been updated.
+	//
+	// The message is also emitted for new runtime registrations.
+	MessageRuntimeUpdated = messageKind(1)
+)

go/consensus/tendermint/apps/registry/transactions.go

@@ -592,7 +592,7 @@ func (app *registryApplication) registerRuntime( // nolint: gocyclo
 	}
 
 	// Notify other interested applications about the new runtime.
-	if existingRt == nil && !suspended {
+	if existingRt == nil {
 		if err = app.md.Publish(ctx, registryApi.MessageNewRuntimeRegistered, rt); err != nil {
 			ctx.Logger().Error("RegisterRuntime: failed to dispatch message",
 				"err", err,
@@ -601,6 +601,13 @@ func (app *registryApplication) registerRuntime( // nolint: gocyclo
 		}
 	}
 
+	if err = app.md.Publish(ctx, registryApi.MessageRuntimeUpdated, rt); err != nil {
+		ctx.Logger().Error("RegisterRuntime: failed to dispatch message",
+			"err", err,
+		)
+		return err
+	}
+
 	if err = state.SetRuntime(ctx, rt, sigRt, suspended); err != nil {
 		ctx.Logger().Error("RegisterRuntime: failed to create runtime",
 			"err", err,

go/consensus/tendermint/apps/registry/transactions_test.go

@@ -29,7 +29,8 @@ func TestRegisterNode(t *testing.T) {
 	ctx := appState.NewContext(abciAPI.ContextDeliverTx, now)
 	defer ctx.Close()
 
-	app := registryApplication{appState}
+	var md abciAPI.NoopMessageDispatcher
+	app := registryApplication{appState, &md}
 	state := registryState.NewMutableState(ctx.State())
 	stakeState := stakingState.NewMutableState(ctx.State())
 

go/consensus/tendermint/apps/roothash/roothash.go

@@ -61,6 +61,7 @@ func (app *rootHashApplication) OnRegister(state tmapi.ApplicationState, md tmap
 
 	// Subscribe to messages emitted by other apps.
 	md.Subscribe(registryApi.MessageNewRuntimeRegistered, app)
+	md.Subscribe(registryApi.MessageRuntimeUpdated, app)
 	md.Subscribe(roothashApi.RuntimeMessageNoop, app)
 }
 
@@ -284,6 +285,13 @@ func (app *rootHashApplication) ExecuteMessage(ctx *tmapi.Context, kind, msg int
 		)
 
 		return app.onNewRuntime(ctx, rt, nil)
+	case registryApi.MessageRuntimeUpdated:
+		// A runtime registration has been updated or a new runtime has been registered.
+		if ctx.IsInitChain() {
+			// Ignore messages emitted during InitChain as we handle these separately.
+			return nil
+		}
+		return app.verifyRuntimeUpdate(ctx, msg.(*registry.Runtime))
 	case roothashApi.RuntimeMessageNoop:
 		// Noop message always succeeds.
 		return nil
@@ -292,6 +300,17 @@ func (app *rootHashApplication) ExecuteMessage(ctx *tmapi.Context, kind, msg int
 	}
 }
 
+func (app *rootHashApplication) verifyRuntimeUpdate(ctx *tmapi.Context, rt *registry.Runtime) error {
+	state := roothashState.NewMutableState(ctx.State())
+
+	params, err := state.ConsensusParameters(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to get consensus parameters: %w", err)
+	}
+
+	return roothash.VerifyRuntimeParameters(ctx.Logger(), rt, params)
+}
+
 func (app *rootHashApplication) ExecuteTx(ctx *tmapi.Context, tx *transaction.Transaction) error {
 	state := roothashState.NewMutableState(ctx.State())
 

go/consensus/tendermint/tests/genesis/genesis.go

@@ -70,6 +70,7 @@ func NewTestNodeGenesisProvider(identity *identity.Identity) (genesis.Provider,
 		RootHash: roothash.Genesis{
 			Parameters: roothash.ConsensusParameters{
 				DebugDoNotSuspendRuntimes: true,
+				MaxRuntimeMessages:        32,
 			},
 		},
 		Consensus: consensus.Genesis{

go/oasis-node/cmd/debug/txsource/workload/registration.go

@@ -51,6 +51,7 @@ func getRuntime(entityID signature.PublicKey, id common.Namespace) *registry.Run
 		Executor: registry.ExecutorParameters{
 			GroupSize:    1,
 			RoundTimeout: 5,
+			MaxMessages:  32,
 		},
 		TxnScheduler: registry.TxnSchedulerParameters{
 			Algorithm:         "simple",

go/oasis-node/cmd/genesis/genesis.go

@@ -78,6 +78,7 @@ const (
 	// Roothash config flags.
 	cfgRoothashDebugDoNotSuspendRuntimes = "roothash.debug.do_not_suspend_runtimes"
 	cfgRoothashDebugBypassStake          = "roothash.debug.bypass_stake" // nolint: gosec
+	cfgRoothashMaxRuntimeMessages        = "roothash.max_runtime_messages"
 
 	// Staking config flags.
 	CfgStakingTokenSymbol        = "staking.token_symbol"
@@ -430,6 +431,7 @@ func AppendRootHashState(doc *genesis.Document, exports []string, l *logging.Log
 		Parameters: roothash.ConsensusParameters{
 			DebugDoNotSuspendRuntimes: viper.GetBool(cfgRoothashDebugDoNotSuspendRuntimes),
 			DebugBypassStake:          viper.GetBool(cfgRoothashDebugBypassStake),
+			MaxRuntimeMessages:        viper.GetUint32(cfgRoothashMaxRuntimeMessages),
 			// TODO: Make these configurable.
 			GasCosts: roothash.DefaultGasCosts,
 		},
@@ -720,6 +722,7 @@ func init() {
 	// Roothash config flags.
 	initGenesisFlags.Bool(cfgRoothashDebugDoNotSuspendRuntimes, false, "do not suspend runtimes (UNSAFE)")
 	initGenesisFlags.Bool(cfgRoothashDebugBypassStake, false, "bypass all roothash stake checks and operations (UNSAFE)")
+	initGenesisFlags.Uint32(cfgRoothashMaxRuntimeMessages, 128, "maximum number of runtime messages submitted in a round")
 	_ = initGenesisFlags.MarkHidden(cfgRoothashDebugDoNotSuspendRuntimes)
 	_ = initGenesisFlags.MarkHidden(cfgRoothashDebugBypassStake)
 

go/registry/tests/tester.go

@@ -759,6 +759,15 @@ func testRegistryRuntime(t *testing.T, backend api.Backend, consensus consensusA
 			false,
 			true,
 		},
+		// Runtime with too large MaxMessages parameter.
+		{
+			"TooBigMaxMessages",
+			func(rt *api.Runtime) {
+				rt.Executor.MaxMessages = 64 // MaxRuntimeMessages in these tests is 32.
+			},
+			false,
+			false,
+		},
 	}
 
 	rtMap := make(map[common.Namespace]*api.Runtime)
@@ -1612,6 +1621,7 @@ func NewTestRuntime(seed []byte, ent *TestEntity, isKeyManager bool) (*TestRunti
 			GroupBackupSize:   5,
 			AllowedStragglers: 1,
 			RoundTimeout:      10,
+			MaxMessages:       32,
 		},
 		TxnScheduler: api.TxnSchedulerParameters{
 			Algorithm:         api.TxnSchedulerSimple,

go/roothash/api/api.go

@@ -9,6 +9,7 @@ import (
 	"github.com/oasisprotocol/oasis-core/go/common"
 	"github.com/oasisprotocol/oasis-core/go/common/crypto/hash"
 	"github.com/oasisprotocol/oasis-core/go/common/errors"
+	"github.com/oasisprotocol/oasis-core/go/common/logging"
 	"github.com/oasisprotocol/oasis-core/go/common/pubsub"
 	"github.com/oasisprotocol/oasis-core/go/consensus/api/transaction"
 	"github.com/oasisprotocol/oasis-core/go/oasis-node/cmd/common/flags"
@@ -54,6 +55,10 @@ var (
 	// ErrProposerTimeoutNotAllowed is the error returned when proposer timeout is not allowed.
 	ErrProposerTimeoutNotAllowed = errors.New(ModuleName, 6, "roothash: proposer timeout not allowed")
 
+	// ErrMaxMessagesTooBig is the error returned when the MaxMessages parameter is set to a value
+	// larger than the MaxRuntimeMessages specified in consensus parameters.
+	ErrMaxMessagesTooBig = errors.New(ModuleName, 7, "roothash: max runtime messages is too big")
+
 	// MethodExecutorCommit is the method name for executor commit submission.
 	MethodExecutorCommit = transaction.NewMethodName(ModuleName, "ExecutorCommit", ExecutorCommit{})
 
@@ -292,3 +297,12 @@ func (g *Genesis) SanityCheck() error {
 	}
 	return nil
 }
+
+// VerifyRuntimeParameters verifies whether the runtime parameters are valid in the context of the
+// roothash service.
+func VerifyRuntimeParameters(logger *logging.Logger, rt *registry.Runtime, params *ConsensusParameters) error {
+	if rt.Executor.MaxMessages > params.MaxRuntimeMessages {
+		return ErrMaxMessagesTooBig
+	}
+	return nil
+}