abci: implement process proposal to spec

abci/client/client.go

@@ -43,6 +43,7 @@ type Client interface {
 	OfferSnapshotAsync(types.RequestOfferSnapshot) *ReqRes
 	LoadSnapshotChunkAsync(types.RequestLoadSnapshotChunk) *ReqRes
 	ApplySnapshotChunkAsync(types.RequestApplySnapshotChunk) *ReqRes
+	ProcessProposalAsync(types.RequestProcessProposal) *ReqRes
 
 	FlushSync() error
 	EchoSync(msg string) (*types.ResponseEcho, error)
@@ -60,6 +61,7 @@ type Client interface {
 	OfferSnapshotSync(types.RequestOfferSnapshot) (*types.ResponseOfferSnapshot, error)
 	LoadSnapshotChunkSync(types.RequestLoadSnapshotChunk) (*types.ResponseLoadSnapshotChunk, error)
 	ApplySnapshotChunkSync(types.RequestApplySnapshotChunk) (*types.ResponseApplySnapshotChunk, error)
+	ProcessProposalSync(types.RequestProcessProposal) (*types.ResponseProcessProposal, error)
 }
 
 //----------------------------------------

abci/client/grpc_client.go

@@ -309,6 +309,16 @@ func (cli *grpcClient) PrepareProposalAsync(params types.RequestPrepareProposal)
 	return cli.finishAsyncCall(req, &types.Response{Value: &types.Response_PrepareProposal{PrepareProposal: res}})
 }
 
+func (cli *grpcClient) ProcessProposalAsync(params types.RequestProcessProposal) *ReqRes {
+	req := types.ToRequestProcessProposal(params)
+	res, err := cli.client.ProcessProposal(context.Background(), req.GetProcessProposal(), grpc.WaitForReady(true))
+	if err != nil {
+		cli.StopForError(err)
+	}
+
+	return cli.finishAsyncCall(req, &types.Response{Value: &types.Response_ProcessProposal{ProcessProposal: 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(req *types.Request, res *types.Response) *ReqRes {
@@ -432,3 +442,8 @@ func (cli *grpcClient) PrepareProposalSync(
 	reqres := cli.PrepareProposalAsync(params)
 	return cli.finishSyncCall(reqres).GetPrepareProposal(), cli.Error()
 }
+
+func (cli *grpcClient) ProcessProposalSync(params types.RequestProcessProposal) (*types.ResponseProcessProposal, error) {
+	reqres := cli.ProcessProposalAsync(params)
+	return cli.finishSyncCall(reqres).GetProcessProposal(), cli.Error()
+}

abci/client/local_client.go

@@ -218,6 +218,17 @@ func (app *localClient) PrepareProposalAsync(req types.RequestPrepareProposal) *
 	)
 }
 
+func (app *localClient) ProcessProposalAsync(req types.RequestProcessProposal) *ReqRes {
+	app.mtx.Lock()
+	defer app.mtx.Unlock()
+
+	res := app.Application.ProcessProposal(req)
+	return app.callback(
+		types.ToRequestProcessProposal(req),
+		types.ToResponseProcessProposal(res),
+	)
+}
+
 //-------------------------------------------------------
 
 func (app *localClient) FlushSync() error {
@@ -342,6 +353,14 @@ func (app *localClient) PrepareProposalSync(req types.RequestPrepareProposal) (*
 	return &res, nil
 }
 
+func (app *localClient) ProcessProposalSync(req types.RequestProcessProposal) (*types.ResponseProcessProposal, error) {
+	app.mtx.Lock()
+	defer app.mtx.Unlock()
+
+	res := app.Application.ProcessProposal(req)
+	return &res, nil
+}
+
 //-------------------------------------------------------
 
 func (app *localClient) callback(req *types.Request, res *types.Response) *ReqRes {

abci/client/socket_client.go

@@ -283,6 +283,10 @@ func (cli *socketClient) PrepareProposalAsync(req types.RequestPrepareProposal)
 	return cli.queueRequest(types.ToRequestPrepareProposal(req))
 }
 
+func (cli *socketClient) ProcessProposalAsync(req types.RequestProcessProposal) *ReqRes {
+	return cli.queueRequest(types.ToRequestProcessProposal(req))
+}
+
 //----------------------------------------
 
 func (cli *socketClient) FlushSync() error {
@@ -430,6 +434,15 @@ func (cli *socketClient) PrepareProposalSync(req types.RequestPrepareProposal) (
 	return reqres.Response.GetPrepareProposal(), cli.Error()
 }
 
+func (cli *socketClient) ProcessProposalSync(req types.RequestProcessProposal) (*types.ResponseProcessProposal, error) {
+	reqres := cli.queueRequest(types.ToRequestProcessProposal(req))
+	if err := cli.FlushSync(); err != nil {
+		return nil, err
+	}
+
+	return reqres.Response.GetProcessProposal(), cli.Error()
+}
+
 //----------------------------------------
 
 func (cli *socketClient) queueRequest(req *types.Request) *ReqRes {
@@ -507,6 +520,8 @@ func resMatchesReq(req *types.Request, res *types.Response) (ok bool) {
 		_, ok = res.Value.(*types.Response_OfferSnapshot)
 	case *types.Request_PrepareProposal:
 		_, ok = res.Value.(*types.Response_PrepareProposal)
+	case *types.Request_ProcessProposal:
+		_, ok = res.Value.(*types.Response_ProcessProposal)
 	}
 	return ok
 }

abci/example/kvstore/kvstore.go

@@ -170,3 +170,13 @@ func (app *Application) Query(reqQuery types.RequestQuery) (resQuery types.Respo
 
 	return resQuery
 }
+
+func (app *Application) ProcessProposal(
+	req types.RequestProcessProposal) types.ResponseProcessProposal {
+	for _, tx := range req.Txs {
+		if len(tx) == 0 {
+			return types.ResponseProcessProposal{Status: types.ResponseProcessProposal_REJECT}
+		}
+	}
+	return types.ResponseProcessProposal{Status: types.ResponseProcessProposal_ACCEPT}
+}

abci/example/kvstore/persistent_kvstore.go

@@ -179,6 +179,16 @@ func (app *PersistentKVStoreApplication) PrepareProposal(
 	return types.ResponsePrepareProposal{TxRecords: app.substPrepareTx(req.Txs, req.MaxTxBytes)}
 }
 
+func (app *PersistentKVStoreApplication) ProcessProposal(
+	req types.RequestProcessProposal) types.ResponseProcessProposal {
+	for _, tx := range req.Txs {
+		if len(tx) == 0 {
+			return types.ResponseProcessProposal{Status: types.ResponseProcessProposal_REJECT}
+		}
+	}
+	return types.ResponseProcessProposal{Status: types.ResponseProcessProposal_ACCEPT}
+}
+
 //---------------------------------------------
 // update validators
 

abci/server/socket_server.go

@@ -233,6 +233,9 @@ func (s *SocketServer) handleRequest(req *types.Request, responses chan<- *types
 	case *types.Request_PrepareProposal:
 		res := s.app.PrepareProposal(*r.PrepareProposal)
 		responses <- types.ToResponsePrepareProposal(res)
+	case *types.Request_ProcessProposal:
+		res := s.app.ProcessProposal(*r.ProcessProposal)
+		responses <- types.ToResponseProcessProposal(res)
 	case *types.Request_LoadSnapshotChunk:
 		res := s.app.LoadSnapshotChunk(*r.LoadSnapshotChunk)
 		responses <- types.ToResponseLoadSnapshotChunk(res)

abci/types/application.go

@@ -4,7 +4,8 @@ import (
 	context "golang.org/x/net/context"
 )
 
-//go:generate mockery --case underscore --name Application
+//go:generate ../../scripts/mockery_generate.sh Application
+
 // Application is an interface that enables any finite, deterministic state machine
 // to be driven by a blockchain-based replication engine via the ABCI.
 // All methods take a RequestXxx argument and return a ResponseXxx argument,
@@ -21,6 +22,7 @@ type Application interface {
 	// Consensus Connection
 	InitChain(RequestInitChain) ResponseInitChain // Initialize blockchain w validators/other info from TendermintCore
 	PrepareProposal(RequestPrepareProposal) ResponsePrepareProposal
+	ProcessProposal(RequestProcessProposal) ResponseProcessProposal
 	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
@@ -113,6 +115,11 @@ func (BaseApplication) PrepareProposal(req RequestPrepareProposal) ResponsePrepa
 	return ResponsePrepareProposal{TxRecords: trs}
 }
 
+func (BaseApplication) ProcessProposal(req RequestProcessProposal) ResponseProcessProposal {
+	return ResponseProcessProposal{
+		Status: ResponseProcessProposal_ACCEPT}
+}
+
 //-------------------------------------------------------
 
 // GRPCApplication is a GRPC wrapper for Application
@@ -206,3 +213,9 @@ func (app *GRPCApplication) PrepareProposal(
 	res := app.app.PrepareProposal(*req)
 	return &res, nil
 }
+
+func (app *GRPCApplication) ProcessProposal(
+	ctx context.Context, req *RequestProcessProposal) (*ResponseProcessProposal, error) {
+	res := app.app.ProcessProposal(*req)
+	return &res, nil
+}

abci/types/messages.go

@@ -165,6 +165,12 @@ func ToRequestPrepareProposal(req RequestPrepareProposal) *Request {
 	}
 }
 
+func ToRequestProcessProposal(req RequestProcessProposal) *Request {
+	return &Request{
+		Value: &Request_ProcessProposal{&req},
+	}
+}
+
 //----------------------------------------
 
 func ToResponseException(errStr string) *Response {
@@ -268,3 +274,9 @@ func ToResponsePrepareProposal(res ResponsePrepareProposal) *Response {
 		Value: &Response_PrepareProposal{&res},
 	}
 }
+
+func ToResponseProcessProposal(res ResponseProcessProposal) *Response {
+	return &Response{
+		Value: &Response_ProcessProposal{&res},
+	}
+}

abci/types/mocks/base.go

@@ -84,6 +84,17 @@ func (m BaseMock) PrepareProposal(input types.RequestPrepareProposal) types.Resp
 	return ret
 }
 
+func (m BaseMock) ProcessProposal(input types.RequestProcessProposal) types.ResponseProcessProposal {
+	var ret types.ResponseProcessProposal
+	defer func() {
+		if r := recover(); r != nil {
+			ret = m.base.ProcessProposal(input)
+		}
+	}()
+	ret = m.Application.ProcessProposal(input)
+	return ret
+}
+
 // Commit the state and return the application Merkle root hash
 func (m BaseMock) Commit() types.ResponseCommit {
 	var ret types.ResponseCommit

abci/types/types.go

@@ -41,6 +41,16 @@ func (r ResponseQuery) IsErr() bool {
 	return r.Code != CodeTypeOK
 }
 
+// IsAccepted returns true if Code is ACCEPT
+func (r ResponseProcessProposal) IsAccepted() bool {
+	return r.Status == ResponseProcessProposal_ACCEPT
+}
+
+// IsStatusUnknown returns true if Code is UNKNOWN
+func (r ResponseProcessProposal) IsStatusUnknown() bool {
+	return r.Status == ResponseProcessProposal_UNKNOWN
+}
+
 //---------------------------------------------------------------------------
 // override JSON marshaling so we emit defaults (ie. disable omitempty)