Don't clear tx pool on epoch transitions, add txsync protocol

.changelog/4579.bugfix.md

@@ -0,0 +1 @@
+Don't clear tx pool on epoch transitions, add txsync protocol

go/oasis-net-runner/fixtures/default.go

@@ -182,9 +182,9 @@ func newDefaultFixture() (*oasis.NetworkFixture, error) {
 					MaxMessages:     128,
 				},
 				TxnScheduler: registry.TxnSchedulerParameters{
-					MaxBatchSize:      1,
+					MaxBatchSize:      1000,
 					MaxBatchSizeBytes: 16 * 1024 * 1024, // 16 MiB
-					BatchFlushTimeout: 20 * time.Second,
+					BatchFlushTimeout: 1 * time.Second,
 					ProposerTimeout:   20,
 				},
 				AdmissionPolicy: registry.RuntimeAdmissionPolicy{

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

@@ -76,6 +76,8 @@ var RuntimeFlags = flag.NewFlagSet("", flag.ContinueOnError)
 
 // TxnCall is a transaction call in the test runtime.
 type TxnCall struct {
+	// Nonce is a nonce.
+	Nonce uint64 `json:"nonce"`
 	// Method is the called method name.
 	Method string `json:"method"`
 	// Args are the method arguments.
@@ -231,15 +233,14 @@ func (r *runtime) doInsertRequest(ctx context.Context, rng *rand.Rand, rtc runti
 
 	// Submit request.
 	req := &TxnCall{
+		Nonce:  rng.Uint64(),
 		Method: "insert",
 		Args: struct {
 			Key   string `json:"key"`
 			Value string `json:"value"`
-			Nonce uint64 `json:"nonce"`
 		}{
 			Key:   key,
 			Value: value,
-			Nonce: rng.Uint64(),
 		},
 	}
 	rsp, round, err := r.submitRuntimeRquest(ctx, rtc, req)
@@ -284,13 +285,12 @@ func (r *runtime) doGetRequest(ctx context.Context, rng *rand.Rand, rtc runtimeC
 
 	// Submit request.
 	req := &TxnCall{
+		Nonce:  rng.Uint64(),
 		Method: "get",
 		Args: struct {
-			Key   string `json:"key"`
-			Nonce uint64 `json:"nonce"`
+			Key string `json:"key"`
 		}{
-			Key:   key,
-			Nonce: rng.Uint64(),
+			Key: key,
 		},
 	}
 	rsp, round, err := r.submitRuntimeRquest(ctx, rtc, req)
@@ -332,13 +332,12 @@ func (r *runtime) doRemoveRequest(ctx context.Context, rng *rand.Rand, rtc runti
 
 	// Submit request.
 	req := &TxnCall{
+		Nonce:  rng.Uint64(),
 		Method: "remove",
 		Args: struct {
-			Key   string `json:"key"`
-			Nonce uint64 `json:"nonce"`
+			Key string `json:"key"`
 		}{
-			Key:   key,
-			Nonce: rng.Uint64(),
+			Key: key,
 		},
 	}
 	rsp, round, err := r.submitRuntimeRquest(ctx, rtc, req)
@@ -386,15 +385,14 @@ func (r *runtime) doInMsgRequest(ctx context.Context, rng *rand.Rand, rtc runtim
 		ID:  r.runtimeID,
 		Tag: 42,
 		Data: cbor.Marshal(&TxnCall{
+			Nonce:  rng.Uint64(),
 			Method: "insert",
 			Args: struct {
 				Key   string `json:"key"`
 				Value string `json:"value"`
-				Nonce uint64 `json:"nonce"`
 			}{
 				Key:   key,
 				Value: value,
-				Nonce: rng.Uint64(),
 			},
 		}),
 	})
@@ -568,16 +566,15 @@ func (r *runtime) doWithdrawRequest(ctx context.Context, rng *rand.Rand, rtc run
 	// Submit message request.
 	amount := *quantity.NewFromUint64(1)
 	req := &TxnCall{
+		Nonce:  rng.Uint64(),
 		Method: "consensus_withdraw",
 		Args: struct {
 			Withdraw staking.Withdraw `json:"withdraw"`
-			Nonce    uint64           `json:"nonce"`
 		}{
 			Withdraw: staking.Withdraw{
 				From:   r.testAddress,
 				Amount: amount,
 			},
-			Nonce: rng.Uint64(),
 		},
 	}
 	rsp, round, err := r.submitRuntimeRquest(ctx, rtc, req)
@@ -613,16 +610,15 @@ func (r *runtime) doTransferRequest(ctx context.Context, rng *rand.Rand, rtc run
 	// Submit message request.
 	amount := *quantity.NewFromUint64(1)
 	req := &TxnCall{
+		Nonce:  rng.Uint64(),
 		Method: "consensus_transfer",
 		Args: struct {
 			Transfer staking.Transfer `json:"transfer"`
-			Nonce    uint64           `json:"nonce"`
 		}{
 			Transfer: staking.Transfer{
 				To:     r.testAddress,
 				Amount: amount,
 			},
-			Nonce: rng.Uint64(),
 		},
 	}
 	rsp, round, err := r.submitRuntimeRquest(ctx, rtc, req)
@@ -658,16 +654,15 @@ func (r *runtime) doAddEscrowRequest(ctx context.Context, rng *rand.Rand, rtc ru
 	// Submit message request.
 	amount := *quantity.NewFromUint64(1)
 	req := &TxnCall{
+		Nonce:  rng.Uint64(),
 		Method: "consensus_add_escrow",
 		Args: struct {
 			Escrow staking.Escrow `json:"escrow"`
-			Nonce  uint64         `json:"nonce"`
 		}{
 			Escrow: staking.Escrow{
 				Account: r.testAddress,
 				Amount:  amount,
 			},
-			Nonce: rng.Uint64(),
 		},
 	}
 	rsp, round, err := r.submitRuntimeRquest(ctx, rtc, req)
@@ -705,16 +700,15 @@ func (r *runtime) doReclaimEscrowRequest(ctx context.Context, rng *rand.Rand, rt
 	// getting any rewards or is being slashed.
 	amount := *quantity.NewFromUint64(1)
 	req := &TxnCall{
+		Nonce:  rng.Uint64(),
 		Method: "consensus_reclaim_escrow",
 		Args: struct {
 			ReclaimEscrow staking.ReclaimEscrow `json:"reclaim_escrow"`
-			Nonce         uint64                `json:"nonce"`
 		}{
 			ReclaimEscrow: staking.ReclaimEscrow{
 				Account: r.testAddress,
 				Shares:  amount,
 			},
-			Nonce: rng.Uint64(),
 		},
 	}
 	rsp, round, err := r.submitRuntimeRquest(ctx, rtc, req)

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

@@ -92,8 +92,8 @@ func (sc *multipleRuntimesImpl) Fixture() (*oasis.NetworkFixture, error) {
 				RoundTimeout:    20,
 			},
 			TxnScheduler: registry.TxnSchedulerParameters{
-				MaxBatchSize:      1,
-				MaxBatchSizeBytes: 1024,
+				MaxBatchSize:      100,
+				MaxBatchSizeBytes: 1024 * 1024,
 				BatchFlushTimeout: 1 * time.Second,
 				ProposerTimeout:   10,
 			},

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

@@ -61,6 +61,8 @@ var (
 
 // TxnCall is a transaction call in the test runtime.
 type TxnCall struct {
+	// Nonce is a nonce.
+	Nonce uint64 `json:"nonce"`
 	// Method is the called method name.
 	Method string `json:"method"`
 	// Args are the method arguments.
@@ -202,8 +204,8 @@ func (sc *runtimeImpl) Fixture() (*oasis.NetworkFixture, error) {
 					MaxMessages:     128,
 				},
 				TxnScheduler: registry.TxnSchedulerParameters{
-					MaxBatchSize:      1,
-					MaxBatchSizeBytes: 1024,
+					MaxBatchSize:      100,
+					MaxBatchSizeBytes: 1024 * 1024,
 					BatchFlushTimeout: 1 * time.Second,
 					ProposerTimeout:   20,
 					MaxInMessages:     128,
@@ -362,9 +364,15 @@ func (sc *runtimeImpl) Run(childEnv *env.Env) error {
 	return sc.waitTestClient()
 }
 
-func (sc *runtimeImpl) submitRuntimeTx(ctx context.Context, id common.Namespace, method string, args interface{}) (cbor.RawMessage, error) {
+func (sc *runtimeImpl) submitRuntimeTx(
+	ctx context.Context,
+	id common.Namespace,
+	nonce uint64,
+	method string,
+	args interface{},
+) (cbor.RawMessage, error) {
 	// Submit a transaction and check the result.
-	metaResp, err := sc.submitRuntimeTxMeta(ctx, id, method, args)
+	metaResp, err := sc.submitRuntimeTxMeta(ctx, id, nonce, method, args)
 	if err != nil {
 		return nil, err
 	}
@@ -378,6 +386,7 @@ func (sc *runtimeImpl) submitRuntimeTx(ctx context.Context, id common.Namespace,
 func (sc *runtimeImpl) submitRuntimeTxMeta(
 	ctx context.Context,
 	id common.Namespace,
+	nonce uint64,
 	method string,
 	args interface{},
 ) (*runtimeClient.SubmitTxMetaResponse, error) {
@@ -390,13 +399,17 @@ func (sc *runtimeImpl) submitRuntimeTxMeta(
 	resp, err := c.SubmitTxMeta(ctx, &runtimeClient.SubmitTxRequest{
 		RuntimeID: id,
 		Data: cbor.Marshal(&TxnCall{
+			Nonce:  nonce,
 			Method: method,
 			Args:   args,
 		}),
 	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to submit runtime meta tx: %w", err)
 	}
+	if resp.CheckTxError != nil {
+		return nil, fmt.Errorf("check tx failed: %s", resp.CheckTxError.Message)
+	}
 
 	return resp, nil
 }
@@ -418,12 +431,10 @@ func (sc *runtimeImpl) submitConsensusXferTx(
 	xfer staking.Transfer,
 	nonce uint64,
 ) error {
-	_, err := sc.submitRuntimeTx(ctx, runtimeID, "consensus_transfer", struct {
+	_, err := sc.submitRuntimeTx(ctx, runtimeID, nonce, "consensus_transfer", struct {
 		Transfer staking.Transfer `json:"transfer"`
-		Nonce    uint64           `json:"nonce"`
 	}{
 		Transfer: xfer,
-		Nonce:    nonce,
 	})
 	return err
 }
@@ -434,16 +445,14 @@ func (sc *runtimeImpl) submitConsensusXferTxMeta(
 	xfer staking.Transfer,
 	nonce uint64,
 ) (*runtimeClient.SubmitTxMetaResponse, error) {
-	return sc.submitRuntimeTxMeta(ctx, runtimeID, "consensus_transfer", struct {
+	return sc.submitRuntimeTxMeta(ctx, runtimeID, nonce, "consensus_transfer", struct {
 		Transfer staking.Transfer `json:"transfer"`
-		Nonce    uint64           `json:"nonce"`
 	}{
 		Transfer: xfer,
-		Nonce:    nonce,
 	})
 }
 
-func (sc *runtimeImpl) submitRuntimeInMsg(ctx context.Context, id common.Namespace, method string, args interface{}) error {
+func (sc *runtimeImpl) submitRuntimeInMsg(ctx context.Context, id common.Namespace, nonce uint64, method string, args interface{}) error {
 	ctrl := sc.Net.ClientController()
 	if ctrl == nil {
 		return fmt.Errorf("client controller not available")
@@ -454,6 +463,7 @@ func (sc *runtimeImpl) submitRuntimeInMsg(ctx context.Context, id common.Namespa
 		ID:  id,
 		Tag: 42,
 		Data: cbor.Marshal(&TxnCall{
+			Nonce:  nonce,
 			Method: method,
 			Args:   args,
 		}),

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

@@ -122,7 +122,7 @@ func (cli *KeyValueTestClient) workload(ctx context.Context) error {
 	// side `to_string()` returns `8000…0000`, and the original Rust
 	// test client was doing a string compare so no one ever noticed
 	// that truncated values were being compared.
-	if _, err = cli.sc.submitKeyValueRuntimeGetRuntimeIDTx(ctx, runtimeID); err != nil {
+	if _, err = cli.sc.submitKeyValueRuntimeGetRuntimeIDTx(ctx, runtimeID, rng.Uint64()); err != nil {
 		return fmt.Errorf("failed to query remote runtime ID: %w", err)
 	}
 
@@ -217,14 +217,12 @@ func (cli *KeyValueTestClient) workload(ctx context.Context) error {
 		inMsgKey   = "in_msg"
 		inMsgValue = "hello world from inmsg"
 	)
-	err = cli.sc.submitRuntimeInMsg(ctx, runtimeID, "insert", struct {
+	err = cli.sc.submitRuntimeInMsg(ctx, runtimeID, rng.Uint64(), "insert", struct {
 		Key   string `json:"key"`
 		Value string `json:"value"`
-		Nonce uint64 `json:"nonce"`
 	}{
 		Key:   inMsgKey,
 		Value: inMsgValue,
-		Nonce: rng.Uint64(),
 	})
 	if err != nil {
 		return fmt.Errorf("failed to submit 'insert' incoming runtime message: %w", err)
@@ -239,7 +237,7 @@ func (cli *KeyValueTestClient) workload(ctx context.Context) error {
 	}
 
 	cli.sc.Logger.Info("testing consensus queries")
-	if _, err = cli.sc.submitRuntimeTx(ctx, runtimeID, "consensus_accounts", nil); err != nil {
+	if _, err = cli.sc.submitRuntimeTx(ctx, runtimeID, rng.Uint64(), "consensus_accounts", nil); err != nil {
 		return fmt.Errorf("failed to submit consensus_accounts query: %w", err)
 	}
 	// TODO: The old test printed out the accounts and delegations, but
@@ -262,14 +260,12 @@ func (sc *runtimeImpl) submitKeyValueRuntimeInsertTx(
 	key, value string,
 	nonce uint64,
 ) (string, error) {
-	rawRsp, err := sc.submitRuntimeTx(ctx, id, "insert", struct {
+	rawRsp, err := sc.submitRuntimeTx(ctx, id, nonce, "insert", struct {
 		Key   string `json:"key"`
 		Value string `json:"value"`
-		Nonce uint64 `json:"nonce"`
 	}{
 		Key:   key,
 		Value: value,
-		Nonce: nonce,
 	})
 	if err != nil {
 		return "", fmt.Errorf("failed to submit insert tx to runtime: %w", err)
@@ -289,12 +285,10 @@ func (sc *runtimeImpl) submitKeyValueRuntimeGetTx(
 	key string,
 	nonce uint64,
 ) (string, error) {
-	rawRsp, err := sc.submitRuntimeTx(ctx, runtimeID, "get", struct {
-		Key   string `json:"key"`
-		Nonce uint64 `json:"nonce"`
+	rawRsp, err := sc.submitRuntimeTx(ctx, runtimeID, nonce, "get", struct {
+		Key string `json:"key"`
 	}{
-		Key:   key,
-		Nonce: nonce,
+		Key: key,
 	})
 	if err != nil {
 		return "", fmt.Errorf("failed to submit get tx to runtime: %w", err)
@@ -311,8 +305,9 @@ func (sc *runtimeImpl) submitKeyValueRuntimeGetTx(
 func (sc *runtimeImpl) submitKeyValueRuntimeGetRuntimeIDTx(
 	ctx context.Context,
 	id common.Namespace,
+	nonce uint64,
 ) (string, error) {
-	rawRsp, err := sc.submitRuntimeTx(ctx, runtimeID, "get_runtime_id", nil)
+	rawRsp, err := sc.submitRuntimeTx(ctx, runtimeID, nonce, "get_runtime_id", nil)
 	if err != nil {
 		return "", fmt.Errorf("failed to submit get_runtime_id tx to runtime: %w", err)
 	}