abci: add preprocess block

.github/workflows/tests.yml

@@ -23,56 +23,56 @@ jobs:
     runs-on: ubuntu-latest
     timeout-minutes: 5
     steps:
+      - uses: actions/setup-go@v2
+        with:
+          go-version: "1.15"
       - uses: actions/checkout@v2
       - uses: technote-space/get-diff-action@v4
         with:
           PATTERNS: |
             **/**.go
             go.mod
             go.sum
-      - name: Set GOBIN
-        run: |
-          echo "::add-path::$(go env GOPATH)/bin"
       - name: install
         run: make install install_abci
         if: "env.GIT_DIFF != ''"
-      - uses: actions/cache@v2.1.2
+      - uses: actions/cache@v2.1.3
         with:
           path: ~/go/pkg/mod
           key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
           restore-keys: |
             ${{ runner.os }}-go-
         if: env.GIT_DIFF
       # Cache binaries for use by other jobs
-      - uses: actions/cache@v2.1.2
+      - uses: actions/cache@v2.1.3
         with:
           path: ~/go/bin
           key: ${{ runner.os }}-${{ github.sha }}-tm-binary
         if: env.GIT_DIFF
 
   test_abci_apps:
     runs-on: ubuntu-latest
-    needs: Build
+    needs: build
     timeout-minutes: 5
     steps:
+      - uses: actions/setup-go@v2
+        with:
+          go-version: "1.15"
       - uses: actions/checkout@v2
       - uses: technote-space/get-diff-action@v4
         with:
           PATTERNS: |
             **/**.go
             go.mod
             go.sum
-      - name: Set GOBIN
-        run: |
-          echo "::add-path::$(go env GOPATH)/bin"
-      - uses: actions/cache@v2.1.2
+      - uses: actions/cache@v2.1.3
         with:
           path: ~/go/pkg/mod
           key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
           restore-keys: |
             ${{ runner.os }}-go-
         if: env.GIT_DIFF
-      - uses: actions/cache@v2.1.2
+      - uses: actions/cache@v2.1.3
         with:
           path: ~/go/bin
           key: ${{ runner.os }}-${{ github.sha }}-tm-binary
@@ -84,27 +84,27 @@ jobs:
 
   test_abci_cli:
     runs-on: ubuntu-latest
-    needs: Build
+    needs: build
     timeout-minutes: 5
     steps:
+      - uses: actions/setup-go@v2
+        with:
+          go-version: "1.15"
       - uses: actions/checkout@v2
       - uses: technote-space/get-diff-action@v4
         with:
           PATTERNS: |
             **/**.go
             go.mod
             go.sum
-      - name: Set GOBIN
-        run: |
-          echo "::add-path::$(go env GOPATH)/bin"
-      - uses: actions/cache@v2.1.2
+      - uses: actions/cache@v2.1.3
         with:
           path: ~/go/pkg/mod
           key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
           restore-keys: |
             ${{ runner.os }}-go-
         if: env.GIT_DIFF
-      - uses: actions/cache@v2.1.2
+      - uses: actions/cache@v2.1.3
         with:
           path: ~/go/bin
           key: ${{ runner.os }}-${{ github.sha }}-tm-binary
@@ -115,27 +115,27 @@ jobs:
 
   test_apps:
     runs-on: ubuntu-latest
-    needs: Build
+    needs: build
     timeout-minutes: 5
     steps:
+      - uses: actions/setup-go@v2
+        with:
+          go-version: "1.15"
       - uses: actions/checkout@v2
       - uses: technote-space/get-diff-action@v4
         with:
           PATTERNS: |
             **/**.go
             go.mod
             go.sum
-      - name: Set GOBIN
-        run: |
-          echo "::add-path::$(go env GOPATH)/bin"
-      - uses: actions/cache@v2.1.2
+      - uses: actions/cache@v2.1.3
         with:
           path: ~/go/pkg/mod
           key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
           restore-keys: |
             ${{ runner.os }}-go-
         if: env.GIT_DIFF
-      - uses: actions/cache@v2.1.2
+      - uses: actions/cache@v2.1.3
         with:
           path: ~/go/bin
           key: ${{ runner.os }}-${{ github.sha }}-tm-binary

abci/client/client.go

@@ -21,6 +21,7 @@ const (
 // All `Sync` methods return the appropriate protobuf ResponseXxx struct and an error.
 // Note these are client errors, eg. ABCI socket connectivity issues.
 // Application-related errors are reflected in response via ABCI error codes and logs.
+//go:generate mockery --case underscore --name Client
 type Client interface {
 	service.Service
 
@@ -41,6 +42,7 @@ type Client interface {
 	OfferSnapshotAsync(types.RequestOfferSnapshot) *ReqRes
 	LoadSnapshotChunkAsync(types.RequestLoadSnapshotChunk) *ReqRes
 	ApplySnapshotChunkAsync(types.RequestApplySnapshotChunk) *ReqRes
+	PreprocessTxsAsync(types.RequestPreprocessTxs) *ReqRes
 
 	FlushSync() error
 	EchoSync(msg string) (*types.ResponseEcho, error)
@@ -56,6 +58,7 @@ type Client interface {
 	OfferSnapshotSync(types.RequestOfferSnapshot) (*types.ResponseOfferSnapshot, error)
 	LoadSnapshotChunkSync(types.RequestLoadSnapshotChunk) (*types.ResponseLoadSnapshotChunk, error)
 	ApplySnapshotChunkSync(types.RequestApplySnapshotChunk) (*types.ResponseApplySnapshotChunk, error)
+	PreprocessTxsSync(types.RequestPreprocessTxs) (*types.ResponsePreprocessTxs, error)
 }
 
 //----------------------------------------

abci/client/grpc_client.go

@@ -252,6 +252,15 @@ func (cli *grpcClient) ApplySnapshotChunkAsync(params types.RequestApplySnapshot
 	return cli.finishAsyncCall(req, &types.Response{Value: &types.Response_ApplySnapshotChunk{ApplySnapshotChunk: res}})
 }
 
+func (cli *grpcClient) PreprocessTxsAsync(params types.RequestPreprocessTxs) *ReqRes {
+	req := types.ToRequestPreprocessTxs(params)
+	res, err := cli.client.PreprocessTxs(context.Background(), req.GetPreprocessTxs(), grpc.WaitForReady(true))
+	if err != nil {
+		cli.StopForError(err)
+	}
+	return cli.finishAsyncCall(req, &types.Response{Value: &types.Response_PreprocessTxs{PreprocessTxs: res}})
+}
+
 func (cli *grpcClient) finishAsyncCall(req *types.Request, res *types.Response) *ReqRes {
 	reqres := NewReqRes(req)
 	reqres.Response = res // Set response
@@ -350,3 +359,9 @@ func (cli *grpcClient) ApplySnapshotChunkSync(
 	reqres := cli.ApplySnapshotChunkAsync(params)
 	return reqres.Response.GetApplySnapshotChunk(), cli.Error()
 }
+
+func (cli *grpcClient) PreprocessTxsSync(
+	params types.RequestPreprocessTxs) (*types.ResponsePreprocessTxs, error) {
+	reqres := cli.PreprocessTxsAsync(params)
+	return reqres.Response.GetPreprocessTxs(), cli.Error()
+}

abci/client/local_client.go

@@ -190,6 +190,17 @@ func (app *localClient) ApplySnapshotChunkAsync(req types.RequestApplySnapshotCh
 	)
 }
 
+func (app *localClient) PreprocessTxsAsync(req types.RequestPreprocessTxs) *ReqRes {
+	app.mtx.Lock()
+	defer app.mtx.Unlock()
+
+	res := app.Application.PreprocessTxs(req)
+	return app.callback(
+		types.ToRequestPreprocessTxs(req),
+		types.ToResponsePreprocessTx(res),
+	)
+}
+
 //-------------------------------------------------------
 
 func (app *localClient) FlushSync() error {
@@ -298,6 +309,15 @@ func (app *localClient) ApplySnapshotChunkSync(
 	return &res, nil
 }
 
+func (app *localClient) PreprocessTxsSync(
+	req types.RequestPreprocessTxs) (*types.ResponsePreprocessTxs, error) {
+	app.mtx.Lock()
+	defer app.mtx.Unlock()
+
+	res := app.Application.PreprocessTxs(req)
+	return &res, nil
+}
+
 //-------------------------------------------------------
 
 func (app *localClient) callback(req *types.Request, res *types.Response) *ReqRes {

abci/client/socket_client.go

@@ -277,6 +277,10 @@ func (cli *socketClient) ApplySnapshotChunkAsync(req types.RequestApplySnapshotC
 	return cli.queueRequest(types.ToRequestApplySnapshotChunk(req))
 }
 
+func (cli *socketClient) PreprocessTxsAsync(req types.RequestPreprocessTxs) *ReqRes {
+	return cli.queueRequest(types.ToRequestPreprocessTxs(req))
+}
+
 //----------------------------------------
 
 func (cli *socketClient) FlushSync() error {
@@ -406,6 +410,15 @@ func (cli *socketClient) ApplySnapshotChunkSync(
 	return reqres.Response.GetApplySnapshotChunk(), cli.Error()
 }
 
+func (cli *socketClient) PreprocessTxsSync(
+	req types.RequestPreprocessTxs) (*types.ResponsePreprocessTxs, error) {
+	reqres := cli.queueRequest(types.ToRequestPreprocessTxs(req))
+	if err := cli.FlushSync(); err != nil {
+		return nil, err
+	}
+	return reqres.Response.GetPreprocessTxs(), cli.Error()
+}
+
 //----------------------------------------
 
 func (cli *socketClient) queueRequest(req *types.Request) *ReqRes {
@@ -479,6 +492,8 @@ func resMatchesReq(req *types.Request, res *types.Response) (ok bool) {
 		_, ok = res.Value.(*types.Response_ListSnapshots)
 	case *types.Request_OfferSnapshot:
 		_, ok = res.Value.(*types.Response_OfferSnapshot)
+	case *types.Request_PreprocessTxs:
+		_, ok = res.Value.(*types.Response_PreprocessTxs)
 	}
 	return ok
 }

abci/example/counter/counter.go

@@ -84,3 +84,8 @@ func (app *Application) Query(reqQuery types.RequestQuery) types.ResponseQuery {
 		return types.ResponseQuery{Log: fmt.Sprintf("Invalid query path. Expected hash or tx, got %v", reqQuery.Path)}
 	}
 }
+
+func (app *Application) PreprocessTxs(
+	req types.RequestPreprocessTxs) types.ResponsePreprocessTxs {
+	return types.ResponsePreprocessTxs{Txs: req.Txs}
+}

abci/example/kvstore/kvstore.go

@@ -170,3 +170,8 @@ func (app *Application) Query(reqQuery types.RequestQuery) (resQuery types.Respo
 
 	return resQuery
 }
+
+func (app *Application) PreprocessTxs(
+	req types.RequestPreprocessTxs) types.ResponsePreprocessTxs {
+	return types.ResponsePreprocessTxs{Txs: req.Txs}
+}

abci/example/kvstore/persistent_kvstore.go

@@ -166,6 +166,11 @@ func (app *PersistentKVStoreApplication) ApplySnapshotChunk(
 	return types.ResponseApplySnapshotChunk{Result: types.ResponseApplySnapshotChunk_ABORT}
 }
 
+func (app *PersistentKVStoreApplication) PreprocessTxs(
+	req types.RequestPreprocessTxs) types.ResponsePreprocessTxs {
+	return types.ResponsePreprocessTxs{Txs: req.Txs}
+}
+
 //---------------------------------------------
 // update validators
 

abci/server/socket_server.go

@@ -233,6 +233,9 @@ func (s *SocketServer) handleRequest(req *types.Request, responses chan<- *types
 	case *types.Request_ApplySnapshotChunk:
 		res := s.app.ApplySnapshotChunk(*r.ApplySnapshotChunk)
 		responses <- types.ToResponseApplySnapshotChunk(res)
+	case *types.Request_PreprocessTxs:
+		res := s.app.PreprocessTxs(*r.PreprocessTxs)
+		responses <- types.ToResponsePreprocessTx(res)
 	default:
 		responses <- types.ToResponseException("Unknown request")
 	}

abci/types/application.go

@@ -8,6 +8,7 @@ import (
 // to be driven by a blockchain-based replication engine via the ABCI.
 // All methods take a RequestXxx argument and return a ResponseXxx argument,
 // except CheckTx/DeliverTx, which take `tx []byte`, and `Commit`, which takes nothing.
+// nolint:lll // ignore for interface
 type Application interface {
 	// Info/Query Connection
 	Info(RequestInfo) ResponseInfo    // Return application info
@@ -17,11 +18,12 @@ type Application interface {
 	CheckTx(RequestCheckTx) ResponseCheckTx // Validate a tx for the mempool
 
 	// Consensus Connection
-	InitChain(RequestInitChain) ResponseInitChain    // Initialize blockchain w validators/other info from TendermintCore
-	BeginBlock(RequestBeginBlock) ResponseBeginBlock // Signals the beginning of a block
-	DeliverTx(RequestDeliverTx) ResponseDeliverTx    // Deliver a tx for full processing
-	EndBlock(RequestEndBlock) ResponseEndBlock       // Signals the end of a block, returns changes to the validator set
-	Commit() ResponseCommit                          // Commit the state and return the application Merkle root hash
+	InitChain(RequestInitChain) ResponseInitChain             // Initialize blockchain w validators/other info from TendermintCore
+	BeginBlock(RequestBeginBlock) ResponseBeginBlock          // Signals the beginning of a block
+	DeliverTx(RequestDeliverTx) ResponseDeliverTx             // Deliver a tx for full processing
+	EndBlock(RequestEndBlock) ResponseEndBlock                // Signals the end of a block, returns changes to the validator set
+	Commit() ResponseCommit                                   // Commit the state and return the application Merkle root hash
+	PreprocessTxs(RequestPreprocessTxs) ResponsePreprocessTxs // State machine preprocessing of txs
 
 	// State Sync Connection
 	ListSnapshots(RequestListSnapshots) ResponseListSnapshots                // List available snapshots
@@ -90,6 +92,10 @@ func (BaseApplication) ApplySnapshotChunk(req RequestApplySnapshotChunk) Respons
 	return ResponseApplySnapshotChunk{}
 }
 
+func (BaseApplication) PreprocessTxs(req RequestPreprocessTxs) ResponsePreprocessTxs {
+	return ResponsePreprocessTxs{}
+}
+
 //-------------------------------------------------------
 
 // GRPCApplication is a GRPC wrapper for Application
@@ -172,3 +178,9 @@ func (app *GRPCApplication) ApplySnapshotChunk(
 	res := app.app.ApplySnapshotChunk(*req)
 	return &res, nil
 }
+
+func (app *GRPCApplication) PreprocessTxs(
+	ctx context.Context, req *RequestPreprocessTxs) (*ResponsePreprocessTxs, error) {
+	res := app.app.PreprocessTxs(*req)
+	return &res, nil
+}