go/runtime/client: Recheck failed blocks and implement max tx age

.changelog/3412.bugfix.md

@@ -0,0 +1 @@
+go/runtime/client: Runtime client should retry processing any failed blocks

.changelog/3443.bugfix.md

@@ -0,0 +1,5 @@
+go/runtime/client: Wait for initial consensus block and group version
+
+Before, the runtime client would publish invalid messages before obtaining the
+initial group version. The messages were correctly retired upon receiving the
+group version, but this resulted in needless messages.

.changelog/3443.cfg.md

@@ -0,0 +1,5 @@
+go/runtime/client: Add max transaction age
+
+Added `runtime.client.max_transaction_age` flag to configure number of
+consensus blocks after which a submitted runtime transaction is considered
+expired. Expired transactions are dropped by the client.

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

@@ -774,6 +774,7 @@ func init() {
 		compute.Flags,
 		p2p.Flags,
 		registration.Flags,
+		runtimeClient.Flags,
 		executor.Flags,
 		workerCommon.Flags,
 		workerStorage.Flags,

go/oasis-test-runner/oasis/args.go

@@ -25,6 +25,7 @@ import (
 	"github.com/oasisprotocol/oasis-core/go/oasis-node/cmd/common/grpc"
 	"github.com/oasisprotocol/oasis-core/go/oasis-node/cmd/common/metrics"
 	"github.com/oasisprotocol/oasis-core/go/oasis-node/cmd/debug/byzantine"
+	runtimeClient "github.com/oasisprotocol/oasis-core/go/runtime/client"
 	runtimeRegistry "github.com/oasisprotocol/oasis-core/go/runtime/registry"
 	workerCommon "github.com/oasisprotocol/oasis-core/go/worker/common"
 	"github.com/oasisprotocol/oasis-core/go/worker/common/p2p"
@@ -227,6 +228,13 @@ func (args *argBuilder) runtimeTagIndexerBackend(backend string) *argBuilder {
 	return args
 }
 
+func (args *argBuilder) runtimeClientMaxTransactionAge(maxTxAge int64) *argBuilder {
+	args.vec = append(args.vec, []string{
+		"--" + runtimeClient.CfgMaxTransactionAge, strconv.Itoa(int(maxTxAge)),
+	}...)
+	return args
+}
+
 func (args *argBuilder) workerClientPort(port uint16) *argBuilder {
 	args.vec = append(args.vec, []string{
 		"--" + workerCommon.CfgClientPort, strconv.Itoa(int(port)),

go/oasis-test-runner/oasis/client.go

@@ -10,13 +10,17 @@ import (
 type Client struct {
 	Node
 
+	maxTransactionAge int64
+
 	consensusPort uint16
 	p2pPort       uint16
 }
 
 // ClientCfg is the Oasis client node provisioning configuration.
 type ClientCfg struct {
 	NodeCfg
+
+	MaxTransactionAge int64
 }
 
 func (client *Client) startNode() error {
@@ -33,6 +37,10 @@ func (client *Client) startNode() error {
 		workerP2pEnabled().
 		runtimeTagIndexerBackend("bleve")
 
+	if client.maxTransactionAge != 0 {
+		args = args.runtimeClientMaxTransactionAge(client.maxTransactionAge)
+	}
+
 	for _, v := range client.net.runtimes {
 		if v.kind != registry.KindCompute {
 			continue
@@ -73,8 +81,9 @@ func (net *Network) NewClient(cfg *ClientCfg) (*Client, error) {
 			dir:       clientDir,
 			consensus: cfg.Consensus,
 		},
-		consensusPort: net.nextNodePort,
-		p2pPort:       net.nextNodePort + 1,
+		maxTransactionAge: cfg.MaxTransactionAge,
+		consensusPort:     net.nextNodePort,
+		p2pPort:           net.nextNodePort + 1,
 	}
 	client.doStartNode = client.startNode
 

go/oasis-test-runner/oasis/fixture.go

@@ -458,6 +458,9 @@ func (f *SentryFixture) Create(net *Network) (*Sentry, error) {
 type ClientFixture struct {
 	// Consensus contains configuration for the consensus backend.
 	Consensus ConsensusFixture `json:"consensus"`
+
+	// MaxTransactionAge configures the MaxTransactionAge configuration of the client.
+	MaxTransactionAge int64 `json:"max_transaction_age"`
 }
 
 // Create instantiates the client node described by the fixture.
@@ -466,6 +469,7 @@ func (f *ClientFixture) Create(net *Network) (*Client, error) {
 		NodeCfg: NodeCfg{
 			Consensus: f.Consensus,
 		},
+		MaxTransactionAge: f.MaxTransactionAge,
 	})
 }
 

go/oasis-test-runner/scenario/e2e/runtime/client_expire.go

@@ -0,0 +1,70 @@
+package runtime
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/oasisprotocol/oasis-core/go/oasis-test-runner/env"
+	"github.com/oasisprotocol/oasis-core/go/oasis-test-runner/oasis"
+	"github.com/oasisprotocol/oasis-core/go/oasis-test-runner/scenario"
+	"github.com/oasisprotocol/oasis-core/go/runtime/client/api"
+)
+
+// ClientExpire is the ClientExpire node scenario.
+var ClientExpire scenario.Scenario = newClientExpireImpl("client-expire", "simple-keyvalue-client", nil)
+
+type clientExpireImpl struct {
+	runtimeImpl
+}
+
+func newClientExpireImpl(name, clientBinary string, clientArgs []string) scenario.Scenario {
+	return &clientExpireImpl{
+		runtimeImpl: *newRuntimeImpl(name, clientBinary, clientArgs),
+	}
+}
+
+func (sc *clientExpireImpl) Clone() scenario.Scenario {
+	return &clientExpireImpl{
+		runtimeImpl: *sc.runtimeImpl.Clone().(*runtimeImpl),
+	}
+}
+
+func (sc *clientExpireImpl) Fixture() (*oasis.NetworkFixture, error) {
+	f, err := sc.runtimeImpl.Fixture()
+	if err != nil {
+		return nil, err
+	}
+
+	// Make client expire all transactions instantly.
+	f.Clients[0].MaxTransactionAge = 1
+
+	return f, nil
+}
+
+func (sc *clientExpireImpl) Run(childEnv *env.Env) error {
+	ctx := context.Background()
+
+	// Start the network.
+	var err error
+	if err = sc.Net.Start(); err != nil {
+		return err
+	}
+
+	// Wait for client to be ready.
+	client := sc.Net.Clients()[0]
+	nodeCtrl, err := oasis.NewController(client.SocketPath())
+	if err != nil {
+		return err
+	}
+	if err = nodeCtrl.WaitReady(ctx); err != nil {
+		return err
+	}
+
+	err = sc.submitKeyValueRuntimeInsertTx(ctx, runtimeID, "hello", "test")
+	if !errors.Is(err, api.ErrTransactionExpired) {
+		return fmt.Errorf("expected error: %v, got: %v", api.ErrTransactionExpired, err)
+	}
+
+	return nil
+}

go/oasis-test-runner/scenario/e2e/runtime/runtime.go

@@ -532,6 +532,8 @@ func RegisterScenarios() error {
 		RuntimeDynamic,
 		// Transaction source test.
 		TxSourceMultiShort,
+		// ClientExpire test.
+		ClientExpire,
 		// Late start test.
 		LateStart,
 		// KeymanagerUpgrade test.

go/runtime/client/api/api.go

@@ -26,6 +26,8 @@ var (
 	ErrNotFound = errors.New(ModuleName, 1, "client: not found")
 	// ErrInternal is an error returned when an unspecified internal error occurs.
 	ErrInternal = errors.New(ModuleName, 2, "client: internal error")
+	// ErrTransactionExpired is an error returned when transaction expired.
+	ErrTransactionExpired = errors.New(ModuleName, 3, "client: transaction expired")
 )
 
 // RuntimeClient is the runtime client interface.

go/runtime/client/client.go

@@ -6,13 +6,17 @@ import (
 	"fmt"
 	"sync"
 
+	flag "github.com/spf13/pflag"
+	"github.com/spf13/viper"
+
 	"github.com/oasisprotocol/oasis-core/go/common"
 	"github.com/oasisprotocol/oasis-core/go/common/crypto/hash"
 	"github.com/oasisprotocol/oasis-core/go/common/logging"
 	"github.com/oasisprotocol/oasis-core/go/common/pubsub"
 	consensus "github.com/oasisprotocol/oasis-core/go/consensus/api"
 	keymanagerAPI "github.com/oasisprotocol/oasis-core/go/keymanager/api"
 	keymanager "github.com/oasisprotocol/oasis-core/go/keymanager/client"
+	cmdFlags "github.com/oasisprotocol/oasis-core/go/oasis-node/cmd/common/flags"
 	roothash "github.com/oasisprotocol/oasis-core/go/roothash/api"
 	"github.com/oasisprotocol/oasis-core/go/roothash/api/block"
 	"github.com/oasisprotocol/oasis-core/go/runtime/client/api"
@@ -25,9 +29,20 @@ import (
 	executor "github.com/oasisprotocol/oasis-core/go/worker/compute/executor/api"
 )
 
+const (
+	// CfgMaxTransactionAge is the number of consensus blocks after which
+	// submitted transactions will be considered expired.
+	CfgMaxTransactionAge = "runtime.client.max_transaction_age"
+
+	minMaxTransactionAge = 30
+)
+
 var (
 	_ api.RuntimeClient    = (*runtimeClient)(nil)
 	_ enclaverpc.Transport = (*runtimeClient)(nil)
+
+	// Flags has the flags used by the runtime client.
+	Flags = flag.NewFlagSet("", flag.ContinueOnError)
 )
 
 type clientCommon struct {
@@ -48,6 +63,8 @@ type runtimeClient struct {
 	watchers  map[common.Namespace]*blockWatcher
 	kmClients map[common.Namespace]*keymanager.Client
 
+	maxTransactionAge int64
+
 	logger *logging.Logger
 }
 
@@ -71,7 +88,7 @@ func (c *runtimeClient) SubmitTx(ctx context.Context, request *api.SubmitTxReque
 	var err error
 	c.Lock()
 	if watcher, ok = c.watchers[request.RuntimeID]; !ok {
-		watcher, err = newWatcher(c.common, request.RuntimeID, c.common.p2p)
+		watcher, err = newWatcher(c.common, request.RuntimeID, c.common.p2p, c.maxTransactionAge)
 		if err != nil {
 			c.Unlock()
 			return nil, err
@@ -118,6 +135,10 @@ func (c *runtimeClient) SubmitTx(ctx context.Context, request *api.SubmitTxReque
 				return nil, fmt.Errorf("client: block watch channel closed unexpectedly (unknown error)")
 			}
 
+			if resp.err != nil {
+				return nil, resp.err
+			}
+
 			// The main event is getting a response from the watcher, handled below. If there is
 			// no result yet, this means that we need to retry publish.
 			if resp.result == nil {
@@ -428,6 +449,11 @@ func New(
 	runtimeRegistry runtimeRegistry.Registry,
 	p2p *p2p.P2P,
 ) (api.RuntimeClient, error) {
+	maxTransactionAge := viper.GetInt64(CfgMaxTransactionAge)
+	if maxTransactionAge < minMaxTransactionAge && !cmdFlags.DebugDontBlameOasis() {
+		return nil, fmt.Errorf("max transaction age too low: %d, minimum: %d", maxTransactionAge, minMaxTransactionAge)
+	}
+
 	c := &runtimeClient{
 		common: &clientCommon{
 			storage:         runtimeRegistry.StorageRouter(),
@@ -436,9 +462,16 @@ func New(
 			ctx:             ctx,
 			p2p:             p2p,
 		},
-		watchers:  make(map[common.Namespace]*blockWatcher),
-		kmClients: make(map[common.Namespace]*keymanager.Client),
-		logger:    logging.GetLogger("runtime/client"),
+		watchers:          make(map[common.Namespace]*blockWatcher),
+		kmClients:         make(map[common.Namespace]*keymanager.Client),
+		maxTransactionAge: maxTransactionAge,
+		logger:            logging.GetLogger("runtime/client"),
 	}
 	return c, nil
 }
+
+func init() {
+	Flags.Int64(CfgMaxTransactionAge, 1500, "number of consensus blocks after which submitted transactions will be considered expired")
+
+	_ = viper.BindPFlags(Flags)
+}