Merge branch 'master' into callum/merge-spec

This commit is contained in:
Callum Waters
2022-02-15 11:55:29 +01:00
166 changed files with 2861 additions and 3975 deletions
+36
View File
@@ -0,0 +1,36 @@
# Runs randomly generated E2E testnets nightly on master
# manually run e2e tests
name: e2e-manual
on:
workflow_dispatch:
jobs:
e2e-nightly-test:
# Run parallel jobs for the listed testnet groups (must match the
# ./build/generator -g flag)
strategy:
fail-fast: false
matrix:
group: ['00', '01', '02', '03']
runs-on: ubuntu-latest
timeout-minutes: 60
steps:
- uses: actions/setup-go@v2
with:
go-version: '1.17'
- uses: actions/checkout@v2.4.0
- name: Build
working-directory: test/e2e
# Run make jobs in parallel, since we can't run steps in parallel.
run: make -j2 docker generator runner tests
- name: Generate testnets
working-directory: test/e2e
# When changing -g, also change the matrix groups above
run: ./build/generator -g 4 -d networks/nightly/
- name: Run ${{ matrix.p2p }} p2p testnets
working-directory: test/e2e
run: ./run-multiple.sh networks/nightly/*-group${{ matrix.group }}-*.toml
-17
View File
@@ -6,7 +6,6 @@
name: e2e-nightly-34x
on:
workflow_dispatch: # allow running workflow manually, in theory
schedule:
- cron: '0 2 * * *'
@@ -58,19 +57,3 @@ jobs:
SLACK_COLOR: danger
SLACK_MESSAGE: Nightly E2E tests failed on v0.34.x
SLACK_FOOTER: ''
e2e-nightly-success: # may turn this off once they seem to pass consistently
needs: e2e-nightly-test
if: ${{ success() }}
runs-on: ubuntu-latest
steps:
- name: Notify Slack on success
uses: rtCamp/action-slack-notify@12e36fc18b0689399306c2e0b3e0f2978b7f1ee7
env:
SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK }}
SLACK_CHANNEL: tendermint-internal
SLACK_USERNAME: Nightly E2E Tests
SLACK_ICON_EMOJI: ':white_check_mark:'
SLACK_COLOR: good
SLACK_MESSAGE: Nightly E2E tests passed on v0.34.x
SLACK_FOOTER: ''
-1
View File
@@ -5,7 +5,6 @@
name: e2e-nightly-35x
on:
workflow_dispatch: # allow running workflow manually
schedule:
- cron: '0 2 * * *'
-1
View File
@@ -5,7 +5,6 @@
name: e2e-nightly-master
on:
workflow_dispatch: # allow running workflow manually
schedule:
- cron: '0 2 * * *'
+1 -4
View File
@@ -33,14 +33,12 @@ type Client interface {
// Asynchronous requests
FlushAsync(context.Context) (*ReqRes, error)
DeliverTxAsync(context.Context, types.RequestDeliverTx) (*ReqRes, error)
CheckTxAsync(context.Context, types.RequestCheckTx) (*ReqRes, error)
// Synchronous requests
Flush(context.Context) error
Echo(ctx context.Context, msg string) (*types.ResponseEcho, error)
Info(context.Context, types.RequestInfo) (*types.ResponseInfo, error)
DeliverTx(context.Context, types.RequestDeliverTx) (*types.ResponseDeliverTx, error)
CheckTx(context.Context, types.RequestCheckTx) (*types.ResponseCheckTx, error)
Query(context.Context, types.RequestQuery) (*types.ResponseQuery, error)
Commit(context.Context) (*types.ResponseCommit, error)
@@ -49,8 +47,7 @@ type Client interface {
ProcessProposal(context.Context, types.RequestProcessProposal) (*types.ResponseProcessProposal, error)
ExtendVote(context.Context, types.RequestExtendVote) (*types.ResponseExtendVote, error)
VerifyVoteExtension(context.Context, types.RequestVerifyVoteExtension) (*types.ResponseVerifyVoteExtension, error)
BeginBlock(context.Context, types.RequestBeginBlock) (*types.ResponseBeginBlock, error)
EndBlock(context.Context, types.RequestEndBlock) (*types.ResponseEndBlock, error)
FinalizeBlock(context.Context, types.RequestFinalizeBlock) (*types.ResponseFinalizeBlock, error)
ListSnapshots(context.Context, types.RequestListSnapshots) (*types.ResponseListSnapshots, error)
OfferSnapshot(context.Context, types.RequestOfferSnapshot) (*types.ResponseOfferSnapshot, error)
LoadSnapshotChunk(context.Context, types.RequestLoadSnapshotChunk) (*types.ResponseLoadSnapshotChunk, error)
+8 -40
View File
@@ -193,16 +193,6 @@ func (cli *grpcClient) FlushAsync(ctx context.Context) (*ReqRes, error) {
return cli.finishAsyncCall(ctx, req, &types.Response{Value: &types.Response_Flush{Flush: res}})
}
// NOTE: call is synchronous, use ctx to break early if needed
func (cli *grpcClient) DeliverTxAsync(ctx context.Context, params types.RequestDeliverTx) (*ReqRes, error) {
req := types.ToRequestDeliverTx(params)
res, err := cli.client.DeliverTx(ctx, req.GetDeliverTx(), grpc.WaitForReady(true))
if err != nil {
return nil, err
}
return cli.finishAsyncCall(ctx, req, &types.Response{Value: &types.Response_DeliverTx{DeliverTx: res}})
}
// NOTE: call is synchronous, use ctx to break early if needed
func (cli *grpcClient) CheckTxAsync(ctx context.Context, params types.RequestCheckTx) (*ReqRes, error) {
req := types.ToRequestCheckTx(params)
@@ -271,18 +261,6 @@ func (cli *grpcClient) Info(
return cli.client.Info(ctx, req.GetInfo(), grpc.WaitForReady(true))
}
func (cli *grpcClient) DeliverTx(
ctx context.Context,
params types.RequestDeliverTx,
) (*types.ResponseDeliverTx, error) {
reqres, err := cli.DeliverTxAsync(ctx, params)
if err != nil {
return nil, err
}
return cli.finishSyncCall(reqres).GetDeliverTx(), cli.Error()
}
func (cli *grpcClient) CheckTx(
ctx context.Context,
params types.RequestCheckTx,
@@ -317,24 +295,6 @@ func (cli *grpcClient) InitChain(
return cli.client.InitChain(ctx, req.GetInitChain(), grpc.WaitForReady(true))
}
func (cli *grpcClient) BeginBlock(
ctx context.Context,
params types.RequestBeginBlock,
) (*types.ResponseBeginBlock, error) {
req := types.ToRequestBeginBlock(params)
return cli.client.BeginBlock(ctx, req.GetBeginBlock(), grpc.WaitForReady(true))
}
func (cli *grpcClient) EndBlock(
ctx context.Context,
params types.RequestEndBlock,
) (*types.ResponseEndBlock, error) {
req := types.ToRequestEndBlock(params)
return cli.client.EndBlock(ctx, req.GetEndBlock(), grpc.WaitForReady(true))
}
func (cli *grpcClient) ListSnapshots(
ctx context.Context,
params types.RequestListSnapshots,
@@ -400,3 +360,11 @@ func (cli *grpcClient) VerifyVoteExtension(
req := types.ToRequestVerifyVoteExtension(params)
return cli.client.VerifyVoteExtension(ctx, req.GetVerifyVoteExtension(), grpc.WaitForReady(true))
}
func (cli *grpcClient) FinalizeBlock(
ctx context.Context,
params types.RequestFinalizeBlock) (*types.ResponseFinalizeBlock, error) {
req := types.ToRequestFinalizeBlock(params)
return cli.client.FinalizeBlock(ctx, req.GetFinalizeBlock(), grpc.WaitForReady(true))
}
+11 -47
View File
@@ -58,17 +58,6 @@ func (app *localClient) FlushAsync(ctx context.Context) (*ReqRes, error) {
return newLocalReqRes(types.ToRequestFlush(), nil), nil
}
func (app *localClient) DeliverTxAsync(ctx context.Context, params types.RequestDeliverTx) (*ReqRes, error) {
app.mtx.Lock()
defer app.mtx.Unlock()
res := app.Application.DeliverTx(params)
return app.callback(
types.ToRequestDeliverTx(params),
types.ToResponseDeliverTx(res),
), nil
}
func (app *localClient) CheckTxAsync(ctx context.Context, req types.RequestCheckTx) (*ReqRes, error) {
app.mtx.Lock()
defer app.mtx.Unlock()
@@ -98,18 +87,6 @@ func (app *localClient) Info(ctx context.Context, req types.RequestInfo) (*types
return &res, nil
}
func (app *localClient) DeliverTx(
ctx context.Context,
req types.RequestDeliverTx,
) (*types.ResponseDeliverTx, error) {
app.mtx.Lock()
defer app.mtx.Unlock()
res := app.Application.DeliverTx(req)
return &res, nil
}
func (app *localClient) CheckTx(
ctx context.Context,
req types.RequestCheckTx,
@@ -152,30 +129,6 @@ func (app *localClient) InitChain(
return &res, nil
}
func (app *localClient) BeginBlock(
ctx context.Context,
req types.RequestBeginBlock,
) (*types.ResponseBeginBlock, error) {
app.mtx.Lock()
defer app.mtx.Unlock()
res := app.Application.BeginBlock(req)
return &res, nil
}
func (app *localClient) EndBlock(
ctx context.Context,
req types.RequestEndBlock,
) (*types.ResponseEndBlock, error) {
app.mtx.Lock()
defer app.mtx.Unlock()
res := app.Application.EndBlock(req)
return &res, nil
}
func (app *localClient) ListSnapshots(
ctx context.Context,
req types.RequestListSnapshots,
@@ -266,6 +219,17 @@ func (app *localClient) VerifyVoteExtension(
return &res, nil
}
func (app *localClient) FinalizeBlock(
ctx context.Context,
req types.RequestFinalizeBlock) (*types.ResponseFinalizeBlock, error) {
app.mtx.Lock()
defer app.mtx.Unlock()
res := app.Application.FinalizeBlock(req)
return &res, nil
}
//-------------------------------------------------------
func (app *localClient) callback(req *types.Request, res *types.Response) *ReqRes {
+23 -92
View File
@@ -40,29 +40,6 @@ func (_m *Client) ApplySnapshotChunk(_a0 context.Context, _a1 types.RequestApply
return r0, r1
}
// BeginBlock provides a mock function with given fields: _a0, _a1
func (_m *Client) BeginBlock(_a0 context.Context, _a1 types.RequestBeginBlock) (*types.ResponseBeginBlock, error) {
ret := _m.Called(_a0, _a1)
var r0 *types.ResponseBeginBlock
if rf, ok := ret.Get(0).(func(context.Context, types.RequestBeginBlock) *types.ResponseBeginBlock); ok {
r0 = rf(_a0, _a1)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*types.ResponseBeginBlock)
}
}
var r1 error
if rf, ok := ret.Get(1).(func(context.Context, types.RequestBeginBlock) error); ok {
r1 = rf(_a0, _a1)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// CheckTx provides a mock function with given fields: _a0, _a1
func (_m *Client) CheckTx(_a0 context.Context, _a1 types.RequestCheckTx) (*types.ResponseCheckTx, error) {
ret := _m.Called(_a0, _a1)
@@ -132,52 +109,6 @@ func (_m *Client) Commit(_a0 context.Context) (*types.ResponseCommit, error) {
return r0, r1
}
// DeliverTx provides a mock function with given fields: _a0, _a1
func (_m *Client) DeliverTx(_a0 context.Context, _a1 types.RequestDeliverTx) (*types.ResponseDeliverTx, error) {
ret := _m.Called(_a0, _a1)
var r0 *types.ResponseDeliverTx
if rf, ok := ret.Get(0).(func(context.Context, types.RequestDeliverTx) *types.ResponseDeliverTx); ok {
r0 = rf(_a0, _a1)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*types.ResponseDeliverTx)
}
}
var r1 error
if rf, ok := ret.Get(1).(func(context.Context, types.RequestDeliverTx) error); ok {
r1 = rf(_a0, _a1)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// DeliverTxAsync provides a mock function with given fields: _a0, _a1
func (_m *Client) DeliverTxAsync(_a0 context.Context, _a1 types.RequestDeliverTx) (*abciclient.ReqRes, error) {
ret := _m.Called(_a0, _a1)
var r0 *abciclient.ReqRes
if rf, ok := ret.Get(0).(func(context.Context, types.RequestDeliverTx) *abciclient.ReqRes); ok {
r0 = rf(_a0, _a1)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*abciclient.ReqRes)
}
}
var r1 error
if rf, ok := ret.Get(1).(func(context.Context, types.RequestDeliverTx) error); ok {
r1 = rf(_a0, _a1)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// Echo provides a mock function with given fields: ctx, msg
func (_m *Client) Echo(ctx context.Context, msg string) (*types.ResponseEcho, error) {
ret := _m.Called(ctx, msg)
@@ -201,29 +132,6 @@ func (_m *Client) Echo(ctx context.Context, msg string) (*types.ResponseEcho, er
return r0, r1
}
// EndBlock provides a mock function with given fields: _a0, _a1
func (_m *Client) EndBlock(_a0 context.Context, _a1 types.RequestEndBlock) (*types.ResponseEndBlock, error) {
ret := _m.Called(_a0, _a1)
var r0 *types.ResponseEndBlock
if rf, ok := ret.Get(0).(func(context.Context, types.RequestEndBlock) *types.ResponseEndBlock); ok {
r0 = rf(_a0, _a1)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*types.ResponseEndBlock)
}
}
var r1 error
if rf, ok := ret.Get(1).(func(context.Context, types.RequestEndBlock) error); ok {
r1 = rf(_a0, _a1)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// Error provides a mock function with given fields:
func (_m *Client) Error() error {
ret := _m.Called()
@@ -261,6 +169,29 @@ func (_m *Client) ExtendVote(_a0 context.Context, _a1 types.RequestExtendVote) (
return r0, r1
}
// FinalizeBlock provides a mock function with given fields: _a0, _a1
func (_m *Client) FinalizeBlock(_a0 context.Context, _a1 types.RequestFinalizeBlock) (*types.ResponseFinalizeBlock, error) {
ret := _m.Called(_a0, _a1)
var r0 *types.ResponseFinalizeBlock
if rf, ok := ret.Get(0).(func(context.Context, types.RequestFinalizeBlock) *types.ResponseFinalizeBlock); ok {
r0 = rf(_a0, _a1)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*types.ResponseFinalizeBlock)
}
}
var r1 error
if rf, ok := ret.Get(1).(func(context.Context, types.RequestFinalizeBlock) error); ok {
r1 = rf(_a0, _a1)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// Flush provides a mock function with given fields: _a0
func (_m *Client) Flush(_a0 context.Context) error {
ret := _m.Called(_a0)
+13 -46
View File
@@ -226,10 +226,6 @@ func (cli *socketClient) FlushAsync(ctx context.Context) (*ReqRes, error) {
return cli.queueRequestAsync(ctx, types.ToRequestFlush())
}
func (cli *socketClient) DeliverTxAsync(ctx context.Context, req types.RequestDeliverTx) (*ReqRes, error) {
return cli.queueRequestAsync(ctx, types.ToRequestDeliverTx(req))
}
func (cli *socketClient) CheckTxAsync(ctx context.Context, req types.RequestCheckTx) (*ReqRes, error) {
return cli.queueRequestAsync(ctx, types.ToRequestCheckTx(req))
}
@@ -280,18 +276,6 @@ func (cli *socketClient) Info(
return reqres.Response.GetInfo(), nil
}
func (cli *socketClient) DeliverTx(
ctx context.Context,
req types.RequestDeliverTx,
) (*types.ResponseDeliverTx, error) {
reqres, err := cli.queueRequestAndFlush(ctx, types.ToRequestDeliverTx(req))
if err != nil {
return nil, err
}
return reqres.Response.GetDeliverTx(), nil
}
func (cli *socketClient) CheckTx(
ctx context.Context,
req types.RequestCheckTx,
@@ -334,30 +318,6 @@ func (cli *socketClient) InitChain(
return reqres.Response.GetInitChain(), nil
}
func (cli *socketClient) BeginBlock(
ctx context.Context,
req types.RequestBeginBlock,
) (*types.ResponseBeginBlock, error) {
reqres, err := cli.queueRequestAndFlush(ctx, types.ToRequestBeginBlock(req))
if err != nil {
return nil, err
}
return reqres.Response.GetBeginBlock(), nil
}
func (cli *socketClient) EndBlock(
ctx context.Context,
req types.RequestEndBlock,
) (*types.ResponseEndBlock, error) {
reqres, err := cli.queueRequestAndFlush(ctx, types.ToRequestEndBlock(req))
if err != nil {
return nil, err
}
return reqres.Response.GetEndBlock(), nil
}
func (cli *socketClient) ListSnapshots(
ctx context.Context,
req types.RequestListSnapshots,
@@ -449,6 +409,17 @@ func (cli *socketClient) VerifyVoteExtension(
return reqres.Response.GetVerifyVoteExtension(), nil
}
func (cli *socketClient) FinalizeBlock(
ctx context.Context,
req types.RequestFinalizeBlock) (*types.ResponseFinalizeBlock, error) {
reqres, err := cli.queueRequestAndFlush(ctx, types.ToRequestFinalizeBlock(req))
if err != nil {
return nil, err
}
return reqres.Response.GetFinalizeBlock(), nil
}
//----------------------------------------
// queueRequest enqueues req onto the queue. If the queue is full, it ether
@@ -550,8 +521,6 @@ func resMatchesReq(req *types.Request, res *types.Response) (ok bool) {
_, ok = res.Value.(*types.Response_Flush)
case *types.Request_Info:
_, ok = res.Value.(*types.Response_Info)
case *types.Request_DeliverTx:
_, ok = res.Value.(*types.Response_DeliverTx)
case *types.Request_CheckTx:
_, ok = res.Value.(*types.Response_CheckTx)
case *types.Request_Commit:
@@ -566,10 +535,6 @@ func resMatchesReq(req *types.Request, res *types.Response) (ok bool) {
_, ok = res.Value.(*types.Response_ExtendVote)
case *types.Request_VerifyVoteExtension:
_, ok = res.Value.(*types.Response_VerifyVoteExtension)
case *types.Request_BeginBlock:
_, ok = res.Value.(*types.Response_BeginBlock)
case *types.Request_EndBlock:
_, ok = res.Value.(*types.Response_EndBlock)
case *types.Request_ApplySnapshotChunk:
_, ok = res.Value.(*types.Response_ApplySnapshotChunk)
case *types.Request_LoadSnapshotChunk:
@@ -578,6 +543,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_FinalizeBlock:
_, ok = res.Value.(*types.Response_FinalizeBlock)
}
return ok
}
+3 -3
View File
@@ -29,7 +29,7 @@ func TestProperSyncCalls(t *testing.T) {
resp := make(chan error, 1)
go func() {
rsp, err := c.BeginBlock(ctx, types.RequestBeginBlock{})
rsp, err := c.FinalizeBlock(ctx, types.RequestFinalizeBlock{})
assert.NoError(t, err)
assert.NoError(t, c.Flush(ctx))
assert.NotNil(t, rsp)
@@ -79,7 +79,7 @@ type slowApp struct {
types.BaseApplication
}
func (slowApp) BeginBlock(req types.RequestBeginBlock) types.ResponseBeginBlock {
func (slowApp) FinalizeBlock(req types.RequestFinalizeBlock) types.ResponseFinalizeBlock {
time.Sleep(200 * time.Millisecond)
return types.ResponseBeginBlock{}
return types.ResponseFinalizeBlock{}
}
+43 -20
View File
@@ -193,7 +193,7 @@ var deliverTxCmd = &cobra.Command{
Short: "deliver a new transaction to the application",
Long: "deliver a new transaction to the application",
Args: cobra.ExactArgs(1),
RunE: cmdDeliverTx,
RunE: cmdFinalizeBlock,
}
var checkTxCmd = &cobra.Command{
@@ -300,17 +300,38 @@ func cmdTest(cmd *cobra.Command, args []string) error {
[]func() error{
func() error { return servertest.InitChain(ctx, client) },
func() error { return servertest.Commit(ctx, client, nil) },
func() error { return servertest.DeliverTx(ctx, client, []byte("abc"), code.CodeTypeBadNonce, nil) },
func() error { return servertest.Commit(ctx, client, nil) },
func() error { return servertest.DeliverTx(ctx, client, []byte{0x00}, code.CodeTypeOK, nil) },
func() error { return servertest.Commit(ctx, client, []byte{0, 0, 0, 0, 0, 0, 0, 1}) },
func() error { return servertest.DeliverTx(ctx, client, []byte{0x00}, code.CodeTypeBadNonce, nil) },
func() error { return servertest.DeliverTx(ctx, client, []byte{0x01}, code.CodeTypeOK, nil) },
func() error { return servertest.DeliverTx(ctx, client, []byte{0x00, 0x02}, code.CodeTypeOK, nil) },
func() error { return servertest.DeliverTx(ctx, client, []byte{0x00, 0x03}, code.CodeTypeOK, nil) },
func() error { return servertest.DeliverTx(ctx, client, []byte{0x00, 0x00, 0x04}, code.CodeTypeOK, nil) },
func() error {
return servertest.DeliverTx(ctx, client, []byte{0x00, 0x00, 0x06}, code.CodeTypeBadNonce, nil)
return servertest.FinalizeBlock(ctx, client, [][]byte{
[]byte("abc"),
}, []uint32{
code.CodeTypeBadNonce,
}, nil)
},
func() error { return servertest.Commit(ctx, client, nil) },
func() error {
return servertest.FinalizeBlock(ctx, client, [][]byte{
{0x00},
}, []uint32{
code.CodeTypeOK,
}, nil)
},
func() error { return servertest.Commit(ctx, client, []byte{0, 0, 0, 0, 0, 0, 0, 1}) },
func() error {
return servertest.FinalizeBlock(ctx, client, [][]byte{
{0x00},
{0x01},
{0x00, 0x02},
{0x00, 0x03},
{0x00, 0x00, 0x04},
{0x00, 0x00, 0x06},
}, []uint32{
code.CodeTypeBadNonce,
code.CodeTypeOK,
code.CodeTypeOK,
code.CodeTypeOK,
code.CodeTypeOK,
code.CodeTypeBadNonce,
}, nil)
},
func() error { return servertest.Commit(ctx, client, []byte{0, 0, 0, 0, 0, 0, 0, 5}) },
})
@@ -406,7 +427,7 @@ func muxOnCommands(cmd *cobra.Command, pArgs []string) error {
case "commit":
return cmdCommit(cmd, actualArgs)
case "deliver_tx":
return cmdDeliverTx(cmd, actualArgs)
return cmdFinalizeBlock(cmd, actualArgs)
case "echo":
return cmdEcho(cmd, actualArgs)
case "info":
@@ -475,7 +496,7 @@ func cmdInfo(cmd *cobra.Command, args []string) error {
const codeBad uint32 = 10
// Append a new tx to application
func cmdDeliverTx(cmd *cobra.Command, args []string) error {
func cmdFinalizeBlock(cmd *cobra.Command, args []string) error {
if len(args) == 0 {
printResponse(cmd, args, response{
Code: codeBad,
@@ -487,16 +508,18 @@ func cmdDeliverTx(cmd *cobra.Command, args []string) error {
if err != nil {
return err
}
res, err := client.DeliverTx(cmd.Context(), types.RequestDeliverTx{Tx: txBytes})
res, err := client.FinalizeBlock(cmd.Context(), types.RequestFinalizeBlock{Txs: [][]byte{txBytes}})
if err != nil {
return err
}
printResponse(cmd, args, response{
Code: res.Code,
Data: res.Data,
Info: res.Info,
Log: res.Log,
})
for _, tx := range res.Txs {
printResponse(cmd, args, response{
Code: tx.Code,
Data: tx.Data,
Info: tx.Info,
Log: tx.Log,
})
}
return nil
}
+27 -65
View File
@@ -6,7 +6,6 @@ import (
"math/rand"
"net"
"os"
"reflect"
"testing"
"time"
@@ -35,7 +34,7 @@ func TestKVStore(t *testing.T) {
logger := log.NewTestingLogger(t)
logger.Info("### Testing KVStore")
testStream(ctx, t, logger, kvstore.NewApplication())
testBulk(ctx, t, logger, kvstore.NewApplication())
}
func TestBaseApp(t *testing.T) {
@@ -44,7 +43,7 @@ func TestBaseApp(t *testing.T) {
logger := log.NewTestingLogger(t)
logger.Info("### Testing BaseApp")
testStream(ctx, t, logger, types.NewBaseApplication())
testBulk(ctx, t, logger, types.NewBaseApplication())
}
func TestGRPC(t *testing.T) {
@@ -57,10 +56,10 @@ func TestGRPC(t *testing.T) {
testGRPCSync(ctx, t, logger, types.NewGRPCApplication(types.NewBaseApplication()))
}
func testStream(ctx context.Context, t *testing.T, logger log.Logger, app types.Application) {
func testBulk(ctx context.Context, t *testing.T, logger log.Logger, app types.Application) {
t.Helper()
const numDeliverTxs = 20000
const numDeliverTxs = 700000
socketFile := fmt.Sprintf("test-%08x.sock", rand.Int31n(1<<30))
defer os.Remove(socketFile)
socket := fmt.Sprintf("unix://%v", socketFile)
@@ -77,51 +76,22 @@ func testStream(ctx context.Context, t *testing.T, logger log.Logger, app types.
err = client.Start(ctx)
require.NoError(t, err)
done := make(chan struct{})
counter := 0
client.SetResponseCallback(func(req *types.Request, res *types.Response) {
// Process response
switch r := res.Value.(type) {
case *types.Response_DeliverTx:
counter++
if r.DeliverTx.Code != code.CodeTypeOK {
t.Error("DeliverTx failed with ret_code", r.DeliverTx.Code)
}
if counter > numDeliverTxs {
t.Fatalf("Too many DeliverTx responses. Got %d, expected %d", counter, numDeliverTxs)
}
if counter == numDeliverTxs {
go func() {
time.Sleep(time.Second * 1) // Wait for a bit to allow counter overflow
close(done)
}()
return
}
case *types.Response_Flush:
// ignore
default:
t.Error("Unexpected response type", reflect.TypeOf(res.Value))
}
})
// Write requests
// Construct request
rfb := types.RequestFinalizeBlock{Txs: make([][]byte, numDeliverTxs)}
for counter := 0; counter < numDeliverTxs; counter++ {
// Send request
_, err = client.DeliverTxAsync(ctx, types.RequestDeliverTx{Tx: []byte("test")})
require.NoError(t, err)
// Sometimes send flush messages
if counter%128 == 0 {
err = client.Flush(ctx)
require.NoError(t, err)
}
rfb.Txs[counter] = []byte("test")
}
// Send bulk request
res, err := client.FinalizeBlock(ctx, rfb)
require.NoError(t, err)
require.Equal(t, numDeliverTxs, len(res.Txs), "Number of txs doesn't match")
for _, tx := range res.Txs {
require.Equal(t, tx.Code, code.CodeTypeOK, "Tx failed")
}
// Send final flush message
_, err = client.FlushAsync(ctx)
err = client.Flush(ctx)
require.NoError(t, err)
<-done
}
//-------------------------
@@ -133,7 +103,7 @@ func dialerFunc(ctx context.Context, addr string) (net.Conn, error) {
func testGRPCSync(ctx context.Context, t *testing.T, logger log.Logger, app types.ABCIApplicationServer) {
t.Helper()
numDeliverTxs := 2000
numDeliverTxs := 680000
socketFile := fmt.Sprintf("/tmp/test-%08x.sock", rand.Int31n(1<<30))
defer os.Remove(socketFile)
socket := fmt.Sprintf("unix://%v", socketFile)
@@ -142,7 +112,7 @@ func testGRPCSync(ctx context.Context, t *testing.T, logger log.Logger, app type
server := abciserver.NewGRPCServer(logger.With("module", "abci-server"), socket, app)
require.NoError(t, server.Start(ctx))
t.Cleanup(func() { server.Wait() })
t.Cleanup(server.Wait)
// Connect to the socket
conn, err := grpc.Dial(socket,
@@ -159,25 +129,17 @@ func testGRPCSync(ctx context.Context, t *testing.T, logger log.Logger, app type
client := types.NewABCIApplicationClient(conn)
// Write requests
// Construct request
rfb := types.RequestFinalizeBlock{Txs: make([][]byte, numDeliverTxs)}
for counter := 0; counter < numDeliverTxs; counter++ {
// Send request
response, err := client.DeliverTx(ctx, &types.RequestDeliverTx{Tx: []byte("test")})
require.NoError(t, err, "Error in GRPC DeliverTx")
counter++
if response.Code != code.CodeTypeOK {
t.Error("DeliverTx failed with ret_code", response.Code)
}
if counter > numDeliverTxs {
t.Fatal("Too many DeliverTx responses")
}
t.Log("response", counter)
if counter == numDeliverTxs {
go func() {
time.Sleep(time.Second * 1) // Wait for a bit to allow counter overflow
}()
}
rfb.Txs[counter] = []byte("test")
}
// Send request
response, err := client.FinalizeBlock(ctx, &rfb)
require.NoError(t, err, "Error in GRPC FinalizeBlock")
require.Equal(t, numDeliverTxs, len(response.Txs), "Number of txs returned via GRPC doesn't match")
for _, tx := range response.Txs {
require.Equal(t, tx.Code, code.CodeTypeOK, "Tx failed")
}
}
+12 -5
View File
@@ -86,14 +86,13 @@ func (app *Application) Info(req types.RequestInfo) (resInfo types.ResponseInfo)
}
// tx is either "key=value" or just arbitrary bytes
func (app *Application) DeliverTx(req types.RequestDeliverTx) types.ResponseDeliverTx {
func (app *Application) HandleTx(tx []byte) *types.ResponseDeliverTx {
var key, value string
parts := bytes.Split(req.Tx, []byte("="))
parts := bytes.Split(tx, []byte("="))
if len(parts) == 2 {
key, value = string(parts[0]), string(parts[1])
} else {
key, value = string(req.Tx), string(req.Tx)
key, value = string(tx), string(tx)
}
err := app.state.db.Set(prefixKey([]byte(key)), []byte(value))
@@ -114,7 +113,15 @@ func (app *Application) DeliverTx(req types.RequestDeliverTx) types.ResponseDeli
},
}
return types.ResponseDeliverTx{Code: code.CodeTypeOK, Events: events}
return &types.ResponseDeliverTx{Code: code.CodeTypeOK, Events: events}
}
func (app *Application) FinalizeBlock(req types.RequestFinalizeBlock) types.ResponseFinalizeBlock {
txs := make([]*types.ResponseDeliverTx, len(req.Txs))
for i, tx := range req.Txs {
txs[i] = app.HandleTx(tx)
}
return types.ResponseFinalizeBlock{Txs: txs}
}
func (app *Application) CheckTx(req types.RequestCheckTx) types.ResponseCheckTx {
+26 -33
View File
@@ -3,7 +3,6 @@ package kvstore
import (
"context"
"fmt"
"os"
"sort"
"testing"
@@ -25,12 +24,14 @@ const (
)
func testKVStore(t *testing.T, app types.Application, tx []byte, key, value string) {
req := types.RequestDeliverTx{Tx: tx}
ar := app.DeliverTx(req)
require.False(t, ar.IsErr(), ar)
req := types.RequestFinalizeBlock{Txs: [][]byte{tx}}
ar := app.FinalizeBlock(req)
require.Equal(t, 1, len(ar.Txs))
require.False(t, ar.Txs[0].IsErr())
// repeating tx doesn't raise error
ar = app.DeliverTx(req)
require.False(t, ar.IsErr(), ar)
ar = app.FinalizeBlock(req)
require.Equal(t, 1, len(ar.Txs))
require.False(t, ar.Txs[0].IsErr())
// commit
app.Commit()
@@ -72,10 +73,7 @@ func TestKVStoreKV(t *testing.T) {
}
func TestPersistentKVStoreKV(t *testing.T) {
dir, err := os.MkdirTemp("/tmp", "abci-kvstore-test") // TODO
if err != nil {
t.Fatal(err)
}
dir := t.TempDir()
logger := log.NewTestingLogger(t)
kvstore := NewPersistentKVStoreApplication(logger, dir)
@@ -90,10 +88,7 @@ func TestPersistentKVStoreKV(t *testing.T) {
}
func TestPersistentKVStoreInfo(t *testing.T) {
dir, err := os.MkdirTemp("/tmp", "abci-kvstore-test") // TODO
if err != nil {
t.Fatal(err)
}
dir := t.TempDir()
logger := log.NewTestingLogger(t)
kvstore := NewPersistentKVStoreApplication(logger, dir)
@@ -111,8 +106,7 @@ func TestPersistentKVStoreInfo(t *testing.T) {
header := tmproto.Header{
Height: height,
}
kvstore.BeginBlock(types.RequestBeginBlock{Hash: hash, Header: header})
kvstore.EndBlock(types.RequestEndBlock{Height: header.Height})
kvstore.FinalizeBlock(types.RequestFinalizeBlock{Hash: hash, Header: header, Height: height})
kvstore.Commit()
resInfo = kvstore.Info(types.RequestInfo{})
@@ -124,10 +118,7 @@ func TestPersistentKVStoreInfo(t *testing.T) {
// add a validator, remove a validator, update a validator
func TestValUpdates(t *testing.T) {
dir, err := os.MkdirTemp("/tmp", "abci-kvstore-test") // TODO
if err != nil {
t.Fatal(err)
}
dir := t.TempDir()
logger := log.NewTestingLogger(t)
kvstore := NewPersistentKVStoreApplication(logger, dir)
@@ -204,16 +195,16 @@ func makeApplyBlock(
Height: height,
}
kvstore.BeginBlock(types.RequestBeginBlock{Hash: hash, Header: header})
for _, tx := range txs {
if r := kvstore.DeliverTx(types.RequestDeliverTx{Tx: tx}); r.IsErr() {
t.Fatal(r)
}
}
resEndBlock := kvstore.EndBlock(types.RequestEndBlock{Height: header.Height})
resFinalizeBlock := kvstore.FinalizeBlock(types.RequestFinalizeBlock{
Hash: hash,
Header: header,
Height: height,
Txs: txs,
})
kvstore.Commit()
valsEqual(t, diff, resEndBlock.ValidatorUpdates)
valsEqual(t, diff, resFinalizeBlock.ValidatorUpdates)
}
@@ -330,13 +321,15 @@ func runClientTests(ctx context.Context, t *testing.T, client abciclient.Client)
}
func testClient(ctx context.Context, t *testing.T, app abciclient.Client, tx []byte, key, value string) {
ar, err := app.DeliverTx(ctx, types.RequestDeliverTx{Tx: tx})
ar, err := app.FinalizeBlock(ctx, types.RequestFinalizeBlock{Txs: [][]byte{tx}})
require.NoError(t, err)
require.False(t, ar.IsErr(), ar)
// repeating tx doesn't raise error
ar, err = app.DeliverTx(ctx, types.RequestDeliverTx{Tx: tx})
require.Equal(t, 1, len(ar.Txs))
require.False(t, ar.Txs[0].IsErr())
// repeating FinalizeBlock doesn't raise error
ar, err = app.FinalizeBlock(ctx, types.RequestFinalizeBlock{Txs: [][]byte{tx}})
require.NoError(t, err)
require.False(t, ar.IsErr(), ar)
require.Equal(t, 1, len(ar.Txs))
require.False(t, ar.Txs[0].IsErr())
// commit
_, err = app.Commit(ctx)
require.NoError(t, err)
+24 -22
View File
@@ -64,21 +64,21 @@ func (app *PersistentKVStoreApplication) Info(req types.RequestInfo) types.Respo
}
// tx is either "val:pubkey!power" or "key=value" or just arbitrary bytes
func (app *PersistentKVStoreApplication) DeliverTx(req types.RequestDeliverTx) types.ResponseDeliverTx {
func (app *PersistentKVStoreApplication) HandleTx(tx []byte) *types.ResponseDeliverTx {
// if it starts with "val:", update the validator set
// format is "val:pubkey!power"
if isValidatorTx(req.Tx) {
if isValidatorTx(tx) {
// update validators in the merkle tree
// and in app.ValUpdates
return app.execValidatorTx(req.Tx)
return app.execValidatorTx(tx)
}
if isPrepareTx(req.Tx) {
return app.execPrepareTx(req.Tx)
if isPrepareTx(tx) {
return app.execPrepareTx(tx)
}
// otherwise, update the key-value store
return app.app.DeliverTx(req)
return app.app.HandleTx(tx)
}
func (app *PersistentKVStoreApplication) CheckTx(req types.RequestCheckTx) types.ResponseCheckTx {
@@ -121,7 +121,9 @@ func (app *PersistentKVStoreApplication) InitChain(req types.RequestInitChain) t
}
// Track the block hash and header information
func (app *PersistentKVStoreApplication) BeginBlock(req types.RequestBeginBlock) types.ResponseBeginBlock {
// Execute transactions
// Update the validator set
func (app *PersistentKVStoreApplication) FinalizeBlock(req types.RequestFinalizeBlock) types.ResponseFinalizeBlock {
// reset valset changes
app.ValUpdates = make([]types.ValidatorUpdate, 0)
@@ -143,12 +145,12 @@ func (app *PersistentKVStoreApplication) BeginBlock(req types.RequestBeginBlock)
}
}
return types.ResponseBeginBlock{}
}
respTxs := make([]*types.ResponseDeliverTx, len(req.Txs))
for i, tx := range req.Txs {
respTxs[i] = app.HandleTx(tx)
}
// Update the validator set
func (app *PersistentKVStoreApplication) EndBlock(req types.RequestEndBlock) types.ResponseEndBlock {
return types.ResponseEndBlock{ValidatorUpdates: app.ValUpdates}
return types.ResponseFinalizeBlock{Txs: respTxs, ValidatorUpdates: app.ValUpdates}
}
func (app *PersistentKVStoreApplication) ListSnapshots(
@@ -238,13 +240,13 @@ func isValidatorTx(tx []byte) bool {
// format is "val:pubkey!power"
// pubkey is a base64-encoded 32-byte ed25519 key
func (app *PersistentKVStoreApplication) execValidatorTx(tx []byte) types.ResponseDeliverTx {
func (app *PersistentKVStoreApplication) execValidatorTx(tx []byte) *types.ResponseDeliverTx {
tx = tx[len(ValidatorSetChangePrefix):]
// get the pubkey and power
pubKeyAndPower := strings.Split(string(tx), "!")
if len(pubKeyAndPower) != 2 {
return types.ResponseDeliverTx{
return &types.ResponseDeliverTx{
Code: code.CodeTypeEncodingError,
Log: fmt.Sprintf("Expected 'pubkey!power'. Got %v", pubKeyAndPower)}
}
@@ -253,7 +255,7 @@ func (app *PersistentKVStoreApplication) execValidatorTx(tx []byte) types.Respon
// decode the pubkey
pubkey, err := base64.StdEncoding.DecodeString(pubkeyS)
if err != nil {
return types.ResponseDeliverTx{
return &types.ResponseDeliverTx{
Code: code.CodeTypeEncodingError,
Log: fmt.Sprintf("Pubkey (%s) is invalid base64", pubkeyS)}
}
@@ -261,7 +263,7 @@ func (app *PersistentKVStoreApplication) execValidatorTx(tx []byte) types.Respon
// decode the power
power, err := strconv.ParseInt(powerS, 10, 64)
if err != nil {
return types.ResponseDeliverTx{
return &types.ResponseDeliverTx{
Code: code.CodeTypeEncodingError,
Log: fmt.Sprintf("Power (%s) is not an int", powerS)}
}
@@ -271,7 +273,7 @@ func (app *PersistentKVStoreApplication) execValidatorTx(tx []byte) types.Respon
}
// add, update, or remove a validator
func (app *PersistentKVStoreApplication) updateValidator(v types.ValidatorUpdate) types.ResponseDeliverTx {
func (app *PersistentKVStoreApplication) updateValidator(v types.ValidatorUpdate) *types.ResponseDeliverTx {
pubkey, err := encoding.PubKeyFromProto(v.PubKey)
if err != nil {
panic(fmt.Errorf("can't decode public key: %w", err))
@@ -286,7 +288,7 @@ func (app *PersistentKVStoreApplication) updateValidator(v types.ValidatorUpdate
}
if !hasKey {
pubStr := base64.StdEncoding.EncodeToString(pubkey.Bytes())
return types.ResponseDeliverTx{
return &types.ResponseDeliverTx{
Code: code.CodeTypeUnauthorized,
Log: fmt.Sprintf("Cannot remove non-existent validator %s", pubStr)}
}
@@ -298,7 +300,7 @@ func (app *PersistentKVStoreApplication) updateValidator(v types.ValidatorUpdate
// add or update validator
value := bytes.NewBuffer(make([]byte, 0))
if err := types.WriteMessage(&v, value); err != nil {
return types.ResponseDeliverTx{
return &types.ResponseDeliverTx{
Code: code.CodeTypeEncodingError,
Log: fmt.Sprintf("error encoding validator: %v", err)}
}
@@ -311,7 +313,7 @@ func (app *PersistentKVStoreApplication) updateValidator(v types.ValidatorUpdate
// we only update the changes array if we successfully updated the tree
app.ValUpdates = append(app.ValUpdates, v)
return types.ResponseDeliverTx{Code: code.CodeTypeOK}
return &types.ResponseDeliverTx{Code: code.CodeTypeOK}
}
// -----------------------------
@@ -324,9 +326,9 @@ func isPrepareTx(tx []byte) bool {
// execPrepareTx is noop. tx data is considered as placeholder
// and is substitute at the PrepareProposal.
func (app *PersistentKVStoreApplication) execPrepareTx(tx []byte) types.ResponseDeliverTx {
func (app *PersistentKVStoreApplication) execPrepareTx(tx []byte) *types.ResponseDeliverTx {
// noop
return types.ResponseDeliverTx{}
return &types.ResponseDeliverTx{}
}
// substPrepareTx subst all the preparetx in the blockdata
+3 -9
View File
@@ -213,9 +213,6 @@ func (s *SocketServer) handleRequest(req *types.Request, responses chan<- *types
case *types.Request_Info:
res := s.app.Info(*r.Info)
responses <- types.ToResponseInfo(res)
case *types.Request_DeliverTx:
res := s.app.DeliverTx(*r.DeliverTx)
responses <- types.ToResponseDeliverTx(res)
case *types.Request_CheckTx:
res := s.app.CheckTx(*r.CheckTx)
responses <- types.ToResponseCheckTx(res)
@@ -228,12 +225,6 @@ func (s *SocketServer) handleRequest(req *types.Request, responses chan<- *types
case *types.Request_InitChain:
res := s.app.InitChain(*r.InitChain)
responses <- types.ToResponseInitChain(res)
case *types.Request_BeginBlock:
res := s.app.BeginBlock(*r.BeginBlock)
responses <- types.ToResponseBeginBlock(res)
case *types.Request_EndBlock:
res := s.app.EndBlock(*r.EndBlock)
responses <- types.ToResponseEndBlock(res)
case *types.Request_ListSnapshots:
res := s.app.ListSnapshots(*r.ListSnapshots)
responses <- types.ToResponseListSnapshots(res)
@@ -258,6 +249,9 @@ func (s *SocketServer) handleRequest(req *types.Request, responses chan<- *types
case *types.Request_VerifyVoteExtension:
res := s.app.VerifyVoteExtension(*r.VerifyVoteExtension)
responses <- types.ToResponseVerifyVoteExtension(res)
case *types.Request_FinalizeBlock:
res := s.app.FinalizeBlock(*r.FinalizeBlock)
responses <- types.ToResponseFinalizeBlock(res)
default:
responses <- types.ToResponseException("Unknown request")
}
+17 -15
View File
@@ -49,22 +49,24 @@ func Commit(ctx context.Context, client abciclient.Client, hashExp []byte) error
return nil
}
func DeliverTx(ctx context.Context, client abciclient.Client, txBytes []byte, codeExp uint32, dataExp []byte) error {
res, _ := client.DeliverTx(ctx, types.RequestDeliverTx{Tx: txBytes})
code, data, log := res.Code, res.Data, res.Log
if code != codeExp {
fmt.Println("Failed test: DeliverTx")
fmt.Printf("DeliverTx response code was unexpected. Got %v expected %v. Log: %v\n",
code, codeExp, log)
return errors.New("deliverTx error")
func FinalizeBlock(ctx context.Context, client abciclient.Client, txBytes [][]byte, codeExp []uint32, dataExp []byte) error {
res, _ := client.FinalizeBlock(ctx, types.RequestFinalizeBlock{Txs: txBytes})
for i, tx := range res.Txs {
code, data, log := tx.Code, tx.Data, tx.Log
if code != codeExp[i] {
fmt.Println("Failed test: FinalizeBlock")
fmt.Printf("FinalizeBlock response code was unexpected. Got %v expected %v. Log: %v\n",
code, codeExp, log)
return errors.New("FinalizeBlock error")
}
if !bytes.Equal(data, dataExp) {
fmt.Println("Failed test: FinalizeBlock")
fmt.Printf("FinalizeBlock response data was unexpected. Got %X expected %X\n",
data, dataExp)
return errors.New("FinalizeBlock error")
}
}
if !bytes.Equal(data, dataExp) {
fmt.Println("Failed test: DeliverTx")
fmt.Printf("DeliverTx response data was unexpected. Got %X expected %X\n",
data, dataExp)
return errors.New("deliverTx error")
}
fmt.Println("Passed test: DeliverTx")
fmt.Println("Passed test: FinalizeBlock")
return nil
}
+18 -33
View File
@@ -20,18 +20,14 @@ type Application interface {
InitChain(RequestInitChain) ResponseInitChain // Initialize blockchain w validators/other info from TendermintCore
PrepareProposal(RequestPrepareProposal) ResponsePrepareProposal
ProcessProposal(RequestProcessProposal) ResponseProcessProposal
// Signals the beginning of a block
BeginBlock(RequestBeginBlock) ResponseBeginBlock
// Deliver a tx for full processing
DeliverTx(RequestDeliverTx) ResponseDeliverTx
// Signals the end of a block, returns changes to the validator set
EndBlock(RequestEndBlock) ResponseEndBlock
// Commit the state and return the application Merkle root hash
Commit() ResponseCommit
// Create application specific vote extension
ExtendVote(RequestExtendVote) ResponseExtendVote
// Verify application's vote extension data
VerifyVoteExtension(RequestVerifyVoteExtension) ResponseVerifyVoteExtension
// Deliver the decided block with its txs to the Application
FinalizeBlock(RequestFinalizeBlock) ResponseFinalizeBlock
// State Sync Connection
ListSnapshots(RequestListSnapshots) ResponseListSnapshots // List available snapshots
@@ -56,10 +52,6 @@ func (BaseApplication) Info(req RequestInfo) ResponseInfo {
return ResponseInfo{}
}
func (BaseApplication) DeliverTx(req RequestDeliverTx) ResponseDeliverTx {
return ResponseDeliverTx{Code: CodeTypeOK}
}
func (BaseApplication) CheckTx(req RequestCheckTx) ResponseCheckTx {
return ResponseCheckTx{Code: CodeTypeOK}
}
@@ -86,14 +78,6 @@ func (BaseApplication) InitChain(req RequestInitChain) ResponseInitChain {
return ResponseInitChain{}
}
func (BaseApplication) BeginBlock(req RequestBeginBlock) ResponseBeginBlock {
return ResponseBeginBlock{}
}
func (BaseApplication) EndBlock(req RequestEndBlock) ResponseEndBlock {
return ResponseEndBlock{}
}
func (BaseApplication) ListSnapshots(req RequestListSnapshots) ResponseListSnapshots {
return ResponseListSnapshots{}
}
@@ -118,6 +102,16 @@ func (BaseApplication) ProcessProposal(req RequestProcessProposal) ResponseProce
return ResponseProcessProposal{}
}
func (BaseApplication) FinalizeBlock(req RequestFinalizeBlock) ResponseFinalizeBlock {
txs := make([]*ResponseDeliverTx, len(req.Txs))
for i := range req.Txs {
txs[i] = &ResponseDeliverTx{Code: CodeTypeOK}
}
return ResponseFinalizeBlock{
Txs: txs,
}
}
//-------------------------------------------------------
// GRPCApplication is a GRPC wrapper for Application
@@ -142,11 +136,6 @@ func (app *GRPCApplication) Info(ctx context.Context, req *RequestInfo) (*Respon
return &res, nil
}
func (app *GRPCApplication) DeliverTx(ctx context.Context, req *RequestDeliverTx) (*ResponseDeliverTx, error) {
res := app.app.DeliverTx(*req)
return &res, nil
}
func (app *GRPCApplication) CheckTx(ctx context.Context, req *RequestCheckTx) (*ResponseCheckTx, error) {
res := app.app.CheckTx(*req)
return &res, nil
@@ -167,16 +156,6 @@ func (app *GRPCApplication) InitChain(ctx context.Context, req *RequestInitChain
return &res, nil
}
func (app *GRPCApplication) BeginBlock(ctx context.Context, req *RequestBeginBlock) (*ResponseBeginBlock, error) {
res := app.app.BeginBlock(*req)
return &res, nil
}
func (app *GRPCApplication) EndBlock(ctx context.Context, req *RequestEndBlock) (*ResponseEndBlock, error) {
res := app.app.EndBlock(*req)
return &res, nil
}
func (app *GRPCApplication) ListSnapshots(
ctx context.Context, req *RequestListSnapshots) (*ResponseListSnapshots, error) {
res := app.app.ListSnapshots(*req)
@@ -224,3 +203,9 @@ func (app *GRPCApplication) ProcessProposal(
res := app.app.ProcessProposal(*req)
return &res, nil
}
func (app *GRPCApplication) FinalizeBlock(
ctx context.Context, req *RequestFinalizeBlock) (*ResponseFinalizeBlock, error) {
res := app.app.FinalizeBlock(*req)
return &res, nil
}
+13 -35
View File
@@ -4,6 +4,7 @@ import (
"io"
"github.com/gogo/protobuf/proto"
"github.com/tendermint/tendermint/internal/libs/protoio"
)
@@ -44,12 +45,6 @@ func ToRequestInfo(req RequestInfo) *Request {
}
}
func ToRequestDeliverTx(req RequestDeliverTx) *Request {
return &Request{
Value: &Request_DeliverTx{&req},
}
}
func ToRequestCheckTx(req RequestCheckTx) *Request {
return &Request{
Value: &Request_CheckTx{&req},
@@ -74,18 +69,6 @@ func ToRequestInitChain(req RequestInitChain) *Request {
}
}
func ToRequestBeginBlock(req RequestBeginBlock) *Request {
return &Request{
Value: &Request_BeginBlock{&req},
}
}
func ToRequestEndBlock(req RequestEndBlock) *Request {
return &Request{
Value: &Request_EndBlock{&req},
}
}
func ToRequestListSnapshots(req RequestListSnapshots) *Request {
return &Request{
Value: &Request_ListSnapshots{&req},
@@ -134,6 +117,12 @@ func ToRequestProcessProposal(req RequestProcessProposal) *Request {
}
}
func ToRequestFinalizeBlock(req RequestFinalizeBlock) *Request {
return &Request{
Value: &Request_FinalizeBlock{&req},
}
}
//----------------------------------------
func ToResponseException(errStr string) *Response {
@@ -159,11 +148,6 @@ func ToResponseInfo(res ResponseInfo) *Response {
Value: &Response_Info{&res},
}
}
func ToResponseDeliverTx(res ResponseDeliverTx) *Response {
return &Response{
Value: &Response_DeliverTx{&res},
}
}
func ToResponseCheckTx(res ResponseCheckTx) *Response {
return &Response{
@@ -189,18 +173,6 @@ func ToResponseInitChain(res ResponseInitChain) *Response {
}
}
func ToResponseBeginBlock(res ResponseBeginBlock) *Response {
return &Response{
Value: &Response_BeginBlock{&res},
}
}
func ToResponseEndBlock(res ResponseEndBlock) *Response {
return &Response{
Value: &Response_EndBlock{&res},
}
}
func ToResponseListSnapshots(res ResponseListSnapshots) *Response {
return &Response{
Value: &Response_ListSnapshots{&res},
@@ -248,3 +220,9 @@ func ToResponseProcessProposal(res ResponseProcessProposal) *Response {
Value: &Response_ProcessProposal{&res},
}
}
func ToResponseFinalizeBlock(res ResponseFinalizeBlock) *Response {
return &Response{
Value: &Response_FinalizeBlock{&res},
}
}
+1248 -1995
View File
File diff suppressed because it is too large Load Diff
+46
View File
@@ -0,0 +1,46 @@
package commands
import (
"fmt"
"github.com/spf13/cobra"
)
// NewCompletionCmd returns a cobra.Command that generates bash and zsh
// completion scripts for the given root command. If hidden is true, the
// command will not show up in the root command's list of available commands.
func NewCompletionCmd(rootCmd *cobra.Command, hidden bool) *cobra.Command {
flagZsh := "zsh"
cmd := &cobra.Command{
Use: "completion",
Short: "Generate shell completion scripts",
Long: fmt.Sprintf(`Generate Bash and Zsh completion scripts and print them to STDOUT.
Once saved to file, a completion script can be loaded in the shell's
current session as shown:
$ . <(%s completion)
To configure your bash shell to load completions for each session add to
your $HOME/.bashrc or $HOME/.profile the following instruction:
. <(%s completion)
`, rootCmd.Use, rootCmd.Use),
RunE: func(cmd *cobra.Command, _ []string) error {
zsh, err := cmd.Flags().GetBool(flagZsh)
if err != nil {
return err
}
if zsh {
return rootCmd.GenZshCompletion(cmd.OutOrStdout())
}
return rootCmd.GenBashCompletion(cmd.OutOrStdout())
},
Hidden: hidden,
Args: cobra.NoArgs,
}
cmd.Flags().Bool(flagZsh, false, "Generate Zsh completion script")
return cmd
}
+1
View File
@@ -5,6 +5,7 @@ import (
"fmt"
"github.com/spf13/cobra"
cfg "github.com/tendermint/tendermint/config"
"github.com/tendermint/tendermint/libs/log"
"github.com/tendermint/tendermint/scripts/keymigrate"
+4 -5
View File
@@ -199,10 +199,9 @@ func eventReIndex(cmd *cobra.Command, args eventReIndexArgs) error {
}
e := types.EventDataNewBlockHeader{
Header: b.Header,
NumTxs: int64(len(b.Txs)),
ResultBeginBlock: *r.BeginBlock,
ResultEndBlock: *r.EndBlock,
Header: b.Header,
NumTxs: int64(len(b.Txs)),
ResultFinalizeBlock: *r.FinalizeBlock,
}
var batch *indexer.Batch
@@ -214,7 +213,7 @@ func eventReIndex(cmd *cobra.Command, args eventReIndexArgs) error {
Height: b.Height,
Index: uint32(i),
Tx: b.Data.Txs[i],
Result: *(r.DeliverTxs[i]),
Result: *(r.FinalizeBlock.Txs[i]),
}
_ = batch.Add(&tr)
@@ -9,6 +9,8 @@ import (
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
dbm "github.com/tendermint/tm-db"
abcitypes "github.com/tendermint/tendermint/abci/types"
"github.com/tendermint/tendermint/config"
"github.com/tendermint/tendermint/internal/state/indexer"
@@ -16,7 +18,6 @@ import (
"github.com/tendermint/tendermint/libs/log"
prototmstate "github.com/tendermint/tendermint/proto/tendermint/state"
"github.com/tendermint/tendermint/types"
dbm "github.com/tendermint/tm-db"
_ "github.com/lib/pq" // for the psql sink
)
@@ -110,7 +111,7 @@ func TestLoadEventSink(t *testing.T) {
}
func TestLoadBlockStore(t *testing.T) {
testCfg, err := config.ResetTestRoot(t.Name())
testCfg, err := config.ResetTestRoot(t.TempDir(), t.Name())
require.NoError(t, err)
testCfg.DBBackend = "goleveldb"
_, _, err = loadStateAndBlockStore(testCfg)
@@ -154,9 +155,9 @@ func TestReIndexEvent(t *testing.T) {
dtx := abcitypes.ResponseDeliverTx{}
abciResp := &prototmstate.ABCIResponses{
DeliverTxs: []*abcitypes.ResponseDeliverTx{&dtx},
EndBlock: &abcitypes.ResponseEndBlock{},
BeginBlock: &abcitypes.ResponseBeginBlock{},
FinalizeBlock: &abcitypes.ResponseFinalizeBlock{
Txs: []*abcitypes.ResponseDeliverTx{&dtx},
},
}
mockStateStore.
+1
View File
@@ -2,6 +2,7 @@ package commands
import (
"github.com/spf13/cobra"
"github.com/tendermint/tendermint/config"
"github.com/tendermint/tendermint/internal/consensus"
"github.com/tendermint/tendermint/libs/log"
+10 -6
View File
@@ -19,10 +19,12 @@ func TestRollbackIntegration(t *testing.T) {
dir := t.TempDir()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
cfg, err := rpctest.CreateConfig(t.Name())
cfg, err := rpctest.CreateConfig(t, t.Name())
require.NoError(t, err)
cfg.BaseConfig.DBBackend = "goleveldb"
app, err := e2e.NewApplication(e2e.DefaultConfig(dir))
require.NoError(t, err)
t.Run("First run", func(t *testing.T) {
ctx, cancel := context.WithCancel(ctx)
@@ -30,27 +32,29 @@ func TestRollbackIntegration(t *testing.T) {
require.NoError(t, err)
node, _, err := rpctest.StartTendermint(ctx, cfg, app, rpctest.SuppressStdout)
require.NoError(t, err)
require.True(t, node.IsRunning())
time.Sleep(3 * time.Second)
cancel()
node.Wait()
require.False(t, node.IsRunning())
})
t.Run("Rollback", func(t *testing.T) {
time.Sleep(time.Second)
require.NoError(t, app.Rollback())
height, _, err = commands.RollbackState(cfg)
require.NoError(t, err)
require.NoError(t, err, "%d", height)
})
t.Run("Restart", func(t *testing.T) {
require.True(t, height > 0, "%d", height)
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
node2, _, err2 := rpctest.StartTendermint(ctx, cfg, app, rpctest.SuppressStdout)
require.NoError(t, err2)
logger := log.NewTestingLogger(t)
logger := log.NewNopLogger()
client, err := local.New(logger, node2.(local.NodeService))
require.NoError(t, err)
+27 -6
View File
@@ -1,6 +1,7 @@
package commands
import (
"context"
"fmt"
"os"
"path/filepath"
@@ -17,6 +18,17 @@ import (
tmos "github.com/tendermint/tendermint/libs/os"
)
// writeConfigVals writes a toml file with the given values.
// It returns an error if writing was impossible.
func writeConfigVals(dir string, vals map[string]string) error {
data := ""
for k, v := range vals {
data += fmt.Sprintf("%s = \"%s\"\n", k, v)
}
cfile := filepath.Join(dir, "config.toml")
return os.WriteFile(cfile, []byte(data), 0600)
}
// clearConfig clears env vars, the given root dir, and resets viper.
func clearConfig(t *testing.T, dir string) *cfg.Config {
t.Helper()
@@ -41,7 +53,7 @@ func testRootCmd(conf *cfg.Config) *cobra.Command {
return cmd
}
func testSetup(t *testing.T, conf *cfg.Config, args []string, env map[string]string) error {
func testSetup(ctx context.Context, t *testing.T, conf *cfg.Config, args []string, env map[string]string) error {
t.Helper()
cmd := testRootCmd(conf)
@@ -49,7 +61,7 @@ func testSetup(t *testing.T, conf *cfg.Config, args []string, env map[string]str
// run with the args and env
args = append([]string{cmd.Use}, args...)
return cli.RunWithArgs(cmd, args, env)
return cli.RunWithArgs(ctx, cmd, args, env)
}
func TestRootHome(t *testing.T) {
@@ -65,11 +77,14 @@ func TestRootHome(t *testing.T) {
{nil, map[string]string{"TMHOME": newRoot}, newRoot},
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
for i, tc := range cases {
t.Run(fmt.Sprint(i), func(t *testing.T) {
conf := clearConfig(t, tc.root)
err := testSetup(t, conf, tc.args, tc.env)
err := testSetup(ctx, t, conf, tc.args, tc.env)
require.NoError(t, err)
require.Equal(t, tc.root, conf.RootDir)
@@ -99,11 +114,14 @@ func TestRootFlagsEnv(t *testing.T) {
{nil, map[string]string{"TM_LOG_LEVEL": "debug"}, "debug"}, // right env
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
for i, tc := range cases {
t.Run(fmt.Sprint(i), func(t *testing.T) {
conf := clearConfig(t, defaultDir)
err := testSetup(t, conf, tc.args, tc.env)
err := testSetup(ctx, t, conf, tc.args, tc.env)
require.NoError(t, err)
assert.Equal(t, tc.logLevel, conf.LogLevel)
@@ -113,6 +131,9 @@ func TestRootFlagsEnv(t *testing.T) {
}
func TestRootConfig(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// write non-default config
nonDefaultLogLvl := "debug"
cvals := map[string]string{
@@ -142,14 +163,14 @@ func TestRootConfig(t *testing.T) {
// write the non-defaults to a different path
// TODO: support writing sub configs so we can test that too
err = WriteConfigVals(configFilePath, cvals)
err = writeConfigVals(configFilePath, cvals)
require.NoError(t, err)
cmd := testRootCmd(conf)
// run with the args and env
tc.args = append([]string{cmd.Use}, tc.args...)
err = cli.RunWithArgs(cmd, tc.args, tc.env)
err = cli.RunWithArgs(ctx, cmd, tc.args, tc.env)
require.NoError(t, err)
require.Equal(t, tc.logLvl, conf.LogLevel)
+1 -1
View File
@@ -117,7 +117,7 @@ func NewRunNodeCmd(nodeProvider cfg.ServiceProvider, conf *cfg.Config, logger lo
return fmt.Errorf("failed to start node: %w", err)
}
logger.Info("started node", "node", n.String())
logger.Info("started node", "chain", conf.ChainID())
<-ctx.Done()
return nil
+1
View File
@@ -4,6 +4,7 @@ import (
"fmt"
"github.com/spf13/cobra"
"github.com/tendermint/tendermint/config"
)
+2 -2
View File
@@ -44,7 +44,7 @@ func main() {
commands.MakeRollbackStateCommand(conf),
commands.MakeKeyMigrateCommand(conf, logger),
debug.DebugCmd,
cli.NewCompletionCmd(rcmd, true),
commands.NewCompletionCmd(rcmd, true),
)
// NOTE:
@@ -60,7 +60,7 @@ func main() {
// Create & start node
rcmd.AddCommand(commands.NewRunNodeCmd(nodeFunc, conf, logger))
if err := rcmd.ExecuteContext(ctx); err != nil {
if err := cli.RunWithTrace(ctx, rcmd); err != nil {
panic(err)
}
}
+4 -4
View File
@@ -504,13 +504,13 @@ namespace = "{{ .Instrumentation.Namespace }}"
/****** these are for test settings ***********/
func ResetTestRoot(testName string) (*Config, error) {
return ResetTestRootWithChainID(testName, "")
func ResetTestRoot(dir, testName string) (*Config, error) {
return ResetTestRootWithChainID(dir, testName, "")
}
func ResetTestRootWithChainID(testName string, chainID string) (*Config, error) {
func ResetTestRootWithChainID(dir, testName string, chainID string) (*Config, error) {
// create a unique, concurrency-safe test directory under os.TempDir()
rootDir, err := os.MkdirTemp("", fmt.Sprintf("%s-%s_", chainID, testName))
rootDir, err := os.MkdirTemp(dir, fmt.Sprintf("%s-%s_", chainID, testName))
if err != nil {
return nil, err
}
+2 -4
View File
@@ -20,9 +20,7 @@ func ensureFiles(t *testing.T, rootDir string, files ...string) {
func TestEnsureRoot(t *testing.T) {
// setup temp dir for test
tmpDir, err := os.MkdirTemp("", "config-test")
require.NoError(t, err)
defer os.RemoveAll(tmpDir)
tmpDir := t.TempDir()
// create root dir
EnsureRoot(tmpDir)
@@ -42,7 +40,7 @@ func TestEnsureTestRoot(t *testing.T) {
testName := "ensureTestRoot"
// create root dir
cfg, err := ResetTestRoot(testName)
cfg, err := ResetTestRoot(t.TempDir(), testName)
require.NoError(t, err)
defer os.RemoveAll(cfg.RootDir)
rootDir := cfg.RootDir
+1
View File
@@ -6,6 +6,7 @@ import (
"testing"
"github.com/stretchr/testify/require"
"github.com/tendermint/tendermint/crypto"
"github.com/tendermint/tendermint/crypto/internal/benchmarking"
)
+2 -1
View File
@@ -9,11 +9,12 @@ import (
"math/big"
secp256k1 "github.com/btcsuite/btcd/btcec"
"github.com/tendermint/tendermint/crypto"
"github.com/tendermint/tendermint/internal/jsontypes"
// necessary for Bitcoin address format
"golang.org/x/crypto/ripemd160" // nolint
"golang.org/x/crypto/ripemd160" //nolint:staticcheck
)
//-------------------------------------
+1
View File
@@ -6,6 +6,7 @@ import (
"testing"
"github.com/stretchr/testify/require"
"github.com/tendermint/tendermint/crypto"
"github.com/tendermint/tendermint/crypto/internal/benchmarking"
)
+4
View File
@@ -67,6 +67,10 @@ Note the context/background should be written in the present tense.
- [ADR-063: Privval-gRPC](./adr-063-privval-grpc.md)
- [ADR-066: E2E-Testing](./adr-066-e2e-testing.md)
- [ADR-072: Restore Requests for Comments](./adr-072-request-for-comments.md)
- [ADR-077: Block Retention](./adr-077-block-retention.md)
- [ADR-078: Non-zero Genesis](./adr-078-nonzero-genesis.md)
- [ADR-079: ED25519 Verification](./adr-079-ed25519-verification.md)
- [ADR-080: Reverse Sync](./adr-080-reverse-sync.md)
### Accepted
+4 -7
View File
@@ -41,18 +41,15 @@ sections.
- [RFC-001: Storage Engines](./rfc-001-storage-engine.rst)
- [RFC-002: Interprocess Communication](./rfc-002-ipc-ecosystem.md)
- [RFC-003: Performance Taxonomy](./rfc-003-performance-questions.md)
- [RFC-004: E2E Test Framework Enhancements](./rfc-004-e2e-framework.md)
- [RFC-004: E2E Test Framework Enhancements](./rfc-004-e2e-framework.rst)
- [RFC-005: Event System](./rfc-005-event-system.rst)
- [RFC-006: Event Subscription](./rfc-006-event-subscription.md)
- [RFC-007: Deterministic Proto Byte Serialization](./rfc-007-deterministic-proto-bytes.md)
- [RFC-008: Don't Panic](./rfc-008-don't-panic.md)
- [RFC-009: Consensus Parameter Upgrades](./rfc-009-consensus-parameter-upgrades.md)
- [RFC-010: P2P Light Client](./rfc-010-p2p-light-client.rst)
- [RFC-011: Block Retention](./rfc-011-block-retention.md)
- [RFC-012: Non-Zero Genesis](./rfc-012-nonzero-genesis.md)
- [RFC-013: ED25519 Verification](./rfc-013-ed25519-verification.md)
- [RFC-014: ABCI++](./rfc-014-abci++.md)
- [RFC-015: Reverse Sync](./rfc-015-reverse-sync.md)
- [RFC-016: Semantic Versioning](./rfc-016-semantic-versioning.md)
- [RFC-011: Delete Gas](./rfc-011-delete-gas.md)
- [RFC-012: ABCI++](./rfc-012-abci++.md)
- [RFC-013: Semantic Versioning](./rfc-013-semantic-versioning.md)
<!-- - [RFC-NNN: Title](./rfc-NNN-title.md) -->
+162
View File
@@ -0,0 +1,162 @@
# RFC 011: Remove Gas From Tendermint
## Changelog
- 03-Feb-2022: Initial draft (@williambanfield).
- 10-Feb-2022: Update in response to feedback (@williambanfield).
- 11-Feb-2022: Add reflection on MaxGas during consensus (@williambanfield).
## Abstract
In the v0.25.0 release, Tendermint added a mechanism for tracking 'Gas' in the mempool.
At a high level, Gas allows applications to specify how much it will cost the network,
often in compute resources, to execute a given transaction. While such a mechanism is common
in blockchain applications, it is not generalizable enough to be a maintained as a part
of Tendermint. This RFC explores the possibility of removing the concept of Gas from
Tendermint while still allowing applications the power to control the contents of
blocks to achieve similar goals.
## Background
The notion of Gas was included in the original Ethereum whitepaper and exists as
an important feature of the Ethereum blockchain.
The [whitepaper describes Gas][eth-whitepaper-messages] as an Anti-DoS mechanism. The Ethereum Virtual Machine
provides a Turing complete execution platform. Without any limitations, malicious
actors could waste computation resources by directing the EVM to perform large
or even infinite computations. Gas serves as a metering mechanism to prevent this.
Gas appears to have been added to Tendermint multiple times, initially as part of
a now defunct `/vm` package, and in its most recent iteration [as part of v0.25.0][gas-add-pr]
as a mechanism to limit the transactions that will be included in the block by an additional
parameter.
Gas has gained adoption within the Cosmos ecosystem [as part of the Cosmos SDK][cosmos-sdk-gas].
The SDK provides facilities for tracking how much 'Gas' a transaction is expected to take
and a mechanism for tracking how much gas a transaction has already taken.
Non-SDK applications also make use of the concept of Gas. Anoma appears to implement
[a gas system][anoma-gas] to meter the transactions it executes.
While the notion of gas is present in projects that make use of Tendermint, it is
not a concern of Tendermint's. Tendermint's value and goal is producing blocks
via a distributed consensus algorithm. Tendermint relies on the application specific
code to decide how to handle the transactions Tendermint has produced (or if the
application wants to consider them at all). Gas is an application concern.
Our implementation of Gas is not currently enforced by consensus. Our current validation check that
occurs during block propagation does not verify that the block is under the configured `MaxGas`.
Ensuring that the transactions in a proposed block do not exceed `MaxGas` would require
input from the application during propagation. The `ProcessProposal` method introduced
as part of ABCI++ would enable such input but would further entwine Tendermint and
the application. The issue of checking `MaxGas` during block propagation is important
because it demonstrates that the feature as it currently exists is not implemented
as fully as it perhaps should be.
Our implementation of Gas is causing issues for node operators and relayers. At
the moment, transactions that overflow the configured 'MaxGas' can be silently rejected
from the mempool. Overflowing MaxGas is the _only_ way that a transaction can be considered
invalid that is not directly a result of failing the `CheckTx`. Operators, and the application,
do not know that a transaction was removed from the mempool for this reason. A stateless check
of this nature is exactly what `CheckTx` exists for and there is no reason for the mempool
to keep track of this data separately. A special [MempoolError][add-mempool-error] field
was added in v0.35 to communicate to clients that a transaction failed after `CheckTx`.
While this should alleviate the pain for operators wishing to understand if their
transaction was included in the mempool, it highlights that the abstraction of
what is included in the mempool is not currently well defined.
Removing Gas from Tendermint and the mempool would allow for the mempool to be a better
abstraction: any transaction that arrived at `CheckTx` and passed the check will either be
a candidate for a later block or evicted after a TTL is reached or to make room for
other, higher priority transactions. All other transactions are completely invalid and can be discarded forever.
Removing gas will not be completely straightforward. It will mean ensuring that
equivalent functionality can be implemented outside of the mempool using the mempool's API.
## Discussion
This section catalogs the functionality that will need to exist within the Tendermint
mempool to allow Gas to be removed and replaced by application-side bookkeeping.
### Requirement: Provide Mempool Tx Sorting Mechanism
Gas produces a market for inclusion in a block. On many networks, a [gas fee][cosmos-sdk-fees] is
included in pending transactions. This fee indicates how much a user is willing to
pay per unit of execution and the fees are distributed to validators.
Validators wishing to extract higher gas fees are incentivized to include transactions
with the highest listed gas fees into each block. This produces a natural ordering
of the pending transactions. Applications wishing to implement a gas mechanism need
to be able to order the transactions in the mempool. This can trivially be accomplished
by sorting transactions using the `priority` field available to applications as part of
v0.35's `ResponseCheckTx` message.
### Requirement: Allow Application-Defined Block Resizing
When creating a block proposal, Tendermint pulls a set of possible transactions out of
the mempool to include in the next block. Tendermint uses MaxGas to limit the set of transactions
it pulls out of the mempool fetching a set of transactions whose sum is less than MaxGas.
By removing gas tracking from Tendermint's mempool, Tendermint will need to provide a way for
applications to determine an acceptable set of transactions to include in the block.
This is what the new ABCI++ `PrepareProposal` method is useful for. Applications
that wish to limit the contents of a block by an application-defined limit may
do so by removing transactions from the proposal it is passed during `PrepareProposal`.
Applications wishing to reach parity with the current Gas implementation may do
so by creating an application-side limit: filtering out transactions from
`PrepareProposal` the cause the proposal the exceed the maximum gas. Additionally,
applications can currently opt to have all transactions in the mempool delivered
during `PrepareProposal` by passing `-1` for `MaxGas` and `MaxBytes` into
[ReapMaxBytesMaxGas][reap-max-bytes-max-gas].
### Requirement: Handle Transaction Metadata
Moving the gas mechanism into applications adds an additional piece of complexity
to applications. The application must now track how much gas it expects a transaction
to consume. The mempool currently handles this bookkeeping responsibility and uses the estimated
gas to determine the set of transactions to include in the block. In order to task
the application with keeping track of this metadata, we should make it easier for the
application to do so. In general, we'll want to keep only one copy of this type
of metadata in the program at a time, either in the application or in Tendermint.
The following sections are possible solutions to the problem of storing transaction
metadata without duplication.
#### Metadata Handling: EvictTx Callback
A possible approach to handling transaction metadata is by adding a new `EvictTx`
ABCI method. Whenever the mempool is removing a transaction, either because it has
reached its TTL or because it failed `RecheckTx`, `EvictTx` would be called with
the transaction hash. This would indicate to the application that it could free any
metadata it was storing about the transaction such as the computed gas fee.
Eviction callbacks are pretty common in caching systems, so this would be very
well-worn territory.
#### Metadata Handling: Application-Specific Metadata Field(s)
An alternative approach to handling transaction metadata would be would be the
addition of a new application-metadata field in the `ResponseCheckTx`. This field
would be a protocol buffer message whose contents were entirely opaque to Tendermint.
The application would be responsible for marshalling and unmarshalling whatever data
it stored in this field. During `PrepareProposal`, the application would be passed
this metadata along with the transaction, allowing the application to use it to perform
any necessary filtering.
If either of these proposed metadata handling techniques are selected, it's likely
useful to enable applications to gossip metadata along with the transaction it is
gossiping. This could easily take the form of an opaque proto message that is
gossiped along with the transaction.
## References
[eth-whitepaper-messages]: https://ethereum.org/en/whitepaper/#messages-and-transactions
[gas-add-pr]: https://github.com/tendermint/tendermint/pull/2360
[cosmos-sdk-gas]: https://github.com/cosmos/cosmos-sdk/blob/c00cedb1427240a730d6eb2be6f7cb01f43869d3/docs/basics/gas-fees.md
[cosmos-sdk-fees]: https://github.com/cosmos/cosmos-sdk/blob/c00cedb1427240a730d6eb2be6f7cb01f43869d3/docs/basics/tx-lifecycle.md#gas-and-fees
[anoma-gas]: https://github.com/anoma/anoma/blob/6974fe1532a59db3574fc02e7f7e65d1216c1eb2/docs/src/specs/ledger.md#transaction-execution
[cosmos-sdk-fee]: https://github.com/cosmos/cosmos-sdk/blob/c00cedb1427240a730d6eb2be6f7cb01f43869d3/types/tx/tx.pb.go#L780-L794
[issue-7750]: https://github.com/tendermint/tendermint/issues/7750
[reap-max-bytes-max-gas]: https://github.com/tendermint/tendermint/blob/1ac58469f32a98f1c0e2905ca1773d9eac7b7103/internal/mempool/types.go#L45
[add-mempool-error]: https://github.com/tendermint/tendermint/blob/205bfca66f6da1b2dded381efb9ad3792f9404cf/rpc/coretypes/responses.go#L239
@@ -1,4 +1,4 @@
# RFC 011: ABCI++
# RFC 012: ABCI++
## Changelog
@@ -1,4 +1,4 @@
# RFC 012: Semantic Versioning
# RFC 013: Semantic Versioning
## Changelog
-1
View File
@@ -23,7 +23,6 @@ require (
github.com/oasisprotocol/curve25519-voi v0.0.0-20210609091139-0a56a4bca00b
github.com/ory/dockertest v3.3.5+incompatible
github.com/prometheus/client_golang v1.12.1
github.com/rcrowley/go-metrics v0.0.0-20200313005456-10cdbea86bc0
github.com/rs/cors v1.8.2
github.com/rs/zerolog v1.26.1
github.com/snikch/goodman v0.0.0-20171125024755-10e37e294daa
-2
View File
@@ -865,8 +865,6 @@ github.com/quasilyte/gogrep v0.0.0-20220103110004-ffaa07af02e3/go.mod h1:wSEyW6O
github.com/quasilyte/regex/syntax v0.0.0-20200407221936-30656e2c4a95 h1:L8QM9bvf68pVdQ3bCFZMDmnt9yqcMBro1pC7F+IPYMY=
github.com/quasilyte/regex/syntax v0.0.0-20200407221936-30656e2c4a95/go.mod h1:rlzQ04UMyJXu/aOvhd8qT+hvDrFpiwqp8MRXDY9szc0=
github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
github.com/rcrowley/go-metrics v0.0.0-20200313005456-10cdbea86bc0 h1:MkV+77GLUNo5oJ0jf870itWm3D0Sjh7+Za9gazKc5LQ=
github.com/rcrowley/go-metrics v0.0.0-20200313005456-10cdbea86bc0/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
+6 -8
View File
@@ -125,7 +125,6 @@ func TestBlockPoolBasic(t *testing.T) {
case err := <-errorsCh:
t.Error(err)
case request := <-requestsCh:
t.Logf("Pulled new BlockRequest %v", request)
if request.Height == 300 {
return // Done!
}
@@ -139,21 +138,19 @@ func TestBlockPoolTimeout(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
logger := log.TestingLogger()
start := int64(42)
peers := makePeers(10, start+1, 1000)
errorsCh := make(chan peerError, 1000)
requestsCh := make(chan BlockRequest, 1000)
pool := NewBlockPool(log.TestingLogger(), start, requestsCh, errorsCh)
pool := NewBlockPool(logger, start, requestsCh, errorsCh)
err := pool.Start(ctx)
if err != nil {
t.Error(err)
}
t.Cleanup(func() { cancel(); pool.Wait() })
for _, peer := range peers {
t.Logf("Peer %v", peer.id)
}
// Introduce each peer.
go func() {
for _, peer := range peers {
@@ -182,7 +179,6 @@ func TestBlockPoolTimeout(t *testing.T) {
for {
select {
case err := <-errorsCh:
t.Log(err)
// consider error to be always timeout here
if _, ok := timedOut[err.peerID]; !ok {
counter++
@@ -191,7 +187,9 @@ func TestBlockPoolTimeout(t *testing.T) {
}
}
case request := <-requestsCh:
t.Logf("Pulled new BlockRequest %+v", request)
logger.Debug("received request",
"counter", counter,
"request", request)
}
}
}
+4 -4
View File
@@ -203,7 +203,7 @@ func TestReactor_AbruptDisconnect(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
cfg, err := config.ResetTestRoot("block_sync_reactor_test")
cfg, err := config.ResetTestRoot(t.TempDir(), "block_sync_reactor_test")
require.NoError(t, err)
defer os.RemoveAll(cfg.RootDir)
@@ -243,7 +243,7 @@ func TestReactor_SyncTime(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
cfg, err := config.ResetTestRoot("block_sync_reactor_test")
cfg, err := config.ResetTestRoot(t.TempDir(), "block_sync_reactor_test")
require.NoError(t, err)
defer os.RemoveAll(cfg.RootDir)
@@ -271,7 +271,7 @@ func TestReactor_NoBlockResponse(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
cfg, err := config.ResetTestRoot("block_sync_reactor_test")
cfg, err := config.ResetTestRoot(t.TempDir(), "block_sync_reactor_test")
require.NoError(t, err)
defer os.RemoveAll(cfg.RootDir)
@@ -323,7 +323,7 @@ func TestReactor_BadBlockStopsPeer(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
cfg, err := config.ResetTestRoot("block_sync_reactor_test")
cfg, err := config.ResetTestRoot(t.TempDir(), "block_sync_reactor_test")
require.NoError(t, err)
defer os.RemoveAll(cfg.RootDir)
+1 -1
View File
@@ -60,7 +60,7 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
require.NoError(t, err)
require.NoError(t, stateStore.Save(state))
thisConfig, err := ResetConfig(fmt.Sprintf("%s_%d", testName, i))
thisConfig, err := ResetConfig(t.TempDir(), fmt.Sprintf("%s_%d", testName, i))
require.NoError(t, err)
defer os.RemoveAll(thisConfig.RootDir)
+13 -14
View File
@@ -50,23 +50,23 @@ type cleanupFunc func()
func configSetup(t *testing.T) *config.Config {
t.Helper()
cfg, err := ResetConfig("consensus_reactor_test")
cfg, err := ResetConfig(t.TempDir(), "consensus_reactor_test")
require.NoError(t, err)
t.Cleanup(func() { os.RemoveAll(cfg.RootDir) })
consensusReplayConfig, err := ResetConfig("consensus_replay_test")
consensusReplayConfig, err := ResetConfig(t.TempDir(), "consensus_replay_test")
require.NoError(t, err)
t.Cleanup(func() { os.RemoveAll(consensusReplayConfig.RootDir) })
configStateTest, err := ResetConfig("consensus_state_test")
configStateTest, err := ResetConfig(t.TempDir(), "consensus_state_test")
require.NoError(t, err)
t.Cleanup(func() { os.RemoveAll(configStateTest.RootDir) })
configMempoolTest, err := ResetConfig("consensus_mempool_test")
configMempoolTest, err := ResetConfig(t.TempDir(), "consensus_mempool_test")
require.NoError(t, err)
t.Cleanup(func() { os.RemoveAll(configMempoolTest.RootDir) })
configByzantineTest, err := ResetConfig("consensus_byzantine_test")
configByzantineTest, err := ResetConfig(t.TempDir(), "consensus_byzantine_test")
require.NoError(t, err)
t.Cleanup(func() { os.RemoveAll(configByzantineTest.RootDir) })
@@ -78,8 +78,8 @@ func ensureDir(t *testing.T, dir string, mode os.FileMode) {
require.NoError(t, tmos.EnsureDir(dir, mode))
}
func ResetConfig(name string) (*config.Config, error) {
return config.ResetTestRoot(name)
func ResetConfig(dir, name string) (*config.Config, error) {
return config.ResetTestRoot(dir, name)
}
//-------------------------------------------------------------------------------
@@ -422,7 +422,7 @@ func newState(
) *State {
t.Helper()
cfg, err := config.ResetTestRoot("consensus_state_test")
cfg, err := config.ResetTestRoot(t.TempDir(), "consensus_state_test")
require.NoError(t, err)
return newStateWithConfig(ctx, t, logger, cfg, state, pv, app)
@@ -769,7 +769,7 @@ func makeConsensusState(
blockStore := store.NewBlockStore(dbm.NewMemDB()) // each state needs its own db
state, err := sm.MakeGenesisState(genDoc)
require.NoError(t, err)
thisConfig, err := ResetConfig(fmt.Sprintf("%s_%d", testName, i))
thisConfig, err := ResetConfig(t.TempDir(), fmt.Sprintf("%s_%d", testName, i))
require.NoError(t, err)
configRootDirs = append(configRootDirs, thisConfig.RootDir)
@@ -827,7 +827,7 @@ func randConsensusNetWithPeers(
configRootDirs := make([]string, 0, nPeers)
for i := 0; i < nPeers; i++ {
state, _ := sm.MakeGenesisState(genDoc)
thisConfig, err := ResetConfig(fmt.Sprintf("%s_%d", testName, i))
thisConfig, err := ResetConfig(t.TempDir(), fmt.Sprintf("%s_%d", testName, i))
require.NoError(t, err)
configRootDirs = append(configRootDirs, thisConfig.RootDir)
@@ -839,10 +839,10 @@ func randConsensusNetWithPeers(
if i < nValidators {
privVal = privVals[i]
} else {
tempKeyFile, err := os.CreateTemp("", "priv_validator_key_")
tempKeyFile, err := os.CreateTemp(t.TempDir(), "priv_validator_key_")
require.NoError(t, err)
tempStateFile, err := os.CreateTemp("", "priv_validator_state_")
tempStateFile, err := os.CreateTemp(t.TempDir(), "priv_validator_state_")
require.NoError(t, err)
privVal, err = privval.GenFilePV(tempKeyFile.Name(), tempStateFile.Name(), "")
@@ -946,8 +946,7 @@ func (*mockTicker) SetLogger(log.Logger) {}
func newPersistentKVStore(t *testing.T, logger log.Logger) abci.Application {
t.Helper()
dir, err := os.MkdirTemp("", "persistent-kvstore")
require.NoError(t, err)
dir := t.TempDir()
return kvstore.NewPersistentKVStoreApplication(logger, dir)
}
+1
View File
@@ -7,6 +7,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/tendermint/tendermint/internal/eventbus"
"github.com/tendermint/tendermint/internal/p2p"
"github.com/tendermint/tendermint/libs/bytes"
+19 -13
View File
@@ -35,7 +35,7 @@ func TestMempoolNoProgressUntilTxsAvailable(t *testing.T) {
baseConfig := configSetup(t)
config, err := ResetConfig("consensus_mempool_txs_available_test")
config, err := ResetConfig(t.TempDir(), "consensus_mempool_txs_available_test")
require.NoError(t, err)
t.Cleanup(func() { _ = os.RemoveAll(config.RootDir) })
@@ -62,7 +62,7 @@ func TestMempoolProgressAfterCreateEmptyBlocksInterval(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
config, err := ResetConfig("consensus_mempool_txs_available_test")
config, err := ResetConfig(t.TempDir(), "consensus_mempool_txs_available_test")
require.NoError(t, err)
t.Cleanup(func() { _ = os.RemoveAll(config.RootDir) })
@@ -87,7 +87,7 @@ func TestMempoolProgressInHigherRound(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
config, err := ResetConfig("consensus_mempool_txs_available_test")
config, err := ResetConfig(t.TempDir(), "consensus_mempool_txs_available_test")
require.NoError(t, err)
t.Cleanup(func() { _ = os.RemoveAll(config.RootDir) })
@@ -192,8 +192,8 @@ func TestMempoolRmBadTx(t *testing.T) {
txBytes := make([]byte, 8)
binary.BigEndian.PutUint64(txBytes, uint64(0))
resDeliver := app.DeliverTx(abci.RequestDeliverTx{Tx: txBytes})
assert.False(t, resDeliver.IsErr(), fmt.Sprintf("expected no error. got %v", resDeliver))
resDeliver := app.FinalizeBlock(abci.RequestFinalizeBlock{Txs: [][]byte{txBytes}})
assert.False(t, resDeliver.Txs[0].IsErr(), fmt.Sprintf("expected no error. got %v", resDeliver))
resCommit := app.Commit()
assert.True(t, len(resCommit.Data) > 0)
@@ -264,15 +264,21 @@ func (app *CounterApplication) Info(req abci.RequestInfo) abci.ResponseInfo {
return abci.ResponseInfo{Data: fmt.Sprintf("txs:%v", app.txCount)}
}
func (app *CounterApplication) DeliverTx(req abci.RequestDeliverTx) abci.ResponseDeliverTx {
txValue := txAsUint64(req.Tx)
if txValue != uint64(app.txCount) {
return abci.ResponseDeliverTx{
Code: code.CodeTypeBadNonce,
Log: fmt.Sprintf("Invalid nonce. Expected %v, got %v", app.txCount, txValue)}
func (app *CounterApplication) FinalizeBlock(req abci.RequestFinalizeBlock) abci.ResponseFinalizeBlock {
respTxs := make([]*abci.ResponseDeliverTx, len(req.Txs))
for i, tx := range req.Txs {
txValue := txAsUint64(tx)
if txValue != uint64(app.txCount) {
respTxs[i] = &abci.ResponseDeliverTx{
Code: code.CodeTypeBadNonce,
Log: fmt.Sprintf("Invalid nonce. Expected %d, got %d", app.txCount, txValue),
}
continue
}
app.txCount++
respTxs[i] = &abci.ResponseDeliverTx{Code: code.CodeTypeOK}
}
app.txCount++
return abci.ResponseDeliverTx{Code: code.CodeTypeOK}
return abci.ResponseFinalizeBlock{Txs: respTxs}
}
func (app *CounterApplication) CheckTx(req abci.RequestCheckTx) abci.ResponseCheckTx {
+1
View File
@@ -3,6 +3,7 @@ package consensus
import (
"github.com/go-kit/kit/metrics"
"github.com/go-kit/kit/metrics/discard"
"github.com/tendermint/tendermint/types"
prometheus "github.com/go-kit/kit/metrics/prometheus"
@@ -4,6 +4,7 @@ package mocks
import (
mock "github.com/stretchr/testify/mock"
state "github.com/tendermint/tendermint/internal/state"
)
@@ -4,6 +4,7 @@ package mocks
import (
mock "github.com/stretchr/testify/mock"
state "github.com/tendermint/tendermint/internal/state"
time "time"
+1 -1
View File
@@ -391,7 +391,7 @@ func TestReactorWithEvidence(t *testing.T) {
stateStore := sm.NewStore(stateDB)
state, err := sm.MakeGenesisState(genDoc)
require.NoError(t, err)
thisConfig, err := ResetConfig(fmt.Sprintf("%s_%d", testName, i))
thisConfig, err := ResetConfig(t.TempDir(), fmt.Sprintf("%s_%d", testName, i))
require.NoError(t, err)
defer os.RemoveAll(thisConfig.RootDir)
+3 -8
View File
@@ -87,20 +87,15 @@ type mockProxyApp struct {
abciResponses *tmstate.ABCIResponses
}
func (mock *mockProxyApp) DeliverTx(req abci.RequestDeliverTx) abci.ResponseDeliverTx {
r := mock.abciResponses.DeliverTxs[mock.txCount]
func (mock *mockProxyApp) FinalizeBlock(req abci.RequestFinalizeBlock) abci.ResponseFinalizeBlock {
r := mock.abciResponses.FinalizeBlock
mock.txCount++
if r == nil {
return abci.ResponseDeliverTx{}
return abci.ResponseFinalizeBlock{}
}
return *r
}
func (mock *mockProxyApp) EndBlock(req abci.RequestEndBlock) abci.ResponseEndBlock {
mock.txCount = 0
return *mock.abciResponses.EndBlock
}
func (mock *mockProxyApp) Commit() abci.ResponseCommit {
return abci.ResponseCommit{Data: mock.appHash}
}
+24 -21
View File
@@ -142,7 +142,7 @@ func TestWALCrash(t *testing.T) {
for _, tc := range testCases {
tc := tc
t.Run(tc.name, func(t *testing.T) {
consensusReplayConfig, err := ResetConfig(tc.name)
consensusReplayConfig, err := ResetConfig(t.TempDir(), tc.name)
require.NoError(t, err)
crashWALandCheckLiveness(ctx, t, consensusReplayConfig, tc.initFn, tc.heightToStop)
})
@@ -665,12 +665,13 @@ func TestMockProxyApp(t *testing.T) {
logger := log.TestingLogger()
var validTxs, invalidTxs = 0, 0
txIndex := 0
txCount := 0
assert.NotPanics(t, func() {
abciResWithEmptyDeliverTx := new(tmstate.ABCIResponses)
abciResWithEmptyDeliverTx.DeliverTxs = make([]*abci.ResponseDeliverTx, 0)
abciResWithEmptyDeliverTx.DeliverTxs = append(abciResWithEmptyDeliverTx.DeliverTxs, &abci.ResponseDeliverTx{})
abciResWithEmptyDeliverTx.FinalizeBlock = new(abci.ResponseFinalizeBlock)
abciResWithEmptyDeliverTx.FinalizeBlock.Txs = make([]*abci.ResponseDeliverTx, 0)
abciResWithEmptyDeliverTx.FinalizeBlock.Txs = append(abciResWithEmptyDeliverTx.FinalizeBlock.Txs, &abci.ResponseDeliverTx{})
// called when saveABCIResponses:
bytes, err := proto.Marshal(abciResWithEmptyDeliverTx)
@@ -685,31 +686,33 @@ func TestMockProxyApp(t *testing.T) {
require.NoError(t, err)
abciRes := new(tmstate.ABCIResponses)
abciRes.DeliverTxs = make([]*abci.ResponseDeliverTx, len(loadedAbciRes.DeliverTxs))
abciRes.FinalizeBlock = new(abci.ResponseFinalizeBlock)
abciRes.FinalizeBlock.Txs = make([]*abci.ResponseDeliverTx, len(loadedAbciRes.FinalizeBlock.Txs))
someTx := []byte("tx")
resp, err := mock.DeliverTx(ctx, abci.RequestDeliverTx{Tx: someTx})
resp, err := mock.FinalizeBlock(ctx, abci.RequestFinalizeBlock{Txs: [][]byte{someTx}})
require.NoError(t, err)
// TODO: make use of res.Log
// TODO: make use of this info
// Blocks may include invalid txs.
if resp.Code == abci.CodeTypeOK {
validTxs++
} else {
invalidTxs++
for _, tx := range resp.Txs {
if tx.Code == abci.CodeTypeOK {
validTxs++
} else {
invalidTxs++
}
txCount++
}
abciRes.DeliverTxs[txIndex] = resp
txIndex++
assert.NoError(t, err)
})
assert.True(t, validTxs == 1)
assert.True(t, invalidTxs == 0)
require.Equal(t, 1, txCount)
require.Equal(t, 1, validTxs)
require.Zero(t, invalidTxs)
}
func tempWALWithData(t *testing.T, data []byte) string {
t.Helper()
walFile, err := os.CreateTemp("", "wal")
walFile, err := os.CreateTemp(t.TempDir(), "wal")
require.NoError(t, err, "failed to create temp WAL file")
_, err = walFile.Write(data)
@@ -743,7 +746,7 @@ func testHandshakeReplay(
logger := log.TestingLogger()
if testValidatorsChange {
testConfig, err := ResetConfig(fmt.Sprintf("%s_%v_m", t.Name(), mode))
testConfig, err := ResetConfig(t.TempDir(), fmt.Sprintf("%s_%v_m", t.Name(), mode))
require.NoError(t, err)
defer func() { _ = os.RemoveAll(testConfig.RootDir) }()
stateDB = dbm.NewMemDB()
@@ -754,7 +757,7 @@ func testHandshakeReplay(
commits = sim.Commits
store = newMockBlockStore(t, cfg, genesisState.ConsensusParams)
} else { // test single node
testConfig, err := ResetConfig(fmt.Sprintf("%s_%v_s", t.Name(), mode))
testConfig, err := ResetConfig(t.TempDir(), fmt.Sprintf("%s_%v_s", t.Name(), mode))
require.NoError(t, err)
defer func() { _ = os.RemoveAll(testConfig.RootDir) }()
walBody, err := WALWithNBlocks(ctx, t, logger, numBlocks)
@@ -1004,7 +1007,7 @@ func TestHandshakePanicsIfAppReturnsWrongAppHash(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
cfg, err := ResetConfig("handshake_test_")
cfg, err := ResetConfig(t.TempDir(), "handshake_test_")
require.NoError(t, err)
t.Cleanup(func() { os.RemoveAll(cfg.RootDir) })
privVal, err := privval.LoadFilePV(cfg.PrivValidator.KeyFile(), cfg.PrivValidator.StateFile())
@@ -1288,7 +1291,7 @@ func TestHandshakeUpdatesValidators(t *testing.T) {
app := &initChainApp{vals: types.TM2PB.ValidatorUpdates(vals)}
clientCreator := abciclient.NewLocalCreator(app)
cfg, err := ResetConfig("handshake_test_")
cfg, err := ResetConfig(t.TempDir(), "handshake_test_")
require.NoError(t, err)
t.Cleanup(func() { _ = os.RemoveAll(cfg.RootDir) })
@@ -2,11 +2,10 @@ package types
import (
"context"
"log"
"os"
"testing"
"github.com/stretchr/testify/require"
"github.com/tendermint/tendermint/config"
"github.com/tendermint/tendermint/crypto/tmhash"
"github.com/tendermint/tendermint/internal/test/factory"
@@ -16,40 +15,33 @@ import (
"github.com/tendermint/tendermint/types"
)
var cfg *config.Config // NOTE: must be reset for each _test.go file
func TestMain(m *testing.M) {
var err error
cfg, err = config.ResetTestRoot("consensus_height_vote_set_test")
if err != nil {
log.Fatal(err)
}
code := m.Run()
os.RemoveAll(cfg.RootDir)
os.Exit(code)
}
func TestPeerCatchupRounds(t *testing.T) {
cfg, err := config.ResetTestRoot(t.TempDir(), "consensus_height_vote_set_test")
if err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
valSet, privVals := factory.ValidatorSet(ctx, t, 10, 1)
hvs := NewHeightVoteSet(cfg.ChainID(), 1, valSet)
chainID := cfg.ChainID()
hvs := NewHeightVoteSet(chainID, 1, valSet)
vote999_0 := makeVoteHR(ctx, t, 1, 0, 999, privVals)
vote999_0 := makeVoteHR(ctx, t, 1, 0, 999, privVals, chainID)
added, err := hvs.AddVote(vote999_0, "peer1")
if !added || err != nil {
t.Error("Expected to successfully add vote from peer", added, err)
}
vote1000_0 := makeVoteHR(ctx, t, 1, 0, 1000, privVals)
vote1000_0 := makeVoteHR(ctx, t, 1, 0, 1000, privVals, chainID)
added, err = hvs.AddVote(vote1000_0, "peer1")
if !added || err != nil {
t.Error("Expected to successfully add vote from peer", added, err)
}
vote1001_0 := makeVoteHR(ctx, t, 1, 0, 1001, privVals)
vote1001_0 := makeVoteHR(ctx, t, 1, 0, 1001, privVals, chainID)
added, err = hvs.AddVote(vote1001_0, "peer1")
if err != ErrGotVoteFromUnwantedRound {
t.Errorf("expected GotVoteFromUnwantedRoundError, but got %v", err)
@@ -71,6 +63,7 @@ func makeVoteHR(
height int64,
valIndex, round int32,
privVals []types.PrivValidator,
chainID string,
) *types.Vote {
t.Helper()
@@ -89,7 +82,6 @@ func makeVoteHR(
Type: tmproto.PrecommitType,
BlockID: types.BlockID{Hash: randBytes, PartSetHeader: types.PartSetHeader{}},
}
chainID := cfg.ChainID()
v := vote.ToProto()
err = privVal.SignVote(ctx, chainID, v)
@@ -4,6 +4,7 @@ import (
"testing"
"github.com/stretchr/testify/require"
"github.com/tendermint/tendermint/libs/bits"
)
+1 -1
View File
@@ -145,7 +145,7 @@ func makeAddrs() (p2pAddr, rpcAddr string) {
// getConfig returns a config for test cases
func getConfig(t *testing.T) *config.Config {
c, err := config.ResetTestRoot(t.Name())
c, err := config.ResetTestRoot(t.TempDir(), t.Name())
require.NoError(t, err)
p2pAddr, rpcAddr := makeAddrs()
+2 -2
View File
@@ -89,7 +89,7 @@ func (b *EventBus) Publish(ctx context.Context, eventValue string, eventData typ
}
func (b *EventBus) PublishEventNewBlock(ctx context.Context, data types.EventDataNewBlock) error {
events := append(data.ResultBeginBlock.Events, data.ResultEndBlock.Events...)
events := data.ResultFinalizeBlock.Events
// add Tendermint-reserved new block event
events = append(events, types.EventNewBlock)
@@ -100,7 +100,7 @@ func (b *EventBus) PublishEventNewBlock(ctx context.Context, data types.EventDat
func (b *EventBus) PublishEventNewBlockHeader(ctx context.Context, data types.EventDataNewBlockHeader) error {
// no explicit deadline for publishing events
events := append(data.ResultBeginBlock.Events, data.ResultEndBlock.Events...)
events := data.ResultFinalizeBlock.Events
// add Tendermint-reserved new block header event
events = append(events, types.EventNewBlockHeader)
+17 -25
View File
@@ -83,14 +83,12 @@ func TestEventBusPublishEventNewBlock(t *testing.T) {
bps, err := block.MakePartSet(types.BlockPartSizeBytes)
require.NoError(t, err)
blockID := types.BlockID{Hash: block.Hash(), PartSetHeader: bps.Header()}
resultBeginBlock := abci.ResponseBeginBlock{
resultFinalizeBlock := abci.ResponseFinalizeBlock{
Events: []abci.Event{
{Type: "testType", Attributes: []abci.EventAttribute{{Key: "baz", Value: "1"}}},
},
}
resultEndBlock := abci.ResponseEndBlock{
Events: []abci.Event{
{Type: "testType", Attributes: []abci.EventAttribute{{Key: "foz", Value: "2"}}},
{Type: "testType", Attributes: []abci.EventAttribute{
{Key: "baz", Value: "1"},
{Key: "foz", Value: "2"},
}},
},
}
@@ -111,15 +109,13 @@ func TestEventBusPublishEventNewBlock(t *testing.T) {
edt := msg.Data().(types.EventDataNewBlock)
assert.Equal(t, block, edt.Block)
assert.Equal(t, blockID, edt.BlockID)
assert.Equal(t, resultBeginBlock, edt.ResultBeginBlock)
assert.Equal(t, resultEndBlock, edt.ResultEndBlock)
assert.Equal(t, resultFinalizeBlock, edt.ResultFinalizeBlock)
}()
err = eventBus.PublishEventNewBlock(ctx, types.EventDataNewBlock{
Block: block,
BlockID: blockID,
ResultBeginBlock: resultBeginBlock,
ResultEndBlock: resultEndBlock,
Block: block,
BlockID: blockID,
ResultFinalizeBlock: resultFinalizeBlock,
})
assert.NoError(t, err)
@@ -256,14 +252,12 @@ func TestEventBusPublishEventNewBlockHeader(t *testing.T) {
require.NoError(t, err)
block := types.MakeBlock(0, []types.Tx{}, nil, []types.Evidence{})
resultBeginBlock := abci.ResponseBeginBlock{
resultFinalizeBlock := abci.ResponseFinalizeBlock{
Events: []abci.Event{
{Type: "testType", Attributes: []abci.EventAttribute{{Key: "baz", Value: "1"}}},
},
}
resultEndBlock := abci.ResponseEndBlock{
Events: []abci.Event{
{Type: "testType", Attributes: []abci.EventAttribute{{Key: "foz", Value: "2"}}},
{Type: "testType", Attributes: []abci.EventAttribute{
{Key: "baz", Value: "1"},
{Key: "foz", Value: "2"},
}},
},
}
@@ -283,14 +277,12 @@ func TestEventBusPublishEventNewBlockHeader(t *testing.T) {
edt := msg.Data().(types.EventDataNewBlockHeader)
assert.Equal(t, block.Header, edt.Header)
assert.Equal(t, resultBeginBlock, edt.ResultBeginBlock)
assert.Equal(t, resultEndBlock, edt.ResultEndBlock)
assert.Equal(t, resultFinalizeBlock, edt.ResultFinalizeBlock)
}()
err = eventBus.PublishEventNewBlockHeader(ctx, types.EventDataNewBlockHeader{
Header: block.Header,
ResultBeginBlock: resultBeginBlock,
ResultEndBlock: resultEndBlock,
Header: block.Header,
ResultFinalizeBlock: resultFinalizeBlock,
})
assert.NoError(t, err)
+1
View File
@@ -4,6 +4,7 @@ package mocks
import (
mock "github.com/stretchr/testify/mock"
types "github.com/tendermint/tendermint/types"
)
+8 -7
View File
@@ -14,6 +14,7 @@ import (
"github.com/fortytw2/leaktest"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
abcitypes "github.com/tendermint/tendermint/abci/types"
"github.com/tendermint/tendermint/config"
"github.com/tendermint/tendermint/internal/inspect"
@@ -28,7 +29,7 @@ import (
)
func TestInspectConstructor(t *testing.T) {
cfg, err := config.ResetTestRoot("test")
cfg, err := config.ResetTestRoot(t.TempDir(), "test")
require.NoError(t, err)
testLogger := log.TestingLogger()
t.Cleanup(leaktest.Check(t))
@@ -43,7 +44,7 @@ func TestInspectConstructor(t *testing.T) {
}
func TestInspectRun(t *testing.T) {
cfg, err := config.ResetTestRoot("test")
cfg, err := config.ResetTestRoot(t.TempDir(), "test")
require.NoError(t, err)
testLogger := log.TestingLogger()
@@ -263,13 +264,13 @@ func TestBlockResults(t *testing.T) {
stateStoreMock := &statemocks.Store{}
// tmstate "github.com/tendermint/tendermint/proto/tendermint/state"
stateStoreMock.On("LoadABCIResponses", testHeight).Return(&state.ABCIResponses{
DeliverTxs: []*abcitypes.ResponseDeliverTx{
{
GasUsed: testGasUsed,
FinalizeBlock: &abcitypes.ResponseFinalizeBlock{
Txs: []*abcitypes.ResponseDeliverTx{
{
GasUsed: testGasUsed,
},
},
},
EndBlock: &abcitypes.ResponseEndBlock{},
BeginBlock: &abcitypes.ResponseBeginBlock{},
}, nil)
blockStoreMock := &statemocks.BlockStore{}
blockStoreMock.On("Base").Return(int64(0))
+3 -9
View File
@@ -25,11 +25,7 @@ func TestSIGHUP(t *testing.T) {
})
// First, create a temporary directory and move into it
dir, err := os.MkdirTemp("", "sighup_test")
require.NoError(t, err)
t.Cleanup(func() {
_ = os.RemoveAll(dir)
})
dir := t.TempDir()
require.NoError(t, os.Chdir(dir))
// Create an AutoFile in the temporary directory
@@ -48,9 +44,7 @@ func TestSIGHUP(t *testing.T) {
require.NoError(t, os.Rename(name, name+"_old"))
// Move into a different temporary directory
otherDir, err := os.MkdirTemp("", "sighup_test_other")
require.NoError(t, err)
t.Cleanup(func() { os.RemoveAll(otherDir) })
otherDir := t.TempDir()
require.NoError(t, os.Chdir(otherDir))
// Send SIGHUP to self.
@@ -112,7 +106,7 @@ func TestAutoFileSize(t *testing.T) {
defer cancel()
// First, create an AutoFile writing to a tempfile dir
f, err := os.CreateTemp("", "sighup_test")
f, err := os.CreateTemp(t.TempDir(), "sighup_test")
require.NoError(t, err)
require.NoError(t, f.Close())
+2 -5
View File
@@ -132,11 +132,8 @@ func TestRotateFile(t *testing.T) {
}
}()
dir, err := os.MkdirTemp("", "rotate_test")
require.NoError(t, err)
defer os.RemoveAll(dir)
err = os.Chdir(dir)
require.NoError(t, err)
dir := t.TempDir()
require.NoError(t, os.Chdir(dir))
require.True(t, filepath.IsAbs(g.Head.Path))
require.True(t, filepath.IsAbs(g.Dir))
-31
View File
@@ -1,31 +0,0 @@
package sync
import "sync"
// Closer implements a primitive to close a channel that signals process
// termination while allowing a caller to call Close multiple times safely. It
// should be used in cases where guarantees cannot be made about when and how
// many times closure is executed.
type Closer struct {
closeOnce sync.Once
doneCh chan struct{}
}
// NewCloser returns a reference to a new Closer.
func NewCloser() *Closer {
return &Closer{doneCh: make(chan struct{})}
}
// Done returns the internal done channel allowing the caller either block or wait
// for the Closer to be terminated/closed.
func (c *Closer) Done() <-chan struct{} {
return c.doneCh
}
// Close gracefully closes the Closer. A caller should only call Close once, but
// it is safe to call it successive times.
func (c *Closer) Close() {
c.closeOnce.Do(func() {
close(c.doneCh)
})
}
-28
View File
@@ -1,28 +0,0 @@
package sync_test
import (
"testing"
"time"
"github.com/stretchr/testify/require"
tmsync "github.com/tendermint/tendermint/internal/libs/sync"
)
func TestCloser(t *testing.T) {
closer := tmsync.NewCloser()
var timeout bool
select {
case <-closer.Done():
case <-time.After(time.Second):
timeout = true
}
for i := 0; i < 10; i++ {
closer.Close()
}
require.True(t, timeout)
<-closer.Done()
}
+1 -1
View File
@@ -21,7 +21,7 @@ func TestWriteFileAtomic(t *testing.T) {
perm os.FileMode = 0600
)
f, err := os.CreateTemp("/tmp", "write-atomic-test-")
f, err := os.CreateTemp(t.TempDir(), "write-atomic-test-")
if err != nil {
t.Fatal(err)
}
+1
View File
@@ -4,6 +4,7 @@ import (
"testing"
"github.com/stretchr/testify/require"
"github.com/tendermint/tendermint/types"
)
+9 -2
View File
@@ -14,8 +14,13 @@ func BenchmarkTxMempool_CheckTx(b *testing.B) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// setup the cache and the mempool number for hitting GetEvictableTxs during the
// benchmark. 5000 is the current default mempool size in the TM config.
txmp := setup(ctx, b, 10000)
txmp.config.Size = 5000
rng := rand.New(rand.NewSource(time.Now().UnixNano()))
const peerID = 1
b.ResetTimer()
@@ -26,9 +31,11 @@ func BenchmarkTxMempool_CheckTx(b *testing.B) {
require.NoError(b, err)
priority := int64(rng.Intn(9999-1000) + 1000)
tx := []byte(fmt.Sprintf("%X=%d", prefix, priority))
tx := []byte(fmt.Sprintf("sender-%d-%d=%X=%d", n, peerID, prefix, priority))
txInfo := TxInfo{SenderID: uint16(peerID)}
b.StartTimer()
require.NoError(b, txmp.CheckTx(ctx, tx, nil, TxInfo{}))
require.NoError(b, txmp.CheckTx(ctx, tx, nil, txInfo))
}
}
+1 -1
View File
@@ -82,7 +82,7 @@ func setup(ctx context.Context, t testing.TB, cacheSize int, options ...TxMempoo
cc := abciclient.NewLocalCreator(app)
logger := log.TestingLogger()
cfg, err := config.ResetTestRoot(strings.ReplaceAll(t.Name(), "/", "|"))
cfg, err := config.ResetTestRoot(t.TempDir(), strings.ReplaceAll(t.Name(), "/", "|"))
require.NoError(t, err)
cfg.Mempool.CacheSize = cacheSize
appConnMem, err := cc(logger)
+2 -1
View File
@@ -12,6 +12,7 @@ import (
"github.com/fortytw2/leaktest"
"github.com/stretchr/testify/require"
"github.com/tendermint/tendermint/abci/example/kvstore"
abci "github.com/tendermint/tendermint/abci/types"
"github.com/tendermint/tendermint/config"
@@ -41,7 +42,7 @@ type reactorTestSuite struct {
func setupReactors(ctx context.Context, t *testing.T, numNodes int, chBuf uint) *reactorTestSuite {
t.Helper()
cfg, err := config.ResetTestRoot(strings.ReplaceAll(t.Name(), "/", "|"))
cfg, err := config.ResetTestRoot(t.TempDir(), strings.ReplaceAll(t.Name(), "/", "|"))
require.NoError(t, err)
t.Cleanup(func() { os.RemoveAll(cfg.RootDir) })
+1
View File
@@ -8,6 +8,7 @@ import (
"time"
"github.com/stretchr/testify/require"
"github.com/tendermint/tendermint/types"
)
+1
View File
@@ -6,6 +6,7 @@ import (
"sync"
"github.com/gogo/protobuf/proto"
"github.com/tendermint/tendermint/types"
)
+1
View File
@@ -4,6 +4,7 @@ import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/tendermint/tendermint/proto/tendermint/p2p"
)
+1
View File
@@ -2,6 +2,7 @@ package p2ptest
import (
gogotypes "github.com/gogo/protobuf/types"
"github.com/tendermint/tendermint/types"
)
+20 -28
View File
@@ -5,10 +5,11 @@ import (
"context"
"sort"
"strconv"
"sync"
"time"
"github.com/gogo/protobuf/proto"
tmsync "github.com/tendermint/tendermint/internal/libs/sync"
"github.com/tendermint/tendermint/libs/log"
)
@@ -78,8 +79,10 @@ type pqScheduler struct {
enqueueCh chan Envelope
dequeueCh chan Envelope
closer *tmsync.Closer
done *tmsync.Closer
closeFn func()
closeCh <-chan struct{}
done chan struct{}
}
func newPQScheduler(
@@ -108,6 +111,9 @@ func newPQScheduler(
pq := make(priorityQueue, 0)
heap.Init(&pq)
closeCh := make(chan struct{})
once := &sync.Once{}
return &pqScheduler{
logger: logger.With("router", "scheduler"),
metrics: m,
@@ -118,32 +124,18 @@ func newPQScheduler(
sizes: sizes,
enqueueCh: make(chan Envelope, enqueueBuf),
dequeueCh: make(chan Envelope, dequeueBuf),
closer: tmsync.NewCloser(),
done: tmsync.NewCloser(),
closeFn: func() { once.Do(func() { close(closeCh) }) },
closeCh: closeCh,
done: make(chan struct{}),
}
}
func (s *pqScheduler) enqueue() chan<- Envelope {
return s.enqueueCh
}
func (s *pqScheduler) dequeue() <-chan Envelope {
return s.dequeueCh
}
func (s *pqScheduler) close() {
s.closer.Close()
<-s.done.Done()
}
func (s *pqScheduler) closed() <-chan struct{} {
return s.closer.Done()
}
// start starts non-blocking process that starts the priority queue scheduler.
func (s *pqScheduler) start(ctx context.Context) {
go s.process(ctx)
}
func (s *pqScheduler) start(ctx context.Context) { go s.process(ctx) }
func (s *pqScheduler) enqueue() chan<- Envelope { return s.enqueueCh }
func (s *pqScheduler) dequeue() <-chan Envelope { return s.dequeueCh }
func (s *pqScheduler) close() { s.closeFn() }
func (s *pqScheduler) closed() <-chan struct{} { return s.done }
// process starts a block process where we listen for Envelopes to enqueue. If
// there is sufficient capacity, it will be enqueued into the priority queue,
@@ -155,7 +147,7 @@ func (s *pqScheduler) start(ctx context.Context) {
// After we attempt to enqueue the incoming Envelope, if the priority queue is
// non-empty, we pop the top Envelope and send it on the dequeueCh.
func (s *pqScheduler) process(ctx context.Context) {
defer s.done.Close()
defer close(s.done)
for {
select {
@@ -264,13 +256,13 @@ func (s *pqScheduler) process(ctx context.Context) {
"peer_id", string(pqEnv.envelope.To)).Add(float64(-pqEnv.size))
select {
case s.dequeueCh <- pqEnv.envelope:
case <-s.closer.Done():
case <-s.closeCh:
return
}
}
case <-ctx.Done():
return
case <-s.closer.Done():
case <-s.closeCh:
return
}
}
+1
View File
@@ -6,6 +6,7 @@ import (
"time"
gogotypes "github.com/gogo/protobuf/types"
"github.com/tendermint/tendermint/libs/log"
)
+12 -18
View File
@@ -1,7 +1,7 @@
package p2p
import (
tmsync "github.com/tendermint/tendermint/internal/libs/sync"
"sync"
)
// default capacity for the size of a queue
@@ -32,28 +32,22 @@ type queue interface {
// in the order they were received, and blocks until message is received.
type fifoQueue struct {
queueCh chan Envelope
closer *tmsync.Closer
closeFn func()
closeCh <-chan struct{}
}
func newFIFOQueue(size int) queue {
closeCh := make(chan struct{})
once := &sync.Once{}
return &fifoQueue{
queueCh: make(chan Envelope, size),
closer: tmsync.NewCloser(),
closeFn: func() { once.Do(func() { close(closeCh) }) },
closeCh: closeCh,
}
}
func (q *fifoQueue) enqueue() chan<- Envelope {
return q.queueCh
}
func (q *fifoQueue) dequeue() <-chan Envelope {
return q.queueCh
}
func (q *fifoQueue) close() {
q.closer.Close()
}
func (q *fifoQueue) closed() <-chan struct{} {
return q.closer.Done()
}
func (q *fifoQueue) enqueue() chan<- Envelope { return q.queueCh }
func (q *fifoQueue) dequeue() <-chan Envelope { return q.queueCh }
func (q *fifoQueue) close() { q.closeFn() }
func (q *fifoQueue) closed() <-chan struct{} { return q.closeCh }
+2 -2
View File
@@ -8,7 +8,7 @@ import (
"time"
"github.com/stretchr/testify/require"
"github.com/tendermint/tendermint/internal/libs/sync"
"github.com/tendermint/tendermint/libs/log"
)
@@ -29,6 +29,6 @@ func TestConnectionFiltering(t *testing.T) {
},
}
require.Equal(t, 0, filterByIPCount)
router.openConnection(ctx, &MemoryConnection{logger: logger, closer: sync.NewCloser()})
router.openConnection(ctx, &MemoryConnection{logger: logger, closeFn: func() {}})
require.Equal(t, 1, filterByIPCount)
}
+1
View File
@@ -6,6 +6,7 @@ import (
"testing"
"github.com/stretchr/testify/require"
"github.com/tendermint/tendermint/libs/log"
"github.com/tendermint/tendermint/types"
)
+8 -8
View File
@@ -19,7 +19,6 @@ import (
dbm "github.com/tendermint/tm-db"
"github.com/tendermint/tendermint/crypto"
tmsync "github.com/tendermint/tendermint/internal/libs/sync"
"github.com/tendermint/tendermint/internal/p2p"
"github.com/tendermint/tendermint/internal/p2p/mocks"
"github.com/tendermint/tendermint/internal/p2p/p2ptest"
@@ -385,12 +384,12 @@ func TestRouter_AcceptPeers(t *testing.T) {
t.Cleanup(leaktest.Check(t))
// Set up a mock transport that handshakes.
closer := tmsync.NewCloser()
connCtx, connCancel := context.WithCancel(context.Background())
mockConnection := &mocks.Connection{}
mockConnection.On("String").Maybe().Return("mock")
mockConnection.On("Handshake", mock.Anything, selfInfo, selfKey).
Return(tc.peerInfo, tc.peerKey, nil)
mockConnection.On("Close").Run(func(_ mock.Arguments) { closer.Close() }).Return(nil).Maybe()
mockConnection.On("Close").Run(func(_ mock.Arguments) { connCancel() }).Return(nil).Maybe()
mockConnection.On("RemoteEndpoint").Return(p2p.Endpoint{})
if tc.ok {
mockConnection.On("ReceiveMessage", mock.Anything).Return(chID, nil, io.EOF).Maybe()
@@ -433,7 +432,7 @@ func TestRouter_AcceptPeers(t *testing.T) {
time.Sleep(time.Millisecond)
} else {
select {
case <-closer.Done():
case <-connCtx.Done():
case <-time.After(100 * time.Millisecond):
require.Fail(t, "connection not closed")
}
@@ -620,13 +619,14 @@ func TestRouter_DialPeers(t *testing.T) {
endpoint := p2p.Endpoint{Protocol: "mock", Path: string(tc.dialID)}
// Set up a mock transport that handshakes.
closer := tmsync.NewCloser()
connCtx, connCancel := context.WithCancel(context.Background())
defer connCancel()
mockConnection := &mocks.Connection{}
mockConnection.On("String").Maybe().Return("mock")
if tc.dialErr == nil {
mockConnection.On("Handshake", mock.Anything, selfInfo, selfKey).
Return(tc.peerInfo, tc.peerKey, nil)
mockConnection.On("Close").Run(func(_ mock.Arguments) { closer.Close() }).Return(nil).Maybe()
mockConnection.On("Close").Run(func(_ mock.Arguments) { connCancel() }).Return(nil).Maybe()
}
if tc.ok {
mockConnection.On("ReceiveMessage", mock.Anything).Return(chID, nil, io.EOF).Maybe()
@@ -644,7 +644,7 @@ func TestRouter_DialPeers(t *testing.T) {
mockTransport.On("Dial", mock.Anything, endpoint).Maybe().Return(nil, io.EOF)
} else {
mockTransport.On("Dial", mock.Anything, endpoint).Once().
Run(func(_ mock.Arguments) { closer.Close() }).
Run(func(_ mock.Arguments) { connCancel() }).
Return(nil, tc.dialErr)
}
@@ -681,7 +681,7 @@ func TestRouter_DialPeers(t *testing.T) {
time.Sleep(time.Millisecond)
} else {
select {
case <-closer.Done():
case <-connCtx.Done():
case <-time.After(100 * time.Millisecond):
require.Fail(t, "connection not closed")
}
+26 -10
View File
@@ -138,19 +138,35 @@ func (m *MConnTransport) Accept(ctx context.Context) (Connection, error) {
return nil, errors.New("transport is not listening")
}
tcpConn, err := m.listener.Accept()
if err != nil {
select {
case <-ctx.Done():
return nil, io.EOF
case <-m.doneCh:
return nil, io.EOF
default:
return nil, err
conCh := make(chan net.Conn)
errCh := make(chan error)
go func() {
tcpConn, err := m.listener.Accept()
if err != nil {
select {
case errCh <- err:
case <-ctx.Done():
}
}
select {
case conCh <- tcpConn:
case <-ctx.Done():
}
}()
select {
case <-ctx.Done():
m.listener.Close()
return nil, io.EOF
case <-m.doneCh:
m.listener.Close()
return nil, io.EOF
case err := <-errCh:
return nil, err
case tcpConn := <-conCh:
return newMConnConnection(m.logger, tcpConn, m.mConnConfig, m.channelDescs), nil
}
return newMConnConnection(m.logger, tcpConn, m.mConnConfig, m.channelDescs), nil
}
// Dial implements Transport.
-3
View File
@@ -154,9 +154,6 @@ func TestMConnTransport_Listen(t *testing.T) {
t.Run(tc.endpoint.String(), func(t *testing.T) {
t.Cleanup(leaktest.Check(t))
ctx, cancel = context.WithCancel(ctx)
defer cancel()
transport := p2p.NewMConnTransport(
log.TestingLogger(),
conn.DefaultMConnConfig(),
+22 -23
View File
@@ -9,7 +9,6 @@ import (
"sync"
"github.com/tendermint/tendermint/crypto"
tmsync "github.com/tendermint/tendermint/internal/libs/sync"
"github.com/tendermint/tendermint/libs/log"
"github.com/tendermint/tendermint/types"
)
@@ -175,10 +174,17 @@ func (t *MemoryTransport) Dial(ctx context.Context, endpoint Endpoint) (Connecti
inCh := make(chan memoryMessage, t.bufferSize)
outCh := make(chan memoryMessage, t.bufferSize)
closer := tmsync.NewCloser()
outConn := newMemoryConnection(t.logger, t.nodeID, peer.nodeID, inCh, outCh, closer)
inConn := newMemoryConnection(peer.logger, peer.nodeID, t.nodeID, outCh, inCh, closer)
once := &sync.Once{}
closeCh := make(chan struct{})
closeFn := func() { once.Do(func() { close(closeCh) }) }
outConn := newMemoryConnection(t.logger, t.nodeID, peer.nodeID, inCh, outCh)
outConn.closeCh = closeCh
outConn.closeFn = closeFn
inConn := newMemoryConnection(peer.logger, peer.nodeID, t.nodeID, outCh, inCh)
inConn.closeCh = closeCh
inConn.closeFn = closeFn
select {
case peer.acceptCh <- inConn:
@@ -202,7 +208,9 @@ type MemoryConnection struct {
receiveCh <-chan memoryMessage
sendCh chan<- memoryMessage
closer *tmsync.Closer
closeFn func()
closeCh <-chan struct{}
}
// memoryMessage is passed internally, containing either a message or handshake.
@@ -222,7 +230,6 @@ func newMemoryConnection(
remoteID types.NodeID,
receiveCh <-chan memoryMessage,
sendCh chan<- memoryMessage,
closer *tmsync.Closer,
) *MemoryConnection {
return &MemoryConnection{
logger: logger.With("remote", remoteID),
@@ -230,7 +237,6 @@ func newMemoryConnection(
remoteID: remoteID,
receiveCh: receiveCh,
sendCh: sendCh,
closer: closer,
}
}
@@ -264,7 +270,7 @@ func (c *MemoryConnection) Handshake(
select {
case c.sendCh <- memoryMessage{nodeInfo: &nodeInfo, pubKey: privKey.PubKey()}:
c.logger.Debug("sent handshake", "nodeInfo", nodeInfo)
case <-c.closer.Done():
case <-c.closeCh:
return types.NodeInfo{}, nil, io.EOF
case <-ctx.Done():
return types.NodeInfo{}, nil, ctx.Err()
@@ -277,7 +283,7 @@ func (c *MemoryConnection) Handshake(
}
c.logger.Debug("received handshake", "peerInfo", msg.nodeInfo)
return *msg.nodeInfo, msg.pubKey, nil
case <-c.closer.Done():
case <-c.closeCh:
return types.NodeInfo{}, nil, io.EOF
case <-ctx.Done():
return types.NodeInfo{}, nil, ctx.Err()
@@ -289,7 +295,7 @@ func (c *MemoryConnection) ReceiveMessage(ctx context.Context) (ChannelID, []byt
// Check close first, since channels are buffered. Otherwise, below select
// may non-deterministically return non-error even when closed.
select {
case <-c.closer.Done():
case <-c.closeCh:
return 0, nil, io.EOF
case <-ctx.Done():
return 0, nil, io.EOF
@@ -300,7 +306,9 @@ func (c *MemoryConnection) ReceiveMessage(ctx context.Context) (ChannelID, []byt
case msg := <-c.receiveCh:
c.logger.Debug("received message", "chID", msg.channelID, "msg", msg.message)
return msg.channelID, msg.message, nil
case <-c.closer.Done():
case <-ctx.Done():
return 0, nil, io.EOF
case <-c.closeCh:
return 0, nil, io.EOF
}
}
@@ -310,7 +318,7 @@ func (c *MemoryConnection) SendMessage(ctx context.Context, chID ChannelID, msg
// Check close first, since channels are buffered. Otherwise, below select
// may non-deterministically return non-error even when closed.
select {
case <-c.closer.Done():
case <-c.closeCh:
return io.EOF
case <-ctx.Done():
return io.EOF
@@ -323,19 +331,10 @@ func (c *MemoryConnection) SendMessage(ctx context.Context, chID ChannelID, msg
return nil
case <-ctx.Done():
return io.EOF
case <-c.closer.Done():
case <-c.closeCh:
return io.EOF
}
}
// Close implements Connection.
func (c *MemoryConnection) Close() error {
select {
case <-c.closer.Done():
return nil
default:
c.closer.Close()
c.logger.Info("closed connection")
}
return nil
}
func (c *MemoryConnection) Close() error { c.closeFn(); return nil }
+1
View File
@@ -6,6 +6,7 @@ import (
"testing"
"github.com/stretchr/testify/require"
"github.com/tendermint/tendermint/internal/p2p"
"github.com/tendermint/tendermint/libs/log"
"github.com/tendermint/tendermint/types"
+7 -24
View File
@@ -5,6 +5,7 @@ import (
"time"
"github.com/go-kit/kit/metrics"
abciclient "github.com/tendermint/tendermint/abci/client"
"github.com/tendermint/tendermint/abci/types"
)
@@ -24,9 +25,7 @@ type AppConnConsensus interface {
ProcessProposal(context.Context, types.RequestProcessProposal) (*types.ResponseProcessProposal, error)
ExtendVote(context.Context, types.RequestExtendVote) (*types.ResponseExtendVote, error)
VerifyVoteExtension(context.Context, types.RequestVerifyVoteExtension) (*types.ResponseVerifyVoteExtension, error)
BeginBlock(context.Context, types.RequestBeginBlock) (*types.ResponseBeginBlock, error)
DeliverTx(context.Context, types.RequestDeliverTx) (*types.ResponseDeliverTx, error)
EndBlock(context.Context, types.RequestEndBlock) (*types.ResponseEndBlock, error)
FinalizeBlock(context.Context, types.RequestFinalizeBlock) (*types.ResponseFinalizeBlock, error)
Commit(context.Context) (*types.ResponseCommit, error)
}
@@ -123,28 +122,12 @@ func (app *appConnConsensus) VerifyVoteExtension(
return app.appConn.VerifyVoteExtension(ctx, req)
}
func (app *appConnConsensus) BeginBlock(
func (app *appConnConsensus) FinalizeBlock(
ctx context.Context,
req types.RequestBeginBlock,
) (*types.ResponseBeginBlock, error) {
defer addTimeSample(app.metrics.MethodTiming.With("method", "begin_block", "type", "sync"))()
return app.appConn.BeginBlock(ctx, req)
}
func (app *appConnConsensus) DeliverTx(
ctx context.Context,
req types.RequestDeliverTx,
) (*types.ResponseDeliverTx, error) {
defer addTimeSample(app.metrics.MethodTiming.With("method", "deliver_tx", "type", "sync"))()
return app.appConn.DeliverTx(ctx, req)
}
func (app *appConnConsensus) EndBlock(
ctx context.Context,
req types.RequestEndBlock,
) (*types.ResponseEndBlock, error) {
defer addTimeSample(app.metrics.MethodTiming.With("method", "deliver_tx", "type", "sync"))()
return app.appConn.EndBlock(ctx, req)
req types.RequestFinalizeBlock,
) (*types.ResponseFinalizeBlock, error) {
defer addTimeSample(app.metrics.MethodTiming.With("method", "finalize_block", "type", "sync"))()
return app.appConn.FinalizeBlock(ctx, req)
}
func (app *appConnConsensus) Commit(ctx context.Context) (*types.ResponseCommit, error) {
+1
View File
@@ -7,6 +7,7 @@ import (
"testing"
"github.com/stretchr/testify/require"
abciclient "github.com/tendermint/tendermint/abci/client"
"github.com/tendermint/tendermint/abci/example/kvstore"
"github.com/tendermint/tendermint/abci/server"
+23 -69
View File
@@ -17,29 +17,6 @@ type AppConnConsensus struct {
mock.Mock
}
// BeginBlock provides a mock function with given fields: _a0, _a1
func (_m *AppConnConsensus) BeginBlock(_a0 context.Context, _a1 types.RequestBeginBlock) (*types.ResponseBeginBlock, error) {
ret := _m.Called(_a0, _a1)
var r0 *types.ResponseBeginBlock
if rf, ok := ret.Get(0).(func(context.Context, types.RequestBeginBlock) *types.ResponseBeginBlock); ok {
r0 = rf(_a0, _a1)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*types.ResponseBeginBlock)
}
}
var r1 error
if rf, ok := ret.Get(1).(func(context.Context, types.RequestBeginBlock) error); ok {
r1 = rf(_a0, _a1)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// Commit provides a mock function with given fields: _a0
func (_m *AppConnConsensus) Commit(_a0 context.Context) (*types.ResponseCommit, error) {
ret := _m.Called(_a0)
@@ -63,52 +40,6 @@ func (_m *AppConnConsensus) Commit(_a0 context.Context) (*types.ResponseCommit,
return r0, r1
}
// DeliverTx provides a mock function with given fields: _a0, _a1
func (_m *AppConnConsensus) DeliverTx(_a0 context.Context, _a1 types.RequestDeliverTx) (*types.ResponseDeliverTx, error) {
ret := _m.Called(_a0, _a1)
var r0 *types.ResponseDeliverTx
if rf, ok := ret.Get(0).(func(context.Context, types.RequestDeliverTx) *types.ResponseDeliverTx); ok {
r0 = rf(_a0, _a1)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*types.ResponseDeliverTx)
}
}
var r1 error
if rf, ok := ret.Get(1).(func(context.Context, types.RequestDeliverTx) error); ok {
r1 = rf(_a0, _a1)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// EndBlock provides a mock function with given fields: _a0, _a1
func (_m *AppConnConsensus) EndBlock(_a0 context.Context, _a1 types.RequestEndBlock) (*types.ResponseEndBlock, error) {
ret := _m.Called(_a0, _a1)
var r0 *types.ResponseEndBlock
if rf, ok := ret.Get(0).(func(context.Context, types.RequestEndBlock) *types.ResponseEndBlock); ok {
r0 = rf(_a0, _a1)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*types.ResponseEndBlock)
}
}
var r1 error
if rf, ok := ret.Get(1).(func(context.Context, types.RequestEndBlock) error); ok {
r1 = rf(_a0, _a1)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// Error provides a mock function with given fields:
func (_m *AppConnConsensus) Error() error {
ret := _m.Called()
@@ -146,6 +77,29 @@ func (_m *AppConnConsensus) ExtendVote(_a0 context.Context, _a1 types.RequestExt
return r0, r1
}
// FinalizeBlock provides a mock function with given fields: _a0, _a1
func (_m *AppConnConsensus) FinalizeBlock(_a0 context.Context, _a1 types.RequestFinalizeBlock) (*types.ResponseFinalizeBlock, error) {
ret := _m.Called(_a0, _a1)
var r0 *types.ResponseFinalizeBlock
if rf, ok := ret.Get(0).(func(context.Context, types.RequestFinalizeBlock) *types.ResponseFinalizeBlock); ok {
r0 = rf(_a0, _a1)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*types.ResponseFinalizeBlock)
}
}
var r1 error
if rf, ok := ret.Get(1).(func(context.Context, types.RequestFinalizeBlock) error); ok {
r1 = rf(_a0, _a1)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// InitChain provides a mock function with given fields: _a0, _a1
func (_m *AppConnConsensus) InitChain(_a0 context.Context, _a1 types.RequestInitChain) (*types.ResponseInitChain, error) {
ret := _m.Called(_a0, _a1)
+3
View File
@@ -281,6 +281,9 @@ func (s *Server) UnsubscribeAll(ctx context.Context, clientID string) error {
s.subs.Lock()
defer s.subs.Unlock()
if s.subs.index == nil {
return ErrServerStopped
}
evict := s.subs.index.findClientID(clientID)
if len(evict) == 0 {
return ErrSubscriptionNotFound
+1
View File
@@ -8,6 +8,7 @@ import (
"time"
"github.com/stretchr/testify/require"
abci "github.com/tendermint/tendermint/abci/types"
"github.com/tendermint/tendermint/internal/pubsub"
"github.com/tendermint/tendermint/internal/pubsub/query"
+1
View File
@@ -5,6 +5,7 @@ import (
"errors"
"github.com/google/uuid"
abci "github.com/tendermint/tendermint/abci/types"
"github.com/tendermint/tendermint/internal/libs/queue"
"github.com/tendermint/tendermint/types"
+5 -6
View File
@@ -208,18 +208,17 @@ func (env *Environment) BlockResults(ctx context.Context, heightPtr *int64) (*co
}
var totalGasUsed int64
for _, tx := range results.GetDeliverTxs() {
for _, tx := range results.FinalizeBlock.GetTxs() {
totalGasUsed += tx.GetGasUsed()
}
return &coretypes.ResultBlockResults{
Height: height,
TxsResults: results.DeliverTxs,
TxsResults: results.FinalizeBlock.Txs,
TotalGasUsed: totalGasUsed,
BeginBlockEvents: results.BeginBlock.Events,
EndBlockEvents: results.EndBlock.Events,
ValidatorUpdates: results.EndBlock.ValidatorUpdates,
ConsensusParamUpdates: results.EndBlock.ConsensusParamUpdates,
FinalizeBlockEvents: results.FinalizeBlock.Events,
ValidatorUpdates: results.FinalizeBlock.ValidatorUpdates,
ConsensusParamUpdates: results.FinalizeBlock.ConsensusParamUpdates,
}, nil
}
+10 -11
View File
@@ -71,13 +71,13 @@ func TestBlockchainInfo(t *testing.T) {
func TestBlockResults(t *testing.T) {
results := &tmstate.ABCIResponses{
DeliverTxs: []*abci.ResponseDeliverTx{
{Code: 0, Data: []byte{0x01}, Log: "ok", GasUsed: 10},
{Code: 0, Data: []byte{0x02}, Log: "ok", GasUsed: 5},
{Code: 1, Log: "not ok", GasUsed: 0},
FinalizeBlock: &abci.ResponseFinalizeBlock{
Txs: []*abci.ResponseDeliverTx{
{Code: 0, Data: []byte{0x01}, Log: "ok", GasUsed: 10},
{Code: 0, Data: []byte{0x02}, Log: "ok", GasUsed: 5},
{Code: 1, Log: "not ok", GasUsed: 0},
},
},
EndBlock: &abci.ResponseEndBlock{},
BeginBlock: &abci.ResponseBeginBlock{},
}
env := &Environment{}
@@ -99,12 +99,11 @@ func TestBlockResults(t *testing.T) {
{101, true, nil},
{100, false, &coretypes.ResultBlockResults{
Height: 100,
TxsResults: results.DeliverTxs,
TxsResults: results.FinalizeBlock.Txs,
TotalGasUsed: 15,
BeginBlockEvents: results.BeginBlock.Events,
EndBlockEvents: results.EndBlock.Events,
ValidatorUpdates: results.EndBlock.ValidatorUpdates,
ConsensusParamUpdates: results.EndBlock.ConsensusParamUpdates,
FinalizeBlockEvents: results.FinalizeBlock.Events,
ValidatorUpdates: results.FinalizeBlock.ValidatorUpdates,
ConsensusParamUpdates: results.FinalizeBlock.ConsensusParamUpdates,
}},
}
+1
View File
@@ -10,6 +10,7 @@ import (
"time"
"github.com/rs/cors"
"github.com/tendermint/tendermint/config"
"github.com/tendermint/tendermint/crypto"
"github.com/tendermint/tendermint/internal/blocksync"
+33 -51
View File
@@ -223,7 +223,7 @@ func (blockExec *BlockExecutor) ApplyBlock(
}
// validate the validator updates and convert to tendermint types
abciValUpdates := abciResponses.EndBlock.ValidatorUpdates
abciValUpdates := abciResponses.FinalizeBlock.ValidatorUpdates
err = validateValidatorUpdates(abciValUpdates, state.ConsensusParams.Validator)
if err != nil {
return state, fmt.Errorf("error in validator updates: %w", err)
@@ -244,7 +244,7 @@ func (blockExec *BlockExecutor) ApplyBlock(
}
// Lock mempool, commit app state, update mempoool.
appHash, retainHeight, err := blockExec.Commit(ctx, state, block, abciResponses.DeliverTxs)
appHash, retainHeight, err := blockExec.Commit(ctx, state, block, abciResponses.FinalizeBlock.Txs)
if err != nil {
return state, fmt.Errorf("commit failed for application: %w", err)
}
@@ -372,12 +372,10 @@ func execBlockOnProxyApp(
store Store,
initialHeight int64,
) (*tmstate.ABCIResponses, error) {
var validTxs, invalidTxs = 0, 0
txIndex := 0
abciResponses := new(tmstate.ABCIResponses)
abciResponses.FinalizeBlock = &abci.ResponseFinalizeBlock{}
dtxs := make([]*abci.ResponseDeliverTx, len(block.Txs))
abciResponses.DeliverTxs = dtxs
abciResponses.FinalizeBlock.Txs = dtxs
commitInfo := getBeginBlockValidatorInfo(block, store, initialHeight)
@@ -393,44 +391,22 @@ func execBlockOnProxyApp(
return nil, errors.New("nil header")
}
abciResponses.BeginBlock, err = proxyAppConn.BeginBlock(
abciResponses.FinalizeBlock, err = proxyAppConn.FinalizeBlock(
ctx,
abci.RequestBeginBlock{
abci.RequestFinalizeBlock{
Hash: block.Hash(),
Header: *pbh,
Height: block.Height,
LastCommitInfo: commitInfo,
ByzantineValidators: byzVals,
Txs: block.Txs.ToSliceOfBytes(),
},
)
if err != nil {
logger.Error("error in proxyAppConn.BeginBlock", "err", err)
logger.Error("error in proxyAppConn.FinalizeBlock", "err", err)
return nil, err
}
// run txs of block
for _, tx := range block.Txs {
resp, err := proxyAppConn.DeliverTx(ctx, abci.RequestDeliverTx{Tx: tx})
if err != nil {
return nil, err
}
if resp.Code == abci.CodeTypeOK {
validTxs++
} else {
logger.Debug("invalid tx", "code", resp.Code, "log", resp.Log)
invalidTxs++
}
abciResponses.DeliverTxs[txIndex] = resp
txIndex++
}
abciResponses.EndBlock, err = proxyAppConn.EndBlock(ctx, abci.RequestEndBlock{Height: block.Height})
if err != nil {
logger.Error("error in proxyAppConn.EndBlock", "err", err)
return nil, err
}
logger.Info("executed block", "height", block.Height, "num_valid_txs", validTxs, "num_invalid_txs", invalidTxs)
logger.Info("executed block", "height", block.Height)
return abciResponses, nil
}
@@ -529,9 +505,9 @@ func updateState(
// Update the params with the latest abciResponses.
nextParams := state.ConsensusParams
lastHeightParamsChanged := state.LastHeightConsensusParamsChanged
if abciResponses.EndBlock.ConsensusParamUpdates != nil {
if abciResponses.FinalizeBlock.ConsensusParamUpdates != nil {
// NOTE: must not mutate s.ConsensusParams
nextParams = state.ConsensusParams.UpdateConsensusParams(abciResponses.EndBlock.ConsensusParamUpdates)
nextParams = state.ConsensusParams.UpdateConsensusParams(abciResponses.FinalizeBlock.ConsensusParamUpdates)
err := nextParams.ValidateConsensusParams()
if err != nil {
return state, fmt.Errorf("error updating consensus params: %w", err)
@@ -578,19 +554,17 @@ func fireEvents(
validatorUpdates []*types.Validator,
) {
if err := eventBus.PublishEventNewBlock(ctx, types.EventDataNewBlock{
Block: block,
BlockID: blockID,
ResultBeginBlock: *abciResponses.BeginBlock,
ResultEndBlock: *abciResponses.EndBlock,
Block: block,
BlockID: blockID,
ResultFinalizeBlock: *abciResponses.FinalizeBlock,
}); err != nil {
logger.Error("failed publishing new block", "err", err)
}
if err := eventBus.PublishEventNewBlockHeader(ctx, types.EventDataNewBlockHeader{
Header: block.Header,
NumTxs: int64(len(block.Txs)),
ResultBeginBlock: *abciResponses.BeginBlock,
ResultEndBlock: *abciResponses.EndBlock,
Header: block.Header,
NumTxs: int64(len(block.Txs)),
ResultFinalizeBlock: *abciResponses.FinalizeBlock,
}); err != nil {
logger.Error("failed publishing new block header", "err", err)
}
@@ -606,13 +580,21 @@ func fireEvents(
}
}
// sanity check
if len(abciResponses.FinalizeBlock.Txs) != len(block.Data.Txs) {
panic(fmt.Sprintf("number of TXs (%d) and ABCI TX responses (%d) do not match",
len(block.Data.Txs), len(abciResponses.FinalizeBlock.Txs)))
}
for i, tx := range block.Data.Txs {
if err := eventBus.PublishEventTx(ctx, types.EventDataTx{TxResult: abci.TxResult{
Height: block.Height,
Index: uint32(i),
Tx: tx,
Result: *(abciResponses.DeliverTxs[i]),
}}); err != nil {
if err := eventBus.PublishEventTx(ctx, types.EventDataTx{
TxResult: abci.TxResult{
Height: block.Height,
Index: uint32(i),
Tx: tx,
Result: *(abciResponses.FinalizeBlock.Txs[i]),
},
}); err != nil {
logger.Error("failed publishing event TX", "err", err)
}
}
@@ -648,7 +630,7 @@ func ExecCommitBlock(
// the BlockExecutor condition is using for the final block replay process.
if be != nil {
abciValUpdates := abciResponses.EndBlock.ValidatorUpdates
abciValUpdates := abciResponses.FinalizeBlock.ValidatorUpdates
err = validateValidatorUpdates(abciValUpdates, s.ConsensusParams.Validator)
if err != nil {
logger.Error("err", err)
+22 -18
View File
@@ -156,8 +156,7 @@ func makeHeaderPartsResponsesValPubKeyChange(
block, err := sf.MakeBlock(state, state.LastBlockHeight+1, new(types.Commit))
require.NoError(t, err)
abciResponses := &tmstate.ABCIResponses{
BeginBlock: &abci.ResponseBeginBlock{},
EndBlock: &abci.ResponseEndBlock{ValidatorUpdates: nil},
FinalizeBlock: &abci.ResponseFinalizeBlock{ValidatorUpdates: nil},
}
// If the pubkey is new, remove the old and add the new.
_, val := state.NextValidators.GetByIndex(0)
@@ -167,7 +166,7 @@ func makeHeaderPartsResponsesValPubKeyChange(
pbPk, err := encoding.PubKeyToProto(pubkey)
require.NoError(t, err)
abciResponses.EndBlock = &abci.ResponseEndBlock{
abciResponses.FinalizeBlock = &abci.ResponseFinalizeBlock{
ValidatorUpdates: []abci.ValidatorUpdate{
{PubKey: vPbPk, Power: 0},
{PubKey: pbPk, Power: 10},
@@ -189,8 +188,7 @@ func makeHeaderPartsResponsesValPowerChange(
require.NoError(t, err)
abciResponses := &tmstate.ABCIResponses{
BeginBlock: &abci.ResponseBeginBlock{},
EndBlock: &abci.ResponseEndBlock{ValidatorUpdates: nil},
FinalizeBlock: &abci.ResponseFinalizeBlock{ValidatorUpdates: nil},
}
// If the pubkey is new, remove the old and add the new.
@@ -199,7 +197,7 @@ func makeHeaderPartsResponsesValPowerChange(
vPbPk, err := encoding.PubKeyToProto(val.PubKey)
require.NoError(t, err)
abciResponses.EndBlock = &abci.ResponseEndBlock{
abciResponses.FinalizeBlock = &abci.ResponseFinalizeBlock{
ValidatorUpdates: []abci.ValidatorUpdate{
{PubKey: vPbPk, Power: power},
},
@@ -220,8 +218,7 @@ func makeHeaderPartsResponsesParams(
require.NoError(t, err)
pbParams := params.ToProto()
abciResponses := &tmstate.ABCIResponses{
BeginBlock: &abci.ResponseBeginBlock{},
EndBlock: &abci.ResponseEndBlock{ConsensusParamUpdates: &pbParams},
FinalizeBlock: &abci.ResponseFinalizeBlock{ConsensusParamUpdates: &pbParams},
}
return block.Header, types.BlockID{Hash: block.Hash(), PartSetHeader: types.PartSetHeader{}}, abciResponses
}
@@ -298,22 +295,29 @@ func (app *testApp) Info(req abci.RequestInfo) (resInfo abci.ResponseInfo) {
return abci.ResponseInfo{}
}
func (app *testApp) BeginBlock(req abci.RequestBeginBlock) abci.ResponseBeginBlock {
func (app *testApp) FinalizeBlock(req abci.RequestFinalizeBlock) abci.ResponseFinalizeBlock {
app.CommitVotes = req.LastCommitInfo.Votes
app.ByzantineValidators = req.ByzantineValidators
return abci.ResponseBeginBlock{}
}
func (app *testApp) EndBlock(req abci.RequestEndBlock) abci.ResponseEndBlock {
return abci.ResponseEndBlock{
resTxs := make([]*abci.ResponseDeliverTx, len(req.Txs))
for i, tx := range req.Txs {
if len(tx) > 0 {
resTxs[i] = &abci.ResponseDeliverTx{Code: abci.CodeTypeOK}
} else {
resTxs[i] = &abci.ResponseDeliverTx{Code: abci.CodeTypeOK + 10} // error
}
}
return abci.ResponseFinalizeBlock{
ValidatorUpdates: app.ValidatorUpdates,
ConsensusParamUpdates: &tmproto.ConsensusParams{
Version: &tmproto.VersionParams{
AppVersion: 1}}}
}
func (app *testApp) DeliverTx(req abci.RequestDeliverTx) abci.ResponseDeliverTx {
return abci.ResponseDeliverTx{Events: []abci.Event{}}
AppVersion: 1,
},
},
Events: []abci.Event{},
Txs: resTxs,
}
}
func (app *testApp) CheckTx(req abci.RequestCheckTx) abci.ResponseCheckTx {
+2 -7
View File
@@ -66,13 +66,8 @@ func (idx *BlockerIndexer) Index(bh types.EventDataNewBlockHeader) error {
}
// 2. index BeginBlock events
if err := idx.indexEvents(batch, bh.ResultBeginBlock.Events, "begin_block", height); err != nil {
return fmt.Errorf("failed to index BeginBlock events: %w", err)
}
// 3. index EndBlock events
if err := idx.indexEvents(batch, bh.ResultEndBlock.Events, "end_block", height); err != nil {
return fmt.Errorf("failed to index EndBlock events: %w", err)
if err := idx.indexEvents(batch, bh.ResultFinalizeBlock.Events, "finalize_block", height); err != nil {
return fmt.Errorf("failed to index FinalizeBlock events: %w", err)
}
return batch.WriteSync()
+14 -23
View File
@@ -20,10 +20,10 @@ func TestBlockIndexer(t *testing.T) {
require.NoError(t, indexer.Index(types.EventDataNewBlockHeader{
Header: types.Header{Height: 1},
ResultBeginBlock: abci.ResponseBeginBlock{
ResultFinalizeBlock: abci.ResponseFinalizeBlock{
Events: []abci.Event{
{
Type: "begin_event",
Type: "finalize_event1",
Attributes: []abci.EventAttribute{
{
Key: "proposer",
@@ -32,12 +32,8 @@ func TestBlockIndexer(t *testing.T) {
},
},
},
},
},
ResultEndBlock: abci.ResponseEndBlock{
Events: []abci.Event{
{
Type: "end_event",
Type: "finalize_event2",
Attributes: []abci.EventAttribute{
{
Key: "foo",
@@ -55,13 +51,12 @@ func TestBlockIndexer(t *testing.T) {
if i%2 == 0 {
index = true
}
require.NoError(t, indexer.Index(types.EventDataNewBlockHeader{
Header: types.Header{Height: int64(i)},
ResultBeginBlock: abci.ResponseBeginBlock{
ResultFinalizeBlock: abci.ResponseFinalizeBlock{
Events: []abci.Event{
{
Type: "begin_event",
Type: "finalize_event1",
Attributes: []abci.EventAttribute{
{
Key: "proposer",
@@ -70,12 +65,8 @@ func TestBlockIndexer(t *testing.T) {
},
},
},
},
},
ResultEndBlock: abci.ResponseEndBlock{
Events: []abci.Event{
{
Type: "end_event",
Type: "finalize_event2",
Attributes: []abci.EventAttribute{
{
Key: "foo",
@@ -102,31 +93,31 @@ func TestBlockIndexer(t *testing.T) {
results: []int64{5},
},
"begin_event.key1 = 'value1'": {
q: query.MustCompile(`begin_event.key1 = 'value1'`),
q: query.MustCompile(`finalize_event1.key1 = 'value1'`),
results: []int64{},
},
"begin_event.proposer = 'FCAA001'": {
q: query.MustCompile(`begin_event.proposer = 'FCAA001'`),
q: query.MustCompile(`finalize_event1.proposer = 'FCAA001'`),
results: []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11},
},
"end_event.foo <= 5": {
q: query.MustCompile(`end_event.foo <= 5`),
q: query.MustCompile(`finalize_event2.foo <= 5`),
results: []int64{2, 4},
},
"end_event.foo >= 100": {
q: query.MustCompile(`end_event.foo >= 100`),
q: query.MustCompile(`finalize_event2.foo >= 100`),
results: []int64{1},
},
"block.height > 2 AND end_event.foo <= 8": {
q: query.MustCompile(`block.height > 2 AND end_event.foo <= 8`),
"block.height > 2 AND finalize_event2.foo <= 8": {
q: query.MustCompile(`block.height > 2 AND finalize_event2.foo <= 8`),
results: []int64{4, 6, 8},
},
"begin_event.proposer CONTAINS 'FFFFFFF'": {
q: query.MustCompile(`begin_event.proposer CONTAINS 'FFFFFFF'`),
q: query.MustCompile(`finalize_event1.proposer CONTAINS 'FFFFFFF'`),
results: []int64{},
},
"begin_event.proposer CONTAINS 'FCAA001'": {
q: query.MustCompile(`begin_event.proposer CONTAINS 'FCAA001'`),
q: query.MustCompile(`finalize_event1.proposer CONTAINS 'FCAA001'`),
results: []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11},
},
}
+1
View File
@@ -6,6 +6,7 @@ import (
"strconv"
"github.com/google/orderedcode"
"github.com/tendermint/tendermint/internal/pubsub/query/syntax"
"github.com/tendermint/tendermint/types"
)
@@ -137,6 +137,9 @@ func setupDB(t *testing.T) (*dockertest.Pool, error) {
t.Helper()
pool, err := dockertest.NewPool(os.Getenv("DOCKER_URL"))
assert.NoError(t, err)
if _, err := pool.Client.Info(); err != nil {
t.Skipf("WARNING: Docker is not available: %v [skipping this test]", err)
}
resource, err = pool.RunWithOptions(&dockertest.RunOptions{
Repository: "postgres",
@@ -6,6 +6,7 @@ import (
context "context"
mock "github.com/stretchr/testify/mock"
indexer "github.com/tendermint/tendermint/internal/state/indexer"
query "github.com/tendermint/tendermint/internal/pubsub/query"

Some files were not shown because too many files have changed in this diff Show More