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 {

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/types/application.go

@@ -14,7 +14,8 @@ type Application interface {
 	Query(RequestQuery) ResponseQuery // Query for state
 
 	// Mempool Connection
-	CheckTx(RequestCheckTx) ResponseCheckTx // Validate a tx for the mempool
+	CheckTx(RequestCheckTx) ResponseCheckTx                   // Validate a tx for the mempool
+	PreprocessTxs(RequestPreprocessTxs) ResponsePreprocessTxs // State machine preprocessing of txs
 
 	// Consensus Connection
 	InitChain(RequestInitChain) ResponseInitChain    // Initialize blockchain w validators/other info from TendermintCore
@@ -90,6 +91,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 +177,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
+}

abci/types/messages.go

@@ -153,6 +153,12 @@ func ToRequestApplySnapshotChunk(req RequestApplySnapshotChunk) *Request {
 	}
 }
 
+func ToRequestPreprocessTxs(res RequestPreprocessTxs) *Request {
+	return &Request{
+		Value: &Request_PreprocessTxs{&res},
+	}
+}
+
 //----------------------------------------
 
 func ToResponseException(errStr string) *Response {
@@ -243,3 +249,9 @@ func ToResponseApplySnapshotChunk(res ResponseApplySnapshotChunk) *Response {
 		Value: &Response_ApplySnapshotChunk{&res},
 	}
 }
+
+func ToResponsePreprocessTx(res ResponsePreprocessTxs) *Response {
+	return &Response{
+		Value: &Response_PreprocessTxs{&res},
+	}
+}