abci: add preprocess block (#110)

* version docs * stash commits * fix url * remove master from sidebar * fix build errors * move to consensus connection * add process * add proof to block * add block metadata in beginBlcok * fix testing * regenerate mocks * add in needed proto changes * regenerate mocks * fix makeblock calls * Apply suggestions from code review * fix rpc tests * fix linting and ci errors * fix consensus tests * add preprocess to e2e * add preprocess to counter app * replace meta_data with messages * fix linting * Update state/execution.go Co-authored-by: Ismail Khoffi <Ismail.Khoffi@gmail.com> * fix comment * fix e2e tests Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com> Co-authored-by: Ismail Khoffi <Ismail.Khoffi@gmail.com> fix test so that it doesn't modify block data after generating the hash fix missed make block api change run make mockery add some docs fix hardcoded test fix linter fix hard coded test
celestiaorg · Sep 23, 2021 · 9133b9c · 9133b9c
1 parent d9c4d6b
commit 9133b9c
Show file tree

Hide file tree

Showing 23 changed files with 1,120 additions and 234 deletions.
diff --git a/abci/client/client.go b/abci/client/client.go
@@ -46,6 +46,7 @@ type Client interface {
 	OfferSnapshotAsync(context.Context, types.RequestOfferSnapshot) (*ReqRes, error)
 	LoadSnapshotChunkAsync(context.Context, types.RequestLoadSnapshotChunk) (*ReqRes, error)
 	ApplySnapshotChunkAsync(context.Context, types.RequestApplySnapshotChunk) (*ReqRes, error)
+	PreprocessTxsAsync(context.Context, types.RequestPreprocessTxs) (*ReqRes, error)
 
 	// Synchronous requests
 	FlushSync(context.Context) error
@@ -62,6 +63,7 @@ type Client interface {
 	OfferSnapshotSync(context.Context, types.RequestOfferSnapshot) (*types.ResponseOfferSnapshot, error)
 	LoadSnapshotChunkSync(context.Context, types.RequestLoadSnapshotChunk) (*types.ResponseLoadSnapshotChunk, error)
 	ApplySnapshotChunkSync(context.Context, types.RequestApplySnapshotChunk) (*types.ResponseApplySnapshotChunk, error)
+	PreprocessTxsSync(context.Context, types.RequestPreprocessTxs) (*types.ResponsePreprocessTxs, error)
 }
 
 //----------------------------------------

diff --git a/abci/client/grpc_client.go b/abci/client/grpc_client.go
@@ -314,6 +314,15 @@ func (cli *grpcClient) ApplySnapshotChunkAsync(
 	)
 }
 
+func (cli *grpcClient) PreprocessTxsAsync(ctx context.Context, params types.RequestPreprocessTxs) (*ReqRes, error) {
+	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(ctx, req, &types.Response{Value: &types.Response_PreprocessTxs{PreprocessTxs: res}})
+}
+
 // finishAsyncCall creates a ReqRes for an async call, and immediately populates it
 // with the response. We don't complete it until it's been ordered via the channel.
 func (cli *grpcClient) finishAsyncCall(ctx context.Context, req *types.Request, res *types.Response) (*ReqRes, error) {
@@ -504,3 +513,15 @@ func (cli *grpcClient) ApplySnapshotChunkSync(
 	}
 	return cli.finishSyncCall(reqres).GetApplySnapshotChunk(), cli.Error()
 }
+
+func (cli *grpcClient) PreprocessTxsSync(
+	ctx context.Context,
+	params types.RequestPreprocessTxs,
+) (*types.ResponsePreprocessTxs, error) {
+
+	reqres, err := cli.PreprocessTxsAsync(ctx, params)
+	if err != nil {
+		return nil, err
+	}
+	return reqres.Response.GetPreprocessTxs(), cli.Error()
+}
diff --git a/abci/client/local_client.go b/abci/client/local_client.go
@@ -204,6 +204,17 @@ func (app *localClient) ApplySnapshotChunkAsync(
 	), nil
 }
 
+func (app *localClient) PreprocessTxsAsync(ctx context.Context, req types.RequestPreprocessTxs) (*ReqRes, error) {
+	app.mtx.Lock()
+	defer app.mtx.Unlock()
+
+	res := app.Application.PreprocessTxs(req)
+	return app.callback(
+		types.ToRequestPreprocessTxs(req),
+		types.ToResponsePreprocessTx(res),
+	), nil
+}
+
 //-------------------------------------------------------
 
 func (app *localClient) FlushSync(ctx context.Context) error {
@@ -346,6 +357,17 @@ func (app *localClient) ApplySnapshotChunkSync(
 	return &res, nil
 }
 
+func (app *localClient) PreprocessTxsSync(
+	ctx context.Context,
+	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 {

diff --git a/abci/client/mocks/client.go b/abci/client/mocks/client.go
diff --git a/abci/client/socket_client.go b/abci/client/socket_client.go
@@ -295,6 +295,10 @@ func (cli *socketClient) ApplySnapshotChunkAsync(
 	return cli.queueRequestAsync(ctx, types.ToRequestApplySnapshotChunk(req))
 }
 
+func (cli *socketClient) PreprocessTxsAsync(ctx context.Context, req types.RequestPreprocessTxs) (*ReqRes, error) {
+	return cli.queueRequestAsync(ctx, types.ToRequestPreprocessTxs(req))
+}
+
 //----------------------------------------
 
 func (cli *socketClient) FlushSync(ctx context.Context) error {
@@ -465,6 +469,17 @@ func (cli *socketClient) ApplySnapshotChunkSync(
 	return reqres.Response.GetApplySnapshotChunk(), nil
 }
 
+func (cli *socketClient) PreprocessTxsSync(
+	ctx context.Context,
+	req types.RequestPreprocessTxs,
+) (*types.ResponsePreprocessTxs, error) {
+	reqres, err := cli.queueRequestAndFlushSync(ctx, types.ToRequestPreprocessTxs(req))
+	if err != nil {
+		return nil, err
+	}
+	return reqres.Response.GetPreprocessTxs(), nil
+}
+
 //----------------------------------------
 
 // queueRequest enqueues req onto the queue. If the queue is full, it ether
@@ -591,6 +606,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
 }

diff --git a/abci/example/kvstore/kvstore.go b/abci/example/kvstore/kvstore.go
@@ -171,3 +171,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}
+}
diff --git a/abci/example/kvstore/persistent_kvstore.go b/abci/example/kvstore/persistent_kvstore.go
@@ -170,6 +170,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
 

diff --git a/abci/server/socket_server.go b/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")
 	}

diff --git a/abci/types/application.go b/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
+}
diff --git a/abci/types/messages.go b/abci/types/messages.go
@@ -110,6 +110,12 @@ func ToRequestApplySnapshotChunk(req RequestApplySnapshotChunk) *Request {
 	}
 }
 
+func ToRequestPreprocessTxs(res RequestPreprocessTxs) *Request {
+	return &Request{
+		Value: &Request_PreprocessTxs{&res},
+	}
+}
+
 //----------------------------------------
 
 func ToResponseException(errStr string) *Response {
@@ -200,3 +206,9 @@ func ToResponseApplySnapshotChunk(res ResponseApplySnapshotChunk) *Response {
 		Value: &Response_ApplySnapshotChunk{&res},
 	}
 }
+
+func ToResponsePreprocessTx(res ResponsePreprocessTxs) *Response {
+	return &Response{
+		Value: &Response_PreprocessTxs{&res},
+	}
+}