Merge pull request #2478 from oasislabs/pro-wh/feature/txsource

Add txsource

.buildkite/longtests.pipeline.yml

@@ -0,0 +1,48 @@
+# Copied from pipeline.yml.
+docker_plugin_default_config: &docker_plugin_default_config
+  image: "oasislabs/testing:0.3.0"
+  always_pull: true
+  workdir: /workdir
+  volumes:
+    - /var/lib/buildkite-agent/.coveralls:/root/.coveralls
+    - /var/lib/buildkite-agent/.codecov:/root/.codecov
+    # Shared Rust incremental compile caches.
+    - /tmp/cargo_ic/debug:/tmp/artifacts/debug/incremental
+    - /tmp/cargo_ic/debug_sgx:/tmp/artifacts/x86_64-unknown-linux-sgx/debug/incremental
+    # Shared Rust package checkouts directory.
+    - /tmp/cargo_pkg/git:/root/.cargo/git
+    - /tmp/cargo_pkg/registry:/root/.cargo/registry
+    # Shared Rust SGX standard library artifacts cache.
+    - /tmp/xargo_cache:/root/.xargo
+    # Shared Go package checkouts directory.
+    - /tmp/go_pkg:/root/go/pkg
+    # Intel SGX Application Enclave Services Manager (AESM) daemon running on
+    # the Buildkite host.
+    - /var/run/aesmd/aesm.socket:/var/run/aesmd/aesm.socket
+  # NOTE: When changing the environment variables below, also copy the changes
+  # to the docker_plugin_sgx_config.
+  environment:
+    - "LC_ALL=C.UTF-8"
+    - "LANG=C.UTF-8"
+    - "CARGO_TARGET_DIR=/tmp/artifacts"
+    - "CARGO_INSTALL_ROOT=/root/.cargo"
+    - "GOPROXY=https://proxy.golang.org/"
+  propagate-environment: true
+  unconfined: true
+
+docker_plugin: &docker_plugin
+  oasislabs/docker#v3.0.1-oasis1:
+    <<: *docker_plugin_default_config
+
+steps:
+  - label: Transaction source test
+    timeout_in_minutes: 480
+    command:
+      - make
+      - ./scripts/run-e2e-txsource.sh
+    env:
+      TEST_BASE_DIR: e2e
+    agents:
+      buildkite_agent_size: large
+    plugins:
+      <<: *docker_plugin

.changelog/2478.feature.md

@@ -0,0 +1,7 @@
+Add txsource.
+
+The so-called "txsource" utility introduced in this PR is a starting point for something like a client that sends
+transactions for a long period of time, for the purpose of creating long-running tests.
+
+With this change is a preliminary sample "workload"--a DRBG-backed schedule of transactions--which transfers staking
+tokens around among a set of test accounts.

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

@@ -7,6 +7,7 @@ import (
 	"github.com/oasislabs/oasis-core/go/oasis-node/cmd/debug/byzantine"
 	"github.com/oasislabs/oasis-core/go/oasis-node/cmd/debug/storage"
 	"github.com/oasislabs/oasis-core/go/oasis-node/cmd/debug/tendermint"
+	"github.com/oasislabs/oasis-core/go/oasis-node/cmd/debug/txsource"
 )
 
 var debugCmd = &cobra.Command{
@@ -19,6 +20,7 @@ func Register(parentCmd *cobra.Command) {
 	storage.Register(debugCmd)
 	tendermint.Register(debugCmd)
 	byzantine.Register(debugCmd)
+	txsource.Register(debugCmd)
 
 	parentCmd.AddCommand(debugCmd)
 }

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

@@ -0,0 +1,133 @@
+package txsource
+
+import (
+	"context"
+	"crypto"
+	"fmt"
+	"math/rand"
+
+	"github.com/spf13/cobra"
+	flag "github.com/spf13/pflag"
+	"github.com/spf13/viper"
+
+	"github.com/oasislabs/oasis-core/go/common/crypto/drbg"
+	"github.com/oasislabs/oasis-core/go/common/crypto/mathrand"
+	"github.com/oasislabs/oasis-core/go/common/logging"
+	consensus "github.com/oasislabs/oasis-core/go/consensus/api"
+	"github.com/oasislabs/oasis-core/go/control/api"
+	genesisFile "github.com/oasislabs/oasis-core/go/genesis/file"
+	"github.com/oasislabs/oasis-core/go/oasis-node/cmd/common"
+	cmdFlags "github.com/oasislabs/oasis-core/go/oasis-node/cmd/common/flags"
+	cmdGrpc "github.com/oasislabs/oasis-core/go/oasis-node/cmd/common/grpc"
+	"github.com/oasislabs/oasis-core/go/oasis-node/cmd/debug/txsource/workload"
+	runtimeClient "github.com/oasislabs/oasis-core/go/runtime/client/api"
+)
+
+const (
+	CfgWorkload  = "workload"
+	CfgSeed      = "seed"
+	CfgTimeLimit = "time_limit"
+)
+
+var (
+	logger      = logging.GetLogger("cmd/txsource")
+	txsourceCmd = &cobra.Command{
+		Use:   "txsource",
+		Short: "send random transactions",
+		RunE:  doRun,
+	}
+)
+
+func doRun(cmd *cobra.Command, args []string) error {
+	cmd.SilenceUsage = true
+
+	if err := common.Init(); err != nil {
+		common.EarlyLogAndExit(err)
+	}
+
+	// Set up the time limit.
+	ctx := context.Background()
+	timeLimit := viper.GetDuration(CfgTimeLimit)
+	if timeLimit != 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, timeLimit)
+		defer cancel()
+	}
+
+	// Set up the genesis system for the signature system's chain context.
+	genesis, err := genesisFile.DefaultFileProvider()
+	if err != nil {
+		return fmt.Errorf("genesisFile.DefaultFileProvider: %w", err)
+	}
+	genesisDoc, err := genesis.GetGenesisDocument()
+	if err != nil {
+		return fmt.Errorf("genesis.GetGenesisDocument: %w", err)
+	}
+	logger.Debug("setting chain context", "chain_context", genesisDoc.ChainContext())
+	genesisDoc.SetChainContext()
+
+	// Resolve the workload.
+	name := viper.GetString(CfgWorkload)
+	w, ok := workload.ByName[name]
+	if !ok {
+		return fmt.Errorf("workload %s not found", name)
+	}
+
+	// Set up the deterministic random source.
+	hash := crypto.SHA512
+	seed := []byte(viper.GetString(CfgSeed))
+	src, err := drbg.New(hash, seed, nil, []byte(fmt.Sprintf("txsource workload generator v1, workload %s", name)))
+	if err != nil {
+		return fmt.Errorf("drbg.New: %w", err)
+	}
+	rng := rand.New(mathrand.New(src))
+
+	// Set up the gRPC client.
+	logger.Debug("dialing node", "addr", viper.GetString(cmdGrpc.CfgAddress))
+	conn, err := cmdGrpc.NewClient(cmd)
+	if err != nil {
+		return fmt.Errorf("cmdGrpc.NewClient: %w", err)
+	}
+	defer conn.Close()
+
+	// Set up the consensus client.
+	cnsc := consensus.NewConsensusClient(conn)
+
+	// Set up the runtime client.
+	rtc := runtimeClient.NewRuntimeClient(conn)
+
+	// Wait for sync before transferring control to the workload.
+	ncc := api.NewNodeControllerClient(conn)
+	logger.Debug("waiting for node sync")
+	if err = ncc.WaitSync(context.Background()); err != nil {
+		return fmt.Errorf("node controller client WaitSync: %w", err)
+	}
+	logger.Debug("node synced")
+
+	logger.Debug("entering workload", "name", name)
+	if err = w.Run(ctx, rng, conn, cnsc, rtc); err != nil {
+		return fmt.Errorf("workload %s: %w", name, err)
+	}
+	logger.Debug("workload returned", "name", name)
+
+	return nil
+}
+
+// Register registers the txsource sub-command.
+func Register(parentCmd *cobra.Command) {
+	parentCmd.AddCommand(txsourceCmd)
+}
+
+func init() {
+	fs := flag.NewFlagSet("", flag.ContinueOnError)
+	fs.String(CfgWorkload, workload.NameTransfer, "Name of the workload to run (see source for listing)")
+	fs.String(CfgSeed, "seeeeeeeeeeeeeeeeeeeeeeeeeeeeeed", "Seed to use for randomized workloads")
+	fs.Duration(CfgTimeLimit, 0, "Exit successfully after this long, or 0 to run forever")
+	_ = viper.BindPFlags(fs)
+	txsourceCmd.Flags().AddFlagSet(fs)
+
+	txsourceCmd.Flags().AddFlagSet(cmdGrpc.ClientFlags)
+	txsourceCmd.Flags().AddFlagSet(cmdFlags.DebugTestEntityFlags)
+	txsourceCmd.Flags().AddFlagSet(cmdFlags.GenesisFileFlags)
+	txsourceCmd.Flags().AddFlagSet(cmdFlags.DebugDontBlameOasisFlag)
+}

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

@@ -0,0 +1,33 @@
+package workload
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+
+	"google.golang.org/grpc"
+
+	"github.com/oasislabs/oasis-core/go/common/logging"
+	consensus "github.com/oasislabs/oasis-core/go/consensus/api"
+	runtimeClient "github.com/oasislabs/oasis-core/go/runtime/client/api"
+)
+
+var (
+	_ Workload = runtimePlaceholder{}
+
+	runtimePlaceholderLogger = logging.GetLogger("cmd/txsource/workload/runtimeplaceholder")
+)
+
+type runtimePlaceholder struct{}
+
+func (runtimePlaceholder) Run(_ context.Context, _ *rand.Rand, _ *grpc.ClientConn, _ consensus.ClientBackend, rtc runtimeClient.RuntimeClient) error {
+	ctx := context.Background()
+	var tx *runtimeClient.SubmitTxRequest
+	// Placeholder for sending a runtime transaction from a workload.
+	out, err := rtc.SubmitTx(ctx, tx)
+	if err != nil {
+		return fmt.Errorf("rtc.SubmitTx: %w", err)
+	}
+	runtimePlaceholderLogger.Debug("output", "out", out)
+	return nil
+}

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

@@ -0,0 +1,129 @@
+package workload
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+
+	"google.golang.org/grpc"
+
+	"github.com/oasislabs/oasis-core/go/common/crypto/signature"
+	memorySigner "github.com/oasislabs/oasis-core/go/common/crypto/signature/signers/memory"
+	"github.com/oasislabs/oasis-core/go/common/logging"
+	"github.com/oasislabs/oasis-core/go/common/quantity"
+	consensus "github.com/oasislabs/oasis-core/go/consensus/api"
+	"github.com/oasislabs/oasis-core/go/consensus/api/transaction"
+	runtimeClient "github.com/oasislabs/oasis-core/go/runtime/client/api"
+	staking "github.com/oasislabs/oasis-core/go/staking/api"
+)
+
+const (
+	NameTransfer        = "transfer"
+	TransferNumAccounts = 10
+	TransferAmount      = 1
+)
+
+var transferLogger = logging.GetLogger("cmd/txsource/workload/transfer")
+
+type transfer struct{}
+
+func (transfer) Run(gracefulExit context.Context, rng *rand.Rand, conn *grpc.ClientConn, cnsc consensus.ClientBackend, rtc runtimeClient.RuntimeClient) error {
+	// Load all the keys up front. Like, how annoyed would you be if down the line one of them turned out to be
+	// corrupted or something, ya know?
+	accounts := make([]struct {
+		signer          signature.Signer
+		reckonedNonce   uint64
+		reckonedBalance quantity.Quantity
+	}, TransferNumAccounts)
+	var err error
+	fac := memorySigner.NewFactory()
+	for i := range accounts {
+		accounts[i].signer, err = fac.Generate(signature.SignerEntity, rng)
+		if err != nil {
+			return fmt.Errorf("memory signer factory Generate account %d: %w", i, err)
+		}
+	}
+
+	// Read all the account info up front.
+	ctx := context.Background()
+	stakingClient := staking.NewStakingClient(conn)
+	for i := range accounts {
+		var account *staking.Account
+		account, err = stakingClient.AccountInfo(ctx, &staking.OwnerQuery{
+			Height: consensus.HeightLatest,
+			Owner:  accounts[i].signer.Public(),
+		})
+		if err != nil {
+			return fmt.Errorf("stakingClient.AccountInfo %s: %w", accounts[i].signer.Public(), err)
+		}
+		transferLogger.Debug("account info",
+			"i", i,
+			"pub", accounts[i].signer.Public(),
+			"info", account,
+		)
+		accounts[i].reckonedNonce = account.General.Nonce
+		accounts[i].reckonedBalance = account.General.Balance
+	}
+
+	fee := transaction.Fee{
+		Gas: 10,
+	}
+	var minBalance quantity.Quantity
+	if err = minBalance.FromInt64(TransferAmount); err != nil {
+		return fmt.Errorf("min balance FromInt64 %d: %w", TransferAmount, err)
+	}
+	if err = minBalance.Add(&fee.Amount); err != nil {
+		return fmt.Errorf("min balance %v Add fee amount %v: %w", minBalance, fee.Amount, err)
+	}
+	for {
+		perm := rng.Perm(TransferNumAccounts)
+		fromPermIdx := 0
+		for ; fromPermIdx < TransferNumAccounts; fromPermIdx++ {
+			if accounts[perm[fromPermIdx]].reckonedBalance.Cmp(&minBalance) >= 0 {
+				break
+			}
+		}
+		if fromPermIdx >= TransferNumAccounts {
+			return fmt.Errorf("all accounts %#v have gone broke", accounts)
+		}
+		toPermIdx := (fromPermIdx + 1) % TransferNumAccounts
+		from := &accounts[perm[fromPermIdx]]
+		to := &accounts[perm[toPermIdx]]
+
+		transfer := staking.Transfer{
+			To: to.signer.Public(),
+		}
+		if err = transfer.Tokens.FromInt64(TransferAmount); err != nil {
+			return fmt.Errorf("transfer tokens FromInt64 %d: %w", TransferAmount, err)
+		}
+		tx := staking.NewTransferTx(from.reckonedNonce, &fee, &transfer)
+		signedTx, err := transaction.Sign(from.signer, tx)
+		if err != nil {
+			return fmt.Errorf("transaction.Sign: %w", err)
+		}
+		transferLogger.Debug("submitting transfer",
+			"from", from.signer.Public(),
+			"to", to.signer.Public(),
+		)
+		if err = cnsc.SubmitTx(ctx, signedTx); err != nil {
+			return fmt.Errorf("cnsc.SubmitTx: %w", err)
+		}
+		from.reckonedNonce++
+		if err = from.reckonedBalance.Sub(&fee.Amount); err != nil {
+			return fmt.Errorf("from reckoned balance %v Sub fee amount %v: %w", from.reckonedBalance, fee.Amount, err)
+		}
+		if err = from.reckonedBalance.Sub(&transfer.Tokens); err != nil {
+			return fmt.Errorf("from reckoned balance %v Sub transfer tokens %v: %w", from.reckonedBalance, transfer.Tokens, err)
+		}
+		if err = to.reckonedBalance.Add(&transfer.Tokens); err != nil {
+			return fmt.Errorf("to reckoned balance %v Add transfer tokens %v: %w", to.reckonedBalance, transfer.Tokens, err)
+		}
+
+		select {
+		case <-gracefulExit.Done():
+			transferLogger.Debug("time's up")
+			return nil
+		default:
+		}
+	}
+}

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

@@ -0,0 +1,25 @@
+package workload
+
+import (
+	"context"
+	"math/rand"
+
+	"google.golang.org/grpc"
+
+	consensus "github.com/oasislabs/oasis-core/go/consensus/api"
+	runtimeClient "github.com/oasislabs/oasis-core/go/runtime/client/api"
+)
+
+// Workload is a DRBG-backed schedule of transactions.
+type Workload interface {
+	// Run executes the workload.
+	// If `gracefulExit`'s deadline passes, it is not an error.
+	// Return `nil` after any short-ish amount of time in that case.
+	// Prefer to do at least one "iteration" even so.
+	Run(gracefulExit context.Context, rng *rand.Rand, conn *grpc.ClientConn, cnsc consensus.ClientBackend, rtc runtimeClient.RuntimeClient) error
+}
+
+// ByName is the registry of workloads that you can access with `--workload <name>` on the command line.
+var ByName = map[string]Workload{
+	NameTransfer: transfer{},
+}