mirrors/tendermint
1
0
Fork 0
mirror of https://github.com/tendermint/tendermint.git synced 2026-09-20 15:04:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into wb/pbts-rebase-master

Browse Source
This commit is contained in:
William Banfield
2022-01-20 18:18:24 -05:00
parent 4d1ad8ec38 db7d4abdae
commit b853afe2ce
63 changed files with 811 additions and 1236 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/consensus/common_test.go
+35 -19
View File
@@ -282,6 +282,9 @@ func validatePrevote(
) {
t.Helper()
cs.mtx.RLock()
defer cs.mtx.RUnlock()
prevotes := cs.Votes.Prevotes(round)
pubKey, err := privVal.GetPubKey(ctx)
require.NoError(t, err)
@@ -355,25 +358,6 @@ func validatePrecommit(
}
}
func validatePrevoteAndPrecommit(
ctx context.Context,
t *testing.T,
cs *State,
thisRound,
lockRound int32,
privVal *validatorStub,
votedBlockHash,
lockedBlockHash []byte,
) {
t.Helper()
// verify the prevote
validatePrevote(ctx, t, cs, thisRound, privVal, votedBlockHash)
// verify precommit
cs.mtx.Lock()
defer cs.mtx.Unlock()
validatePrecommit(ctx, t, cs, thisRound, lockRound, privVal, votedBlockHash, lockedBlockHash)
}
func subscribeToVoter(ctx context.Context, t *testing.T, cs *State, addr []byte) <-chan tmpubsub.Message {
t.Helper()
@@ -683,6 +667,38 @@ func ensurePrevote(t *testing.T, voteCh <-chan tmpubsub.Message, height int64, r
ensureVote(t, voteCh, height, round, tmproto.PrevoteType)
}
func ensurePrevoteMatch(t *testing.T, voteCh <-chan tmpubsub.Message, height int64, round int32, hash []byte) {
t.Helper()
ensureVoteMatch(t, voteCh, height, round, hash, tmproto.PrevoteType)
}
func ensurePrecommitMatch(t *testing.T, voteCh <-chan tmpubsub.Message, height int64, round int32, hash []byte) {
t.Helper()
ensureVoteMatch(t, voteCh, height, round, hash, tmproto.PrecommitType)
}
func ensureVoteMatch(t *testing.T, voteCh <-chan tmpubsub.Message, height int64, round int32, hash []byte, voteType tmproto.SignedMsgType) {
t.Helper()
select {
case <-time.After(ensureTimeout):
t.Fatal("Timeout expired while waiting for NewVote event")
case msg := <-voteCh:
voteEvent, ok := msg.Data().(types.EventDataVote)
require.True(t, ok, "expected a EventDataVote, got %T. Wrong subscription channel?",
msg.Data())
vote := voteEvent.Vote
require.Equal(t, height, vote.Height)
require.Equal(t, round, vote.Round)
require.Equal(t, voteType, vote.Type)
if hash == nil {
require.Nil(t, vote.BlockID.Hash, "Expected prevote to be for nil, got %X", vote.BlockID.Hash)
} else {
require.True(t, bytes.Equal(vote.BlockID.Hash, hash), "Expected prevote to be for %X, got %X", hash, vote.BlockID.Hash)
}
}
}
func ensureVote(t *testing.T, voteCh <-chan tmpubsub.Message, height int64, round int32, voteType tmproto.SignedMsgType) {
t.Helper()
msg := ensureMessageBeforeTimeout(t, voteCh, ensureTimeout)
internal/consensus/reactor_test.go
+3 -5
View File
@@ -81,7 +81,7 @@ func setup(
rts.voteSetBitsChannels = rts.network.MakeChannelsNoCleanup(ctx, t, chDesc(VoteSetBitsChannel, size))
ctx, cancel := context.WithCancel(ctx)
// Canceled during cleanup (see below).
t.Cleanup(cancel)
chCreator := func(nodeID types.NodeID) p2p.ChannelCreator {
return func(ctx context.Context, desc *p2p.ChannelDescriptor) (*p2p.Channel, error) {
@@ -142,6 +142,7 @@ func setup(
require.NoError(t, reactor.Start(ctx))
require.True(t, reactor.IsRunning())
t.Cleanup(reactor.Wait)
i++
}
@@ -151,10 +152,7 @@ func setup(
// start the in-memory network and connect all peers with each other
rts.network.Start(ctx, t)
t.Cleanup(func() {
cancel()
leaktest.Check(t)
})
t.Cleanup(leaktest.Check(t))
return rts
}
internal/consensus/replay.go
+2 -2
View File
@@ -240,7 +240,7 @@ func (h *Handshaker) NBlocks() int {
func (h *Handshaker) Handshake(ctx context.Context, proxyApp proxy.AppConns) error {
// Handshake is done via ABCI Info on the query conn.
res, err := proxyApp.Query().InfoSync(ctx, proxy.RequestInfo)
res, err := proxyApp.Query().Info(ctx, proxy.RequestInfo)
if err != nil {
return fmt.Errorf("error calling Info: %w", err)
}
@@ -316,7 +316,7 @@ func (h *Handshaker) ReplayBlocks(
Validators: nextVals,
AppStateBytes: h.genDoc.AppState,
}
res, err := proxyApp.Consensus().InitChainSync(ctx, req)
res, err := proxyApp.Consensus().InitChain(ctx, req)
if err != nil {
return nil, err
}
internal/consensus/replay_test.go
+15 -21
View File
@@ -686,26 +686,20 @@ func TestMockProxyApp(t *testing.T) {
abciRes := new(tmstate.ABCIResponses)
abciRes.DeliverTxs = make([]*abci.ResponseDeliverTx, len(loadedAbciRes.DeliverTxs))
// Execute transactions and get hash.
proxyCb := func(req *abci.Request, res *abci.Response) {
if r, ok := res.Value.(*abci.Response_DeliverTx); ok {
// TODO: make use of res.Log
// TODO: make use of this info
// Blocks may include invalid txs.
txRes := r.DeliverTx
if txRes.Code == abci.CodeTypeOK {
validTxs++
} else {
invalidTxs++
}
abciRes.DeliverTxs[txIndex] = txRes
txIndex++
}
}
mock.SetResponseCallback(proxyCb)
someTx := []byte("tx")
_, err = mock.DeliverTxAsync(ctx, abci.RequestDeliverTx{Tx: someTx})
resp, err := mock.DeliverTx(ctx, abci.RequestDeliverTx{Tx: someTx})
// 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++
}
abciRes.DeliverTxs[txIndex] = resp
txIndex++
assert.NoError(t, err)
})
assert.True(t, validTxs == 1)
@@ -847,7 +841,7 @@ func testHandshakeReplay(
require.NoError(t, err, "Error on abci handshake")
// get the latest app hash from the app
res, err := proxyApp.Query().InfoSync(ctx, abci.RequestInfo{Version: ""})
res, err := proxyApp.Query().Info(ctx, abci.RequestInfo{Version: ""})
if err != nil {
t.Fatal(err)
}
@@ -913,7 +907,7 @@ func buildAppStateFromChain(
state.Version.Consensus.App = kvstore.ProtocolVersion // simulate handshake, receive app version
validators := types.TM2PB.ValidatorUpdates(state.Validators)
_, err := proxyApp.Consensus().InitChainSync(ctx, abci.RequestInitChain{
_, err := proxyApp.Consensus().InitChain(ctx, abci.RequestInitChain{
Validators: validators,
})
require.NoError(t, err)
@@ -970,7 +964,7 @@ func buildTMStateFromChain(
state.Version.Consensus.App = kvstore.ProtocolVersion // simulate handshake, receive app version
validators := types.TM2PB.ValidatorUpdates(state.Validators)
_, err := proxyApp.Consensus().InitChainSync(ctx, abci.RequestInitChain{
_, err := proxyApp.Consensus().InitChain(ctx, abci.RequestInitChain{
Validators: validators,
})
require.NoError(t, err)
internal/consensus/state.go
+5
View File
@@ -2451,11 +2451,16 @@ func (cs *State) checkDoubleSigningRisk(height int64) error {
}
func (cs *State) calculatePrevoteMessageDelayMetrics() {
if cs.Proposal == nil {
return
}
ps := cs.Votes.Prevotes(cs.Round)
pl := ps.List()
sort.Slice(pl, func(i, j int) bool {
return pl[i].Timestamp.Before(pl[j].Timestamp)
})
var votingPowerSeen int64
for _, v := range pl {
_, val := cs.Validators.GetByAddress(v.ValidatorAddress)
internal/consensus/state_test.go
+26 -60
View File
@@ -254,8 +254,7 @@ func TestStateBadProposal(t *testing.T) {
ensureProposal(t, proposalCh, height, round, blockID)
// wait for prevote
ensurePrevote(t, voteCh, height, round)
validatePrevote(ctx, t, cs1, round, vss[0], nil)
ensurePrevoteMatch(t, voteCh, height, round, nil)
// add bad prevote from vs2 and wait for it
signAddVotes(ctx, t, cs1, tmproto.PrevoteType, config.ChainID(), blockID, vs2)
@@ -321,8 +320,7 @@ func TestStateOversizedBlock(t *testing.T) {
// and then should send nil prevote and precommit regardless of whether other validators prevote and
// precommit on it
ensurePrevote(t, voteCh, height, round)
validatePrevote(ctx, t, cs1, round, vss[0], nil)
ensurePrevoteMatch(t, voteCh, height, round, nil)
signAddVotes(ctx, t, cs1, tmproto.PrevoteType, config.ChainID(), blockID, vs2)
ensurePrevote(t, voteCh, height, round)
ensurePrecommit(t, voteCh, height, round)
@@ -343,19 +341,9 @@ func TestStateFullRound1(t *testing.T) {
cs, vss := makeState(ctx, t, config, logger, 1)
height, round := cs.Height, cs.Round
// NOTE: buffer capacity of 0 ensures we can validate prevote and last commit
// before consensus can move to the next height (and cause a race condition)
if err := cs.eventBus.Stop(); err != nil {
t.Error(err)
}
eventBus := eventbus.NewDefault(logger.With("module", "events"))
cs.SetEventBus(eventBus)
if err := eventBus.Start(ctx); err != nil {
t.Error(err)
}
voteCh := subscribe(ctx, t, cs.eventBus, types.EventQueryVote)
pv, err := cs.privValidator.GetPubKey(ctx)
require.NoError(t, err)
voteCh := subscribeToVoter(ctx, t, cs, pv.Address())
propCh := subscribe(ctx, t, cs.eventBus, types.EventQueryCompleteProposal)
newRoundCh := subscribe(ctx, t, cs.eventBus, types.EventQueryNewRound)
@@ -367,8 +355,7 @@ func TestStateFullRound1(t *testing.T) {
ensureNewProposal(t, propCh, height, round)
propBlockHash := cs.GetRoundState().ProposalBlock.Hash()
ensurePrevote(t, voteCh, height, round) // wait for prevote
validatePrevote(ctx, t, cs, round, vss[0], propBlockHash)
ensurePrevoteMatch(t, voteCh, height, round, propBlockHash) // wait for prevote
ensurePrecommit(t, voteCh, height, round) // wait for precommit
@@ -385,7 +372,7 @@ func TestStateFullRoundNil(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
cs, vss := makeState(ctx, t, config, logger, 1)
cs, _ := makeState(ctx, t, config, logger, 1)
height, round := cs.Height, cs.Round
voteCh := subscribe(ctx, t, cs.eventBus, types.EventQueryVote)
@@ -393,11 +380,8 @@ func TestStateFullRoundNil(t *testing.T) {
cs.enterPrevote(ctx, height, round)
cs.startRoutines(ctx, 4)
ensurePrevote(t, voteCh, height, round) // prevote
ensurePrecommit(t, voteCh, height, round) // precommit
// should prevote and precommit nil
validatePrevoteAndPrecommit(ctx, t, cs, round, -1, vss[0], nil, nil)
ensurePrevoteMatch(t, voteCh, height, round, nil) // prevote
ensurePrecommitMatch(t, voteCh, height, round, nil) // precommit
}
// run through propose, prevote, precommit commit with two validators
@@ -746,8 +730,7 @@ func TestStateLock_POLUpdateLock(t *testing.T) {
ensureNewProposal(t, proposalCh, height, round)
// Prevote our nil since the proposal does not match our locked block.
ensurePrevote(t, voteCh, height, round)
validatePrevote(ctx, t, cs1, round, vss[0], nil)
ensurePrevoteMatch(t, voteCh, height, round, nil)
// Add prevotes from the remainder of the validators for the new locked block.
signAddVotes(ctx, t, cs1, tmproto.PrevoteType, config.ChainID(), r1BlockID, vs2, vs3, vs4)
@@ -1096,8 +1079,7 @@ func TestStateLock_POLDoesNotUnlock(t *testing.T) {
PartSetHeader: rs.ProposalBlockParts.Header(),
}
ensurePrevote(t, voteCh, height, round)
validatePrevote(ctx, t, cs1, round, vss[0], blockID.Hash)
ensurePrevoteMatch(t, voteCh, height, round, blockID.Hash)
signAddVotes(ctx, t, cs1, tmproto.PrevoteType, config.ChainID(), blockID, vs2, vs3, vs4)
@@ -1142,8 +1124,7 @@ func TestStateLock_POLDoesNotUnlock(t *testing.T) {
ensureNewProposal(t, proposalCh, height, round)
// Prevote for nil since the proposed block does not match our locked block.
ensurePrevote(t, voteCh, height, round)
validatePrevote(ctx, t, cs1, round, vss[0], nil)
ensurePrevoteMatch(t, voteCh, height, round, nil)
// add >2/3 prevotes for nil from all other validators
signAddVotes(ctx, t, cs1, tmproto.PrevoteType, config.ChainID(), types.BlockID{}, vs2, vs3, vs4)
@@ -1270,8 +1251,7 @@ func TestStateLock_MissingProposalWhenPOLSeenDoesNotUpdateLock(t *testing.T) {
ensureNewRound(t, newRoundCh, height, round)
// prevote for nil since the proposal was not seen.
ensurePrevote(t, voteCh, height, round)
validatePrevote(ctx, t, cs1, round, vss[0], nil)
ensurePrevoteMatch(t, voteCh, height, round, nil)
// now lets add prevotes from everyone else for the new block
signAddVotes(ctx, t, cs1, tmproto.PrevoteType, config.ChainID(), secondBlockID, vs2, vs3, vs4)
@@ -1391,8 +1371,7 @@ func TestStateLock_POLSafety1(t *testing.T) {
rs := cs1.GetRoundState()
propBlock := rs.ProposalBlock
ensurePrevote(t, voteCh, height, round)
validatePrevote(ctx, t, cs1, round, vss[0], propBlock.Hash())
ensurePrevoteMatch(t, voteCh, height, round, propBlock.Hash())
partSet, err := propBlock.MakePartSet(partSize)
require.NoError(t, err)
blockID := types.BlockID{Hash: propBlock.Hash(), PartSetHeader: partSet.Header()}
@@ -1439,8 +1418,7 @@ func TestStateLock_POLSafety1(t *testing.T) {
t.Logf("new prop hash %v", fmt.Sprintf("%X", propBlock.Hash()))
// go to prevote, prevote for proposal block
ensurePrevote(t, voteCh, height, round)
validatePrevote(ctx, t, cs1, round, vss[0], r2BlockID.Hash)
ensurePrevoteMatch(t, voteCh, height, round, r2BlockID.Hash)
// now we see the others prevote for it, so we should lock on it
signAddVotes(ctx, t, cs1, tmproto.PrevoteType, config.ChainID(), r2BlockID, vs2, vs3, vs4)
@@ -1467,9 +1445,7 @@ func TestStateLock_POLSafety1(t *testing.T) {
ensureNewTimeout(t, timeoutProposeCh, height, round, cs1.config.Propose(round).Nanoseconds())
// finish prevote
ensurePrevote(t, voteCh, height, round)
// we should prevote for nil
validatePrevote(ctx, t, cs1, round, vss[0], nil)
ensurePrevoteMatch(t, voteCh, height, round, nil)
newStepCh := subscribe(ctx, t, cs1.eventBus, types.EventQueryNewRoundStep)
@@ -1536,8 +1512,7 @@ func TestStateLock_POLSafety2(t *testing.T) {
require.NoError(t, err)
ensureNewProposal(t, proposalCh, height, round)
ensurePrevote(t, voteCh, height, round)
validatePrevote(ctx, t, cs1, round, vss[0], propBlockID1.Hash)
ensurePrevoteMatch(t, voteCh, height, round, propBlockID1.Hash)
signAddVotes(ctx, t, cs1, tmproto.PrevoteType, config.ChainID(), propBlockID1, vs2, vs3, vs4)
@@ -1762,8 +1737,7 @@ func TestProposeValidBlock(t *testing.T) {
PartSetHeader: partSet.Header(),
}
ensurePrevote(t, voteCh, height, round)
validatePrevote(ctx, t, cs1, round, vss[0], blockID.Hash)
ensurePrevoteMatch(t, voteCh, height, round, blockID.Hash)
// the others sign a polka
signAddVotes(ctx, t, cs1, tmproto.PrevoteType, cfg.ChainID(), blockID, vs2, vs3, vs4)
@@ -1787,8 +1761,7 @@ func TestProposeValidBlock(t *testing.T) {
ensureNewTimeout(t, timeoutProposeCh, height, round, cs1.config.Propose(round).Nanoseconds())
// We did not see a valid proposal within this round, so prevote nil.
ensurePrevote(t, voteCh, height, round)
validatePrevote(ctx, t, cs1, round, vss[0], nil)
ensurePrevoteMatch(t, voteCh, height, round, nil)
signAddVotes(ctx, t, cs1, tmproto.PrecommitType, cfg.ChainID(), types.BlockID{}, vs2, vs3, vs4)
@@ -1859,8 +1832,7 @@ func TestSetValidBlockOnDelayedPrevote(t *testing.T) {
PartSetHeader: partSet.Header(),
}
ensurePrevote(t, voteCh, height, round)
validatePrevote(ctx, t, cs1, round, vss[0], blockID.Hash)
ensurePrevoteMatch(t, voteCh, height, round, blockID.Hash)
// vs2 send prevote for propBlock
signAddVotes(ctx, t, cs1, tmproto.PrevoteType, config.ChainID(), blockID, vs2)
@@ -1925,8 +1897,7 @@ func TestSetValidBlockOnDelayedProposal(t *testing.T) {
ensureNewTimeout(t, timeoutProposeCh, height, round, cs1.config.Propose(round).Nanoseconds())
ensurePrevote(t, voteCh, height, round)
validatePrevote(ctx, t, cs1, round, vss[0], nil)
ensurePrevoteMatch(t, voteCh, height, round, nil)
prop, propBlock := decideProposal(ctx, t, cs1, vs2, vs2.Height, vs2.Round+1)
partSet, err := propBlock.MakePartSet(partSize)
@@ -2021,8 +1992,7 @@ func TestWaitingTimeoutProposeOnNewRound(t *testing.T) {
ensureNewTimeout(t, timeoutWaitCh, height, round, cs1.config.Propose(round).Nanoseconds())
ensurePrevote(t, voteCh, height, round)
validatePrevote(ctx, t, cs1, round, vss[0], nil)
ensurePrevoteMatch(t, voteCh, height, round, nil)
}
// 4 vals, 3 Precommits for nil from the higher round.
@@ -2095,8 +2065,7 @@ func TestWaitTimeoutProposeOnNilPolkaForTheCurrentRound(t *testing.T) {
ensureNewTimeout(t, timeoutProposeCh, height, round, cs1.config.Propose(round).Nanoseconds())
ensurePrevote(t, voteCh, height, round)
validatePrevote(ctx, t, cs1, round, vss[0], nil)
ensurePrevoteMatch(t, voteCh, height, round, nil)
}
// What we want:
@@ -2241,8 +2210,7 @@ func TestStartNextHeightCorrectlyAfterTimeout(t *testing.T) {
PartSetHeader: rs.ProposalBlockParts.Header(),
}
ensurePrevote(t, voteCh, height, round)
validatePrevote(ctx, t, cs1, round, vss[0], blockID.Hash)
ensurePrevoteMatch(t, voteCh, height, round, blockID.Hash)
signAddVotes(ctx, t, cs1, tmproto.PrevoteType, config.ChainID(), blockID, vs2, vs3, vs4)
@@ -2308,8 +2276,7 @@ func TestResetTimeoutPrecommitUponNewHeight(t *testing.T) {
PartSetHeader: rs.ProposalBlockParts.Header(),
}
ensurePrevote(t, voteCh, height, round)
validatePrevote(ctx, t, cs1, round, vss[0], blockID.Hash)
ensurePrevoteMatch(t, voteCh, height, round, blockID.Hash)
signAddVotes(ctx, t, cs1, tmproto.PrevoteType, config.ChainID(), blockID, vs2, vs3, vs4)
@@ -2410,8 +2377,7 @@ func TestStateHalt1(t *testing.T) {
*/
// prevote for nil since we did not receive a proposal in this round.
ensurePrevote(t, voteCh, height, round)
validatePrevote(ctx, t, cs1, round, vss[0], nil)
ensurePrevoteMatch(t, voteCh, height, round, rs.LockedBlock.Hash())
// now we receive the precommit from the previous round
addVotes(cs1, precommit4)
internal/consensus/wal_test.go
+5
View File
@@ -9,6 +9,7 @@ import (
"testing"
"time"
"github.com/fortytw2/leaktest"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -50,6 +51,10 @@ func TestWALTruncate(t *testing.T) {
err = WALGenerateNBlocks(ctx, t, logger, wal.Group(), 60)
require.NoError(t, err)
// put the leakcheck here so it runs after other cleanup
// functions.
t.Cleanup(leaktest.CheckTimeout(t, 500*time.Millisecond))
time.Sleep(1 * time.Millisecond) // wait groupCheckDuration, make sure RotateFile run
if err := wal.FlushAndSync(); err != nil {
internal/libs/autofile/group.go
+26 -14
View File
@@ -56,7 +56,6 @@ assuming that marker lines are written occasionally.
type Group struct {
service.BaseService
logger log.Logger
ctx context.Context
ID string
Head *AutoFile // The head AutoFile to write to
@@ -93,7 +92,6 @@ func OpenGroup(ctx context.Context, logger log.Logger, headPath string, groupOpt
g := &Group{
logger: logger,
ctx: ctx,
ID: "group:" + head.ID,
Head: head,
headBuf: bufio.NewWriterSize(head, 4096*10),
@@ -250,14 +248,14 @@ func (g *Group) processTicks(ctx context.Context) {
case <-ctx.Done():
return
case <-g.ticker.C:
g.checkHeadSizeLimit()
g.checkTotalSizeLimit()
g.checkHeadSizeLimit(ctx)
g.checkTotalSizeLimit(ctx)
}
}
}
// NOTE: this function is called manually in tests.
func (g *Group) checkHeadSizeLimit() {
func (g *Group) checkHeadSizeLimit(ctx context.Context) {
limit := g.HeadSizeLimit()
if limit == 0 {
return
@@ -268,13 +266,15 @@ func (g *Group) checkHeadSizeLimit() {
return
}
if size >= limit {
g.RotateFile()
g.rotateFile(ctx)
}
}
func (g *Group) checkTotalSizeLimit() {
limit := g.TotalSizeLimit()
if limit == 0 {
func (g *Group) checkTotalSizeLimit(ctx context.Context) {
g.mtx.Lock()
defer g.mtx.Unlock()
if g.totalSizeLimit == 0 {
return
}
@@ -282,7 +282,7 @@ func (g *Group) checkTotalSizeLimit() {
totalSize := gInfo.TotalSize
for i := 0; i < maxFilesToRemove; i++ {
index := gInfo.MinIndex + i
if totalSize < limit {
if totalSize < g.totalSizeLimit {
return
}
if index == gInfo.MaxIndex {
@@ -296,8 +296,12 @@ func (g *Group) checkTotalSizeLimit() {
g.logger.Error("Failed to fetch info for file", "file", pathToRemove)
continue
}
err = os.Remove(pathToRemove)
if err != nil {
if ctx.Err() != nil {
return
}
if err = os.Remove(pathToRemove); err != nil {
g.logger.Error("Failed to remove path", "path", pathToRemove)
return
}
@@ -305,8 +309,8 @@ func (g *Group) checkTotalSizeLimit() {
}
}
// RotateFile causes group to close the current head and assign it some index.
func (g *Group) RotateFile() {
// rotateFile causes group to close the current head and assign it some index.
func (g *Group) rotateFile(ctx context.Context) {
g.mtx.Lock()
defer g.mtx.Unlock()
@@ -319,6 +323,10 @@ func (g *Group) RotateFile() {
panic(err)
}
err := g.Head.withLock(func() error {
if err := ctx.Err(); err != nil {
return err
}
if err := g.Head.unsyncCloseFile(); err != nil {
return err
}
@@ -326,9 +334,13 @@ func (g *Group) RotateFile() {
indexPath := filePathForIndex(headPath, g.maxIndex, g.maxIndex+1)
return os.Rename(headPath, indexPath)
})
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
return
}
if err != nil {
panic(err)
}
g.maxIndex++
}
internal/libs/autofile/group_test.go
+9 -9
View File
@@ -64,7 +64,7 @@ func TestCheckHeadSizeLimit(t *testing.T) {
assertGroupInfo(t, g.ReadGroupInfo(), 0, 0, 999000, 999000)
// Even calling checkHeadSizeLimit manually won't rotate it.
g.checkHeadSizeLimit()
g.checkHeadSizeLimit(ctx)
assertGroupInfo(t, g.ReadGroupInfo(), 0, 0, 999000, 999000)
// Write 1000 more bytes.
@@ -74,7 +74,7 @@ func TestCheckHeadSizeLimit(t *testing.T) {
require.NoError(t, err)
// Calling checkHeadSizeLimit this time rolls it.
g.checkHeadSizeLimit()
g.checkHeadSizeLimit(ctx)
assertGroupInfo(t, g.ReadGroupInfo(), 0, 1, 1000000, 0)
// Write 1000 more bytes.
@@ -84,7 +84,7 @@ func TestCheckHeadSizeLimit(t *testing.T) {
require.NoError(t, err)
// Calling checkHeadSizeLimit does nothing.
g.checkHeadSizeLimit()
g.checkHeadSizeLimit(ctx)
assertGroupInfo(t, g.ReadGroupInfo(), 0, 1, 1001000, 1000)
// Write 1000 bytes 999 times.
@@ -97,7 +97,7 @@ func TestCheckHeadSizeLimit(t *testing.T) {
assertGroupInfo(t, g.ReadGroupInfo(), 0, 1, 2000000, 1000000)
// Calling checkHeadSizeLimit rolls it again.
g.checkHeadSizeLimit()
g.checkHeadSizeLimit(ctx)
assertGroupInfo(t, g.ReadGroupInfo(), 0, 2, 2000000, 0)
// Write 1000 more bytes.
@@ -108,7 +108,7 @@ func TestCheckHeadSizeLimit(t *testing.T) {
assertGroupInfo(t, g.ReadGroupInfo(), 0, 2, 2001000, 1000)
// Calling checkHeadSizeLimit does nothing.
g.checkHeadSizeLimit()
g.checkHeadSizeLimit(ctx)
assertGroupInfo(t, g.ReadGroupInfo(), 0, 2, 2001000, 1000)
// Cleanup
@@ -150,7 +150,7 @@ func TestRotateFile(t *testing.T) {
require.NoError(t, err)
err = g.FlushAndSync()
require.NoError(t, err)
g.RotateFile()
g.rotateFile(ctx)
err = g.WriteLine("Line 4")
require.NoError(t, err)
err = g.WriteLine("Line 5")
@@ -224,7 +224,7 @@ func TestGroupReaderRead(t *testing.T) {
require.NoError(t, err)
err = g.FlushAndSync()
require.NoError(t, err)
g.RotateFile()
g.rotateFile(ctx)
frankenstein := []byte("Frankenstein's Monster")
_, err = g.Write(frankenstein)
require.NoError(t, err)
@@ -262,7 +262,7 @@ func TestGroupReaderRead2(t *testing.T) {
require.NoError(t, err)
err = g.FlushAndSync()
require.NoError(t, err)
g.RotateFile()
g.rotateFile(ctx)
frankenstein := []byte("Frankenstein's Monster")
frankensteinPart := []byte("Frankenstein")
_, err = g.Write(frankensteinPart) // note writing only a part
@@ -315,7 +315,7 @@ func TestMaxIndex(t *testing.T) {
require.NoError(t, err)
err = g.FlushAndSync()
require.NoError(t, err)
g.RotateFile()
g.rotateFile(ctx)
assert.Equal(t, 1, g.MaxIndex(), "MaxIndex should point to the last file")
internal/mempool/mempool.go
+1 -1
View File
@@ -176,7 +176,7 @@ func (txmp *TxMempool) SizeBytes() int64 {
//
// NOTE: The caller must obtain a write-lock prior to execution.
func (txmp *TxMempool) FlushAppConn(ctx context.Context) error {
return txmp.proxyAppConn.FlushSync(ctx)
return txmp.proxyAppConn.Flush(ctx)
}
// WaitForNextTx returns a blocking channel that will be closed when the next
internal/mempool/reactor_test.go
+3 -1
View File
@@ -45,7 +45,7 @@ func setupReactors(ctx context.Context, t *testing.T, numNodes int, chBuf uint)
t.Cleanup(func() { os.RemoveAll(cfg.RootDir) })
rts := &reactorTestSuite{
logger: log.TestingLogger().With("testCase", t.Name()),
logger: log.NewNopLogger().With("testCase", t.Name()),
network: p2ptest.MakeNetwork(ctx, t, p2ptest.NetworkOptions{NumNodes: numNodes}),
reactors: make(map[types.NodeID]*Reactor, numNodes),
mempoolChannels: make(map[types.NodeID]*p2p.Channel, numNodes),
@@ -95,8 +95,10 @@ func setupReactors(ctx context.Context, t *testing.T, numNodes int, chBuf uint)
for nodeID := range rts.reactors {
if rts.reactors[nodeID].IsRunning() {
require.NoError(t, rts.reactors[nodeID].Stop())
rts.reactors[nodeID].Wait()
require.False(t, rts.reactors[nodeID].IsRunning())
}
}
})
internal/p2p/peermanager.go
+11
View File
@@ -136,6 +136,10 @@ type PeerManagerOptions struct {
// consider private and never gossip.
PrivatePeers map[types.NodeID]struct{}
// SelfAddress is the address that will be advertised to peers for them to dial back to us.
// If Hostname and Port are unset, Advertise() will include no self-announcement
SelfAddress NodeAddress
// persistentPeers provides fast PersistentPeers lookups. It is built
// by optimize().
persistentPeers map[types.NodeID]bool
@@ -791,6 +795,13 @@ func (m *PeerManager) Advertise(peerID types.NodeID, limit uint16) []NodeAddress
defer m.mtx.Unlock()
addresses := make([]NodeAddress, 0, limit)
// advertise ourselves, to let everyone know how to dial us back
// and enable mutual address discovery
if m.options.SelfAddress.Hostname != "" && m.options.SelfAddress.Port != 0 {
addresses = append(addresses, m.options.SelfAddress)
}
for _, peer := range m.store.Ranked() {
if peer.ID == peerID {
continue
internal/p2p/peermanager_test.go
+17
View File
@@ -1828,6 +1828,23 @@ func TestPeerManager_Advertise(t *testing.T) {
}, peerManager.Advertise(dID, 2))
}
func TestPeerManager_Advertise_Self(t *testing.T) {
dID := types.NodeID(strings.Repeat("d", 40))
self := p2p.NodeAddress{Protocol: "tcp", NodeID: selfID, Hostname: "2001:db8::1", Port: 26657}
// Create a peer manager with SelfAddress defined.
peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
SelfAddress: self,
})
require.NoError(t, err)
// peer manager should always advertise its SelfAddress.
require.ElementsMatch(t, []p2p.NodeAddress{
self,
}, peerManager.Advertise(dID, 100))
}
func TestPeerManager_SetHeight_GetHeight(t *testing.T) {
a := p2p.NodeAddress{Protocol: "memory", NodeID: types.NodeID(strings.Repeat("a", 40))}
b := p2p.NodeAddress{Protocol: "memory", NodeID: types.NodeID(strings.Repeat("b", 40))}
internal/p2p/pex/reactor_test.go
+3
View File
@@ -1,3 +1,6 @@
// Temporarily disabled pending ttps://github.com/tendermint/tendermint/issues/7626.
//go:build issue7626
package pex_test
import (
internal/proxy/app_conn.go
+44 -44
View File
@@ -18,12 +18,12 @@ type AppConnConsensus interface {
SetResponseCallback(abciclient.Callback)
Error() error
InitChainSync(context.Context, types.RequestInitChain) (*types.ResponseInitChain, error)
InitChain(context.Context, types.RequestInitChain) (*types.ResponseInitChain, error)
BeginBlockSync(context.Context, types.RequestBeginBlock) (*types.ResponseBeginBlock, error)
DeliverTxAsync(context.Context, types.RequestDeliverTx) (*abciclient.ReqRes, error)
EndBlockSync(context.Context, types.RequestEndBlock) (*types.ResponseEndBlock, error)
CommitSync(context.Context) (*types.ResponseCommit, 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)
Commit(context.Context) (*types.ResponseCommit, error)
}
type AppConnMempool interface {
@@ -31,27 +31,27 @@ type AppConnMempool interface {
Error() error
CheckTxAsync(context.Context, types.RequestCheckTx) (*abciclient.ReqRes, error)
CheckTxSync(context.Context, types.RequestCheckTx) (*types.ResponseCheckTx, error)
CheckTx(context.Context, types.RequestCheckTx) (*types.ResponseCheckTx, error)
FlushAsync(context.Context) (*abciclient.ReqRes, error)
FlushSync(context.Context) error
Flush(context.Context) error
}
type AppConnQuery interface {
Error() error
EchoSync(context.Context, string) (*types.ResponseEcho, error)
InfoSync(context.Context, types.RequestInfo) (*types.ResponseInfo, error)
QuerySync(context.Context, types.RequestQuery) (*types.ResponseQuery, error)
Echo(context.Context, string) (*types.ResponseEcho, error)
Info(context.Context, types.RequestInfo) (*types.ResponseInfo, error)
Query(context.Context, types.RequestQuery) (*types.ResponseQuery, error)
}
type AppConnSnapshot interface {
Error() error
ListSnapshotsSync(context.Context, types.RequestListSnapshots) (*types.ResponseListSnapshots, error)
OfferSnapshotSync(context.Context, types.RequestOfferSnapshot) (*types.ResponseOfferSnapshot, error)
LoadSnapshotChunkSync(context.Context, types.RequestLoadSnapshotChunk) (*types.ResponseLoadSnapshotChunk, error)
ApplySnapshotChunkSync(context.Context, types.RequestApplySnapshotChunk) (*types.ResponseApplySnapshotChunk, 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)
ApplySnapshotChunk(context.Context, types.RequestApplySnapshotChunk) (*types.ResponseApplySnapshotChunk, error)
}
//-----------------------------------------------------------------------------------------
@@ -77,41 +77,41 @@ func (app *appConnConsensus) Error() error {
return app.appConn.Error()
}
func (app *appConnConsensus) InitChainSync(
func (app *appConnConsensus) InitChain(
ctx context.Context,
req types.RequestInitChain,
) (*types.ResponseInitChain, error) {
defer addTimeSample(app.metrics.MethodTiming.With("method", "init_chain", "type", "sync"))()
return app.appConn.InitChainSync(ctx, req)
return app.appConn.InitChain(ctx, req)
}
func (app *appConnConsensus) BeginBlockSync(
func (app *appConnConsensus) BeginBlock(
ctx context.Context,
req types.RequestBeginBlock,
) (*types.ResponseBeginBlock, error) {
defer addTimeSample(app.metrics.MethodTiming.With("method", "begin_block", "type", "sync"))()
return app.appConn.BeginBlockSync(ctx, req)
return app.appConn.BeginBlock(ctx, req)
}
func (app *appConnConsensus) DeliverTxAsync(
func (app *appConnConsensus) DeliverTx(
ctx context.Context,
req types.RequestDeliverTx,
) (*abciclient.ReqRes, error) {
defer addTimeSample(app.metrics.MethodTiming.With("method", "deliver_tx", "type", "async"))()
return app.appConn.DeliverTxAsync(ctx, req)
) (*types.ResponseDeliverTx, error) {
defer addTimeSample(app.metrics.MethodTiming.With("method", "deliver_tx", "type", "sync"))()
return app.appConn.DeliverTx(ctx, req)
}
func (app *appConnConsensus) EndBlockSync(
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.EndBlockSync(ctx, req)
return app.appConn.EndBlock(ctx, req)
}
func (app *appConnConsensus) CommitSync(ctx context.Context) (*types.ResponseCommit, error) {
func (app *appConnConsensus) Commit(ctx context.Context) (*types.ResponseCommit, error) {
defer addTimeSample(app.metrics.MethodTiming.With("method", "commit", "type", "sync"))()
return app.appConn.CommitSync(ctx)
return app.appConn.Commit(ctx)
}
//------------------------------------------------
@@ -142,9 +142,9 @@ func (app *appConnMempool) FlushAsync(ctx context.Context) (*abciclient.ReqRes,
return app.appConn.FlushAsync(ctx)
}
func (app *appConnMempool) FlushSync(ctx context.Context) error {
func (app *appConnMempool) Flush(ctx context.Context) error {
defer addTimeSample(app.metrics.MethodTiming.With("method", "flush", "type", "sync"))()
return app.appConn.FlushSync(ctx)
return app.appConn.Flush(ctx)
}
func (app *appConnMempool) CheckTxAsync(ctx context.Context, req types.RequestCheckTx) (*abciclient.ReqRes, error) {
@@ -152,9 +152,9 @@ func (app *appConnMempool) CheckTxAsync(ctx context.Context, req types.RequestCh
return app.appConn.CheckTxAsync(ctx, req)
}
func (app *appConnMempool) CheckTxSync(ctx context.Context, req types.RequestCheckTx) (*types.ResponseCheckTx, error) {
func (app *appConnMempool) CheckTx(ctx context.Context, req types.RequestCheckTx) (*types.ResponseCheckTx, error) {
defer addTimeSample(app.metrics.MethodTiming.With("method", "check_tx", "type", "sync"))()
return app.appConn.CheckTxSync(ctx, req)
return app.appConn.CheckTx(ctx, req)
}
//------------------------------------------------
@@ -176,19 +176,19 @@ func (app *appConnQuery) Error() error {
return app.appConn.Error()
}
func (app *appConnQuery) EchoSync(ctx context.Context, msg string) (*types.ResponseEcho, error) {
func (app *appConnQuery) Echo(ctx context.Context, msg string) (*types.ResponseEcho, error) {
defer addTimeSample(app.metrics.MethodTiming.With("method", "echo", "type", "sync"))()
return app.appConn.EchoSync(ctx, msg)
return app.appConn.Echo(ctx, msg)
}
func (app *appConnQuery) InfoSync(ctx context.Context, req types.RequestInfo) (*types.ResponseInfo, error) {
func (app *appConnQuery) Info(ctx context.Context, req types.RequestInfo) (*types.ResponseInfo, error) {
defer addTimeSample(app.metrics.MethodTiming.With("method", "info", "type", "sync"))()
return app.appConn.InfoSync(ctx, req)
return app.appConn.Info(ctx, req)
}
func (app *appConnQuery) QuerySync(ctx context.Context, reqQuery types.RequestQuery) (*types.ResponseQuery, error) {
func (app *appConnQuery) Query(ctx context.Context, reqQuery types.RequestQuery) (*types.ResponseQuery, error) {
defer addTimeSample(app.metrics.MethodTiming.With("method", "query", "type", "sync"))()
return app.appConn.QuerySync(ctx, reqQuery)
return app.appConn.Query(ctx, reqQuery)
}
//------------------------------------------------
@@ -210,34 +210,34 @@ func (app *appConnSnapshot) Error() error {
return app.appConn.Error()
}
func (app *appConnSnapshot) ListSnapshotsSync(
func (app *appConnSnapshot) ListSnapshots(
ctx context.Context,
req types.RequestListSnapshots,
) (*types.ResponseListSnapshots, error) {
defer addTimeSample(app.metrics.MethodTiming.With("method", "list_snapshots", "type", "sync"))()
return app.appConn.ListSnapshotsSync(ctx, req)
return app.appConn.ListSnapshots(ctx, req)
}
func (app *appConnSnapshot) OfferSnapshotSync(
func (app *appConnSnapshot) OfferSnapshot(
ctx context.Context,
req types.RequestOfferSnapshot,
) (*types.ResponseOfferSnapshot, error) {
defer addTimeSample(app.metrics.MethodTiming.With("method", "offer_snapshot", "type", "sync"))()
return app.appConn.OfferSnapshotSync(ctx, req)
return app.appConn.OfferSnapshot(ctx, req)
}
func (app *appConnSnapshot) LoadSnapshotChunkSync(
func (app *appConnSnapshot) LoadSnapshotChunk(
ctx context.Context,
req types.RequestLoadSnapshotChunk) (*types.ResponseLoadSnapshotChunk, error) {
defer addTimeSample(app.metrics.MethodTiming.With("method", "load_snapshot_chunk", "type", "sync"))()
return app.appConn.LoadSnapshotChunkSync(ctx, req)
return app.appConn.LoadSnapshotChunk(ctx, req)
}
func (app *appConnSnapshot) ApplySnapshotChunkSync(
func (app *appConnSnapshot) ApplySnapshotChunk(
ctx context.Context,
req types.RequestApplySnapshotChunk) (*types.ResponseApplySnapshotChunk, error) {
defer addTimeSample(app.metrics.MethodTiming.With("method", "apply_snapshot_chunk", "type", "sync"))()
return app.appConn.ApplySnapshotChunkSync(ctx, req)
return app.appConn.ApplySnapshotChunk(ctx, req)
}
// addTimeSample returns a function that, when called, adds an observation to m.
internal/proxy/app_conn_test.go
+17 -17
View File
@@ -18,9 +18,9 @@ import (
//----------------------------------------
type appConnTestI interface {
EchoAsync(ctx context.Context, msg string) (*abciclient.ReqRes, error)
FlushSync(context.Context) error
InfoSync(context.Context, types.RequestInfo) (*types.ResponseInfo, error)
Echo(context.Context, string) (*types.ResponseEcho, error)
Flush(context.Context) error
Info(context.Context, types.RequestInfo) (*types.ResponseInfo, error)
}
type appConnTest struct {
@@ -31,16 +31,16 @@ func newAppConnTest(appConn abciclient.Client) appConnTestI {
return &appConnTest{appConn}
}
func (app *appConnTest) EchoAsync(ctx context.Context, msg string) (*abciclient.ReqRes, error) {
return app.appConn.EchoAsync(ctx, msg)
func (app *appConnTest) Echo(ctx context.Context, msg string) (*types.ResponseEcho, error) {
return app.appConn.Echo(ctx, msg)
}
func (app *appConnTest) FlushSync(ctx context.Context) error {
return app.appConn.FlushSync(ctx)
func (app *appConnTest) Flush(ctx context.Context) error {
return app.appConn.Flush(ctx)
}
func (app *appConnTest) InfoSync(ctx context.Context, req types.RequestInfo) (*types.ResponseInfo, error) {
return app.appConn.InfoSync(ctx, req)
func (app *appConnTest) Info(ctx context.Context, req types.RequestInfo) (*types.ResponseInfo, error) {
return app.appConn.Info(ctx, req)
}
//----------------------------------------
@@ -70,18 +70,18 @@ func TestEcho(t *testing.T) {
t.Log("Connected")
for i := 0; i < 1000; i++ {
_, err = proxy.EchoAsync(ctx, fmt.Sprintf("echo-%v", i))
_, err = proxy.Echo(ctx, fmt.Sprintf("echo-%v", i))
if err != nil {
t.Error(err)
}
// flush sometimes
if i%128 == 0 {
if err := proxy.FlushSync(ctx); err != nil {
if err := proxy.Flush(ctx); err != nil {
t.Error(err)
}
}
}
if err := proxy.FlushSync(ctx); err != nil {
if err := proxy.Flush(ctx); err != nil {
t.Error(err)
}
}
@@ -112,23 +112,23 @@ func BenchmarkEcho(b *testing.B) {
b.StartTimer() // Start benchmarking tests
for i := 0; i < b.N; i++ {
_, err = proxy.EchoAsync(ctx, echoString)
_, err = proxy.Echo(ctx, echoString)
if err != nil {
b.Error(err)
}
// flush sometimes
if i%128 == 0 {
if err := proxy.FlushSync(ctx); err != nil {
if err := proxy.Flush(ctx); err != nil {
b.Error(err)
}
}
}
if err := proxy.FlushSync(ctx); err != nil {
if err := proxy.Flush(ctx); err != nil {
b.Error(err)
}
b.StopTimer()
// info := proxy.InfoSync(types.RequestInfo{""})
// info := proxy.Info(types.RequestInfo{""})
// b.Log("N: ", b.N, info)
}
@@ -154,7 +154,7 @@ func TestInfo(t *testing.T) {
proxy := newAppConnTest(cli)
t.Log("Connected")
resInfo, err := proxy.InfoSync(ctx, RequestInfo)
resInfo, err := proxy.Info(ctx, RequestInfo)
require.NoError(t, err)
if resInfo.Data != "{\"size\":0}" {
internal/proxy/mocks/app_conn_consensus.go
+13 -13
View File
@@ -17,8 +17,8 @@ type AppConnConsensus struct {
mock.Mock
}
// BeginBlockSync provides a mock function with given fields: _a0, _a1
func (_m *AppConnConsensus) BeginBlockSync(_a0 context.Context, _a1 types.RequestBeginBlock) (*types.ResponseBeginBlock, error) {
// 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
@@ -40,8 +40,8 @@ func (_m *AppConnConsensus) BeginBlockSync(_a0 context.Context, _a1 types.Reques
return r0, r1
}
// CommitSync provides a mock function with given fields: _a0
func (_m *AppConnConsensus) CommitSync(_a0 context.Context) (*types.ResponseCommit, error) {
// Commit provides a mock function with given fields: _a0
func (_m *AppConnConsensus) Commit(_a0 context.Context) (*types.ResponseCommit, error) {
ret := _m.Called(_a0)
var r0 *types.ResponseCommit
@@ -63,16 +63,16 @@ func (_m *AppConnConsensus) CommitSync(_a0 context.Context) (*types.ResponseComm
return r0, r1
}
// DeliverTxAsync provides a mock function with given fields: _a0, _a1
func (_m *AppConnConsensus) DeliverTxAsync(_a0 context.Context, _a1 types.RequestDeliverTx) (*abciclient.ReqRes, error) {
// 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 *abciclient.ReqRes
if rf, ok := ret.Get(0).(func(context.Context, types.RequestDeliverTx) *abciclient.ReqRes); ok {
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).(*abciclient.ReqRes)
r0 = ret.Get(0).(*types.ResponseDeliverTx)
}
}
@@ -86,8 +86,8 @@ func (_m *AppConnConsensus) DeliverTxAsync(_a0 context.Context, _a1 types.Reques
return r0, r1
}
// EndBlockSync provides a mock function with given fields: _a0, _a1
func (_m *AppConnConsensus) EndBlockSync(_a0 context.Context, _a1 types.RequestEndBlock) (*types.ResponseEndBlock, error) {
// 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
@@ -123,8 +123,8 @@ func (_m *AppConnConsensus) Error() error {
return r0
}
// InitChainSync provides a mock function with given fields: _a0, _a1
func (_m *AppConnConsensus) InitChainSync(_a0 context.Context, _a1 types.RequestInitChain) (*types.ResponseInitChain, error) {
// 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)
var r0 *types.ResponseInitChain
internal/proxy/mocks/app_conn_mempool.go
+37 -37
View File
@@ -17,6 +17,29 @@ type AppConnMempool struct {
mock.Mock
}
// CheckTx provides a mock function with given fields: _a0, _a1
func (_m *AppConnMempool) CheckTx(_a0 context.Context, _a1 types.RequestCheckTx) (*types.ResponseCheckTx, error) {
ret := _m.Called(_a0, _a1)
var r0 *types.ResponseCheckTx
if rf, ok := ret.Get(0).(func(context.Context, types.RequestCheckTx) *types.ResponseCheckTx); ok {
r0 = rf(_a0, _a1)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*types.ResponseCheckTx)
}
}
var r1 error
if rf, ok := ret.Get(1).(func(context.Context, types.RequestCheckTx) error); ok {
r1 = rf(_a0, _a1)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// CheckTxAsync provides a mock function with given fields: _a0, _a1
func (_m *AppConnMempool) CheckTxAsync(_a0 context.Context, _a1 types.RequestCheckTx) (*abciclient.ReqRes, error) {
ret := _m.Called(_a0, _a1)
@@ -40,29 +63,6 @@ func (_m *AppConnMempool) CheckTxAsync(_a0 context.Context, _a1 types.RequestChe
return r0, r1
}
// CheckTxSync provides a mock function with given fields: _a0, _a1
func (_m *AppConnMempool) CheckTxSync(_a0 context.Context, _a1 types.RequestCheckTx) (*types.ResponseCheckTx, error) {
ret := _m.Called(_a0, _a1)
var r0 *types.ResponseCheckTx
if rf, ok := ret.Get(0).(func(context.Context, types.RequestCheckTx) *types.ResponseCheckTx); ok {
r0 = rf(_a0, _a1)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*types.ResponseCheckTx)
}
}
var r1 error
if rf, ok := ret.Get(1).(func(context.Context, types.RequestCheckTx) error); ok {
r1 = rf(_a0, _a1)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// Error provides a mock function with given fields:
func (_m *AppConnMempool) Error() error {
ret := _m.Called()
@@ -77,6 +77,20 @@ func (_m *AppConnMempool) Error() error {
return r0
}
// Flush provides a mock function with given fields: _a0
func (_m *AppConnMempool) Flush(_a0 context.Context) error {
ret := _m.Called(_a0)
var r0 error
if rf, ok := ret.Get(0).(func(context.Context) error); ok {
r0 = rf(_a0)
} else {
r0 = ret.Error(0)
}
return r0
}
// FlushAsync provides a mock function with given fields: _a0
func (_m *AppConnMempool) FlushAsync(_a0 context.Context) (*abciclient.ReqRes, error) {
ret := _m.Called(_a0)
@@ -100,20 +114,6 @@ func (_m *AppConnMempool) FlushAsync(_a0 context.Context) (*abciclient.ReqRes, e
return r0, r1
}
// FlushSync provides a mock function with given fields: _a0
func (_m *AppConnMempool) FlushSync(_a0 context.Context) error {
ret := _m.Called(_a0)
var r0 error
if rf, ok := ret.Get(0).(func(context.Context) error); ok {
r0 = rf(_a0)
} else {
r0 = ret.Error(0)
}
return r0
}
// SetResponseCallback provides a mock function with given fields: _a0
func (_m *AppConnMempool) SetResponseCallback(_a0 abciclient.Callback) {
_m.Called(_a0)
internal/proxy/mocks/app_conn_query.go
+6 -6
View File
@@ -15,8 +15,8 @@ type AppConnQuery struct {
mock.Mock
}
// EchoSync provides a mock function with given fields: _a0, _a1
func (_m *AppConnQuery) EchoSync(_a0 context.Context, _a1 string) (*types.ResponseEcho, error) {
// Echo provides a mock function with given fields: _a0, _a1
func (_m *AppConnQuery) Echo(_a0 context.Context, _a1 string) (*types.ResponseEcho, error) {
ret := _m.Called(_a0, _a1)
var r0 *types.ResponseEcho
@@ -52,8 +52,8 @@ func (_m *AppConnQuery) Error() error {
return r0
}
// InfoSync provides a mock function with given fields: _a0, _a1
func (_m *AppConnQuery) InfoSync(_a0 context.Context, _a1 types.RequestInfo) (*types.ResponseInfo, error) {
// Info provides a mock function with given fields: _a0, _a1
func (_m *AppConnQuery) Info(_a0 context.Context, _a1 types.RequestInfo) (*types.ResponseInfo, error) {
ret := _m.Called(_a0, _a1)
var r0 *types.ResponseInfo
@@ -75,8 +75,8 @@ func (_m *AppConnQuery) InfoSync(_a0 context.Context, _a1 types.RequestInfo) (*t
return r0, r1
}
// QuerySync provides a mock function with given fields: _a0, _a1
func (_m *AppConnQuery) QuerySync(_a0 context.Context, _a1 types.RequestQuery) (*types.ResponseQuery, error) {
// Query provides a mock function with given fields: _a0, _a1
func (_m *AppConnQuery) Query(_a0 context.Context, _a1 types.RequestQuery) (*types.ResponseQuery, error) {
ret := _m.Called(_a0, _a1)
var r0 *types.ResponseQuery
internal/proxy/mocks/app_conn_snapshot.go
+8 -8
View File
@@ -15,8 +15,8 @@ type AppConnSnapshot struct {
mock.Mock
}
// ApplySnapshotChunkSync provides a mock function with given fields: _a0, _a1
func (_m *AppConnSnapshot) ApplySnapshotChunkSync(_a0 context.Context, _a1 types.RequestApplySnapshotChunk) (*types.ResponseApplySnapshotChunk, error) {
// ApplySnapshotChunk provides a mock function with given fields: _a0, _a1
func (_m *AppConnSnapshot) ApplySnapshotChunk(_a0 context.Context, _a1 types.RequestApplySnapshotChunk) (*types.ResponseApplySnapshotChunk, error) {
ret := _m.Called(_a0, _a1)
var r0 *types.ResponseApplySnapshotChunk
@@ -52,8 +52,8 @@ func (_m *AppConnSnapshot) Error() error {
return r0
}
// ListSnapshotsSync provides a mock function with given fields: _a0, _a1
func (_m *AppConnSnapshot) ListSnapshotsSync(_a0 context.Context, _a1 types.RequestListSnapshots) (*types.ResponseListSnapshots, error) {
// ListSnapshots provides a mock function with given fields: _a0, _a1
func (_m *AppConnSnapshot) ListSnapshots(_a0 context.Context, _a1 types.RequestListSnapshots) (*types.ResponseListSnapshots, error) {
ret := _m.Called(_a0, _a1)
var r0 *types.ResponseListSnapshots
@@ -75,8 +75,8 @@ func (_m *AppConnSnapshot) ListSnapshotsSync(_a0 context.Context, _a1 types.Requ
return r0, r1
}
// LoadSnapshotChunkSync provides a mock function with given fields: _a0, _a1
func (_m *AppConnSnapshot) LoadSnapshotChunkSync(_a0 context.Context, _a1 types.RequestLoadSnapshotChunk) (*types.ResponseLoadSnapshotChunk, error) {
// LoadSnapshotChunk provides a mock function with given fields: _a0, _a1
func (_m *AppConnSnapshot) LoadSnapshotChunk(_a0 context.Context, _a1 types.RequestLoadSnapshotChunk) (*types.ResponseLoadSnapshotChunk, error) {
ret := _m.Called(_a0, _a1)
var r0 *types.ResponseLoadSnapshotChunk
@@ -98,8 +98,8 @@ func (_m *AppConnSnapshot) LoadSnapshotChunkSync(_a0 context.Context, _a1 types.
return r0, r1
}
// OfferSnapshotSync provides a mock function with given fields: _a0, _a1
func (_m *AppConnSnapshot) OfferSnapshotSync(_a0 context.Context, _a1 types.RequestOfferSnapshot) (*types.ResponseOfferSnapshot, error) {
// OfferSnapshot provides a mock function with given fields: _a0, _a1
func (_m *AppConnSnapshot) OfferSnapshot(_a0 context.Context, _a1 types.RequestOfferSnapshot) (*types.ResponseOfferSnapshot, error) {
ret := _m.Called(_a0, _a1)
var r0 *types.ResponseOfferSnapshot
internal/rpc/core/abci.go
+2 -2
View File
@@ -18,7 +18,7 @@ func (env *Environment) ABCIQuery(
height int64,
prove bool,
) (*coretypes.ResultABCIQuery, error) {
resQuery, err := env.ProxyAppQuery.QuerySync(ctx, abci.RequestQuery{
resQuery, err := env.ProxyAppQuery.Query(ctx, abci.RequestQuery{
Path: path,
Data: data,
Height: height,
@@ -34,7 +34,7 @@ func (env *Environment) ABCIQuery(
// ABCIInfo gets some info about the application.
// More: https://docs.tendermint.com/master/rpc/#/ABCI/abci_info
func (env *Environment) ABCIInfo(ctx context.Context) (*coretypes.ResultABCIInfo, error) {
resInfo, err := env.ProxyAppQuery.InfoSync(ctx, proxy.RequestInfo)
resInfo, err := env.ProxyAppQuery.Info(ctx, proxy.RequestInfo)
if err != nil {
return nil, err
}
internal/rpc/core/mempool.go
+1 -1
View File
@@ -153,7 +153,7 @@ func (env *Environment) NumUnconfirmedTxs(ctx context.Context) (*coretypes.Resul
// be added to the mempool either.
// More: https://docs.tendermint.com/master/rpc/#/Tx/check_tx
func (env *Environment) CheckTx(ctx context.Context, tx types.Tx) (*coretypes.ResultCheckTx, error) {
res, err := env.ProxyAppMempool.CheckTxSync(ctx, abci.RequestCheckTx{Tx: tx})
res, err := env.ProxyAppMempool.CheckTx(ctx, abci.RequestCheckTx{Tx: tx})
if err != nil {
return nil, err
}
internal/state/execution.go
+16 -27
View File
@@ -255,9 +255,9 @@ func (blockExec *BlockExecutor) Commit(
}
// Commit block, get hash back
res, err := blockExec.proxyApp.CommitSync(ctx)
res, err := blockExec.proxyApp.Commit(ctx)
if err != nil {
blockExec.logger.Error("client error during proxyAppConn.CommitSync", "err", err)
blockExec.logger.Error("client error during proxyAppConn.Commit", "err", err)
return nil, 0, err
}
@@ -302,26 +302,6 @@ func execBlockOnProxyApp(
dtxs := make([]*abci.ResponseDeliverTx, len(block.Txs))
abciResponses.DeliverTxs = dtxs
// Execute transactions and get hash.
proxyCb := func(req *abci.Request, res *abci.Response) {
if r, ok := res.Value.(*abci.Response_DeliverTx); ok {
// TODO: make use of res.Log
// TODO: make use of this info
// Blocks may include invalid txs.
txRes := r.DeliverTx
if txRes.Code == abci.CodeTypeOK {
validTxs++
} else {
logger.Debug("invalid tx", "code", txRes.Code, "log", txRes.Log)
invalidTxs++
}
abciResponses.DeliverTxs[txIndex] = txRes
txIndex++
}
}
proxyAppConn.SetResponseCallback(proxyCb)
commitInfo := getBeginBlockValidatorInfo(block, store, initialHeight)
byzVals := make([]abci.Evidence, 0)
@@ -336,7 +316,7 @@ func execBlockOnProxyApp(
return nil, errors.New("nil header")
}
abciResponses.BeginBlock, err = proxyAppConn.BeginBlockSync(
abciResponses.BeginBlock, err = proxyAppConn.BeginBlock(
ctx,
abci.RequestBeginBlock{
Hash: block.Hash(),
@@ -352,13 +332,22 @@ func execBlockOnProxyApp(
// run txs of block
for _, tx := range block.Txs {
_, err = proxyAppConn.DeliverTxAsync(ctx, abci.RequestDeliverTx{Tx: tx})
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.EndBlockSync(ctx, abci.RequestEndBlock{Height: block.Height})
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
@@ -604,9 +593,9 @@ func ExecCommitBlock(
}
// Commit block, get hash back
res, err := appConnConsensus.CommitSync(ctx)
res, err := appConnConsensus.Commit(ctx)
if err != nil {
logger.Error("client error during proxyAppConn.CommitSync", "err", res)
logger.Error("client error during proxyAppConn.Commit", "err", res)
return nil, err
}
internal/statesync/dispatcher.go
+3
View File
@@ -224,6 +224,9 @@ func (p *BlockProvider) ReportEvidence(ctx context.Context, ev types.Evidence) e
// String implements stringer interface
func (p *BlockProvider) String() string { return string(p.peer) }
// Returns the ID address of the provider (NodeID of peer)
func (p *BlockProvider) ID() string { return string(p.peer) }
//----------------------------------------------------------------
// peerList is a rolling list of peers. This is used to distribute the load of
internal/statesync/reactor.go
+2 -2
View File
@@ -618,7 +618,7 @@ func (r *Reactor) handleChunkMessage(ctx context.Context, envelope *p2p.Envelope
"chunk", msg.Index,
"peer", envelope.From,
)
resp, err := r.conn.LoadSnapshotChunkSync(ctx, abci.RequestLoadSnapshotChunk{
resp, err := r.conn.LoadSnapshotChunk(ctx, abci.RequestLoadSnapshotChunk{
Height: msg.Height,
Format: msg.Format,
Chunk: msg.Index,
@@ -911,7 +911,7 @@ func (r *Reactor) processPeerUpdates(ctx context.Context) {
// recentSnapshots fetches the n most recent snapshots from the app
func (r *Reactor) recentSnapshots(ctx context.Context, n uint32) ([]*snapshot, error) {
resp, err := r.conn.ListSnapshotsSync(ctx, abci.RequestListSnapshots{})
resp, err := r.conn.ListSnapshots(ctx, abci.RequestListSnapshots{})
if err != nil {
return nil, err
}
internal/statesync/reactor_test.go
+5 -5
View File
@@ -214,15 +214,15 @@ func TestReactor_Sync(t *testing.T) {
rts := setup(ctx, t, nil, nil, nil, 2)
chain := buildLightBlockChain(ctx, t, 1, 10, time.Now())
// app accepts any snapshot
rts.conn.On("OfferSnapshotSync", ctx, mock.AnythingOfType("types.RequestOfferSnapshot")).
rts.conn.On("OfferSnapshot", ctx, mock.AnythingOfType("types.RequestOfferSnapshot")).
Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_ACCEPT}, nil)
// app accepts every chunk
rts.conn.On("ApplySnapshotChunkSync", ctx, mock.AnythingOfType("types.RequestApplySnapshotChunk")).
rts.conn.On("ApplySnapshotChunk", ctx, mock.AnythingOfType("types.RequestApplySnapshotChunk")).
Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
// app query returns valid state app hash
rts.connQuery.On("InfoSync", mock.Anything, proxy.RequestInfo).Return(&abci.ResponseInfo{
rts.connQuery.On("Info", mock.Anything, proxy.RequestInfo).Return(&abci.ResponseInfo{
AppVersion: testAppVersion,
LastBlockHeight: snapshotHeight,
LastBlockAppHash: chain[snapshotHeight+1].AppHash,
@@ -317,7 +317,7 @@ func TestReactor_ChunkRequest(t *testing.T) {
// mock ABCI connection to return local snapshots
conn := &proxymocks.AppConnSnapshot{}
conn.On("LoadSnapshotChunkSync", mock.Anything, abci.RequestLoadSnapshotChunk{
conn.On("LoadSnapshotChunk", mock.Anything, abci.RequestLoadSnapshotChunk{
Height: tc.request.Height,
Format: tc.request.Format,
Chunk: tc.request.Index,
@@ -404,7 +404,7 @@ func TestReactor_SnapshotsRequest(t *testing.T) {
// mock ABCI connection to return local snapshots
conn := &proxymocks.AppConnSnapshot{}
conn.On("ListSnapshotsSync", mock.Anything, abci.RequestListSnapshots{}).Return(&abci.ResponseListSnapshots{
conn.On("ListSnapshots", mock.Anything, abci.RequestListSnapshots{}).Return(&abci.ResponseListSnapshots{
Snapshots: tc.snapshots,
}, nil)
internal/statesync/syncer.go
+3 -3
View File
@@ -367,7 +367,7 @@ func (s *syncer) Sync(ctx context.Context, snapshot *snapshot, chunks *chunkQueu
func (s *syncer) offerSnapshot(ctx context.Context, snapshot *snapshot) error {
s.logger.Info("Offering snapshot to ABCI app", "height", snapshot.Height,
"format", snapshot.Format, "hash", snapshot.Hash)
resp, err := s.conn.OfferSnapshotSync(ctx, abci.RequestOfferSnapshot{
resp, err := s.conn.OfferSnapshot(ctx, abci.RequestOfferSnapshot{
Snapshot: &abci.Snapshot{
Height: snapshot.Height,
Format: snapshot.Format,
@@ -409,7 +409,7 @@ func (s *syncer) applyChunks(ctx context.Context, chunks *chunkQueue, start time
return fmt.Errorf("failed to fetch chunk: %w", err)
}
resp, err := s.conn.ApplySnapshotChunkSync(ctx, abci.RequestApplySnapshotChunk{
resp, err := s.conn.ApplySnapshotChunk(ctx, abci.RequestApplySnapshotChunk{
Index: chunk.Index,
Chunk: chunk.Chunk,
Sender: string(chunk.Sender),
@@ -550,7 +550,7 @@ func (s *syncer) requestChunk(ctx context.Context, snapshot *snapshot, chunk uin
// verifyApp verifies the sync, checking the app hash and last block height. It returns the
// app version, which should be returned as part of the initial state.
func (s *syncer) verifyApp(ctx context.Context, snapshot *snapshot) (uint64, error) {
resp, err := s.connQuery.InfoSync(ctx, proxy.RequestInfo)
resp, err := s.connQuery.Info(ctx, proxy.RequestInfo)
if err != nil {
return 0, fmt.Errorf("failed to query ABCI app for appHash: %w", err)
}
internal/statesync/syncer_test.go
+27 -27
View File
@@ -110,7 +110,7 @@ func TestSyncer_SyncAny(t *testing.T) {
// We start a sync, with peers sending back chunks when requested. We first reject the snapshot
// with height 2 format 2, and accept the snapshot at height 1.
connSnapshot.On("OfferSnapshotSync", mock.Anything, abci.RequestOfferSnapshot{
connSnapshot.On("OfferSnapshot", mock.Anything, abci.RequestOfferSnapshot{
Snapshot: &abci.Snapshot{
Height: 2,
Format: 2,
@@ -119,7 +119,7 @@ func TestSyncer_SyncAny(t *testing.T) {
},
AppHash: []byte("app_hash_2"),
}).Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT_FORMAT}, nil)
connSnapshot.On("OfferSnapshotSync", mock.Anything, abci.RequestOfferSnapshot{
connSnapshot.On("OfferSnapshot", mock.Anything, abci.RequestOfferSnapshot{
Snapshot: &abci.Snapshot{
Height: s.Height,
Format: s.Format,
@@ -171,7 +171,7 @@ func TestSyncer_SyncAny(t *testing.T) {
// The first time we're applying chunk 2 we tell it to retry the snapshot and discard chunk 1,
// which should cause it to keep the existing chunk 0 and 2, and restart restoration from
// beginning. We also wait for a little while, to exercise the retry logic in fetchChunks().
connSnapshot.On("ApplySnapshotChunkSync", mock.Anything, abci.RequestApplySnapshotChunk{
connSnapshot.On("ApplySnapshotChunk", mock.Anything, abci.RequestApplySnapshotChunk{
Index: 2, Chunk: []byte{1, 1, 2},
}).Once().Run(func(args mock.Arguments) { time.Sleep(2 * time.Second) }).Return(
&abci.ResponseApplySnapshotChunk{
@@ -179,16 +179,16 @@ func TestSyncer_SyncAny(t *testing.T) {
RefetchChunks: []uint32{1},
}, nil)
connSnapshot.On("ApplySnapshotChunkSync", mock.Anything, abci.RequestApplySnapshotChunk{
connSnapshot.On("ApplySnapshotChunk", mock.Anything, abci.RequestApplySnapshotChunk{
Index: 0, Chunk: []byte{1, 1, 0},
}).Times(2).Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
connSnapshot.On("ApplySnapshotChunkSync", mock.Anything, abci.RequestApplySnapshotChunk{
connSnapshot.On("ApplySnapshotChunk", mock.Anything, abci.RequestApplySnapshotChunk{
Index: 1, Chunk: []byte{1, 1, 1},
}).Times(2).Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
connSnapshot.On("ApplySnapshotChunkSync", mock.Anything, abci.RequestApplySnapshotChunk{
connSnapshot.On("ApplySnapshotChunk", mock.Anything, abci.RequestApplySnapshotChunk{
Index: 2, Chunk: []byte{1, 1, 2},
}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
connQuery.On("InfoSync", mock.Anything, proxy.RequestInfo).Return(&abci.ResponseInfo{
connQuery.On("Info", mock.Anything, proxy.RequestInfo).Return(&abci.ResponseInfo{
AppVersion: 9,
LastBlockHeight: 1,
LastBlockAppHash: []byte("app_hash"),
@@ -252,7 +252,7 @@ func TestSyncer_SyncAny_abort(t *testing.T) {
_, err := rts.syncer.AddSnapshot(peerID, s)
require.NoError(t, err)
rts.conn.On("OfferSnapshotSync", mock.Anything, abci.RequestOfferSnapshot{
rts.conn.On("OfferSnapshot", mock.Anything, abci.RequestOfferSnapshot{
Snapshot: toABCI(s), AppHash: []byte("app_hash"),
}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_ABORT}, nil)
@@ -286,15 +286,15 @@ func TestSyncer_SyncAny_reject(t *testing.T) {
_, err = rts.syncer.AddSnapshot(peerID, s11)
require.NoError(t, err)
rts.conn.On("OfferSnapshotSync", mock.Anything, abci.RequestOfferSnapshot{
rts.conn.On("OfferSnapshot", mock.Anything, abci.RequestOfferSnapshot{
Snapshot: toABCI(s22), AppHash: []byte("app_hash"),
}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT}, nil)
rts.conn.On("OfferSnapshotSync", mock.Anything, abci.RequestOfferSnapshot{
rts.conn.On("OfferSnapshot", mock.Anything, abci.RequestOfferSnapshot{
Snapshot: toABCI(s12), AppHash: []byte("app_hash"),
}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT}, nil)
rts.conn.On("OfferSnapshotSync", mock.Anything, abci.RequestOfferSnapshot{
rts.conn.On("OfferSnapshot", mock.Anything, abci.RequestOfferSnapshot{
Snapshot: toABCI(s11), AppHash: []byte("app_hash"),
}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT}, nil)
@@ -328,11 +328,11 @@ func TestSyncer_SyncAny_reject_format(t *testing.T) {
_, err = rts.syncer.AddSnapshot(peerID, s11)
require.NoError(t, err)
rts.conn.On("OfferSnapshotSync", mock.Anything, abci.RequestOfferSnapshot{
rts.conn.On("OfferSnapshot", mock.Anything, abci.RequestOfferSnapshot{
Snapshot: toABCI(s22), AppHash: []byte("app_hash"),
}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT_FORMAT}, nil)
rts.conn.On("OfferSnapshotSync", mock.Anything, abci.RequestOfferSnapshot{
rts.conn.On("OfferSnapshot", mock.Anything, abci.RequestOfferSnapshot{
Snapshot: toABCI(s11), AppHash: []byte("app_hash"),
}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_ABORT}, nil)
@@ -377,11 +377,11 @@ func TestSyncer_SyncAny_reject_sender(t *testing.T) {
_, err = rts.syncer.AddSnapshot(peerCID, sbc)
require.NoError(t, err)
rts.conn.On("OfferSnapshotSync", mock.Anything, abci.RequestOfferSnapshot{
rts.conn.On("OfferSnapshot", mock.Anything, abci.RequestOfferSnapshot{
Snapshot: toABCI(sbc), AppHash: []byte("app_hash"),
}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT_SENDER}, nil)
rts.conn.On("OfferSnapshotSync", mock.Anything, abci.RequestOfferSnapshot{
rts.conn.On("OfferSnapshot", mock.Anything, abci.RequestOfferSnapshot{
Snapshot: toABCI(sa), AppHash: []byte("app_hash"),
}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT}, nil)
@@ -407,7 +407,7 @@ func TestSyncer_SyncAny_abciError(t *testing.T) {
_, err := rts.syncer.AddSnapshot(peerID, s)
require.NoError(t, err)
rts.conn.On("OfferSnapshotSync", mock.Anything, abci.RequestOfferSnapshot{
rts.conn.On("OfferSnapshot", mock.Anything, abci.RequestOfferSnapshot{
Snapshot: toABCI(s), AppHash: []byte("app_hash"),
}).Once().Return(nil, errBoom)
@@ -450,7 +450,7 @@ func TestSyncer_offerSnapshot(t *testing.T) {
rts := setup(ctx, t, nil, nil, stateProvider, 2)
s := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}, trustedAppHash: []byte("app_hash")}
rts.conn.On("OfferSnapshotSync", mock.Anything, abci.RequestOfferSnapshot{
rts.conn.On("OfferSnapshot", mock.Anything, abci.RequestOfferSnapshot{
Snapshot: toABCI(s),
AppHash: []byte("app_hash"),
}).Return(&abci.ResponseOfferSnapshot{Result: tc.result}, tc.err)
@@ -511,11 +511,11 @@ func TestSyncer_applyChunks_Results(t *testing.T) {
_, err = chunks.Add(&chunk{Height: 1, Format: 1, Index: 0, Chunk: body})
require.NoError(t, err)
rts.conn.On("ApplySnapshotChunkSync", mock.Anything, abci.RequestApplySnapshotChunk{
rts.conn.On("ApplySnapshotChunk", mock.Anything, abci.RequestApplySnapshotChunk{
Index: 0, Chunk: body,
}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: tc.result}, tc.err)
if tc.result == abci.ResponseApplySnapshotChunk_RETRY {
rts.conn.On("ApplySnapshotChunkSync", mock.Anything, abci.RequestApplySnapshotChunk{
rts.conn.On("ApplySnapshotChunk", mock.Anything, abci.RequestApplySnapshotChunk{
Index: 0, Chunk: body,
}).Once().Return(&abci.ResponseApplySnapshotChunk{
Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
@@ -578,13 +578,13 @@ func TestSyncer_applyChunks_RefetchChunks(t *testing.T) {
require.NoError(t, err)
// The first two chunks are accepted, before the last one asks for 1 to be refetched
rts.conn.On("ApplySnapshotChunkSync", mock.Anything, abci.RequestApplySnapshotChunk{
rts.conn.On("ApplySnapshotChunk", mock.Anything, abci.RequestApplySnapshotChunk{
Index: 0, Chunk: []byte{0},
}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
rts.conn.On("ApplySnapshotChunkSync", mock.Anything, abci.RequestApplySnapshotChunk{
rts.conn.On("ApplySnapshotChunk", mock.Anything, abci.RequestApplySnapshotChunk{
Index: 1, Chunk: []byte{1},
}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
rts.conn.On("ApplySnapshotChunkSync", mock.Anything, abci.RequestApplySnapshotChunk{
rts.conn.On("ApplySnapshotChunk", mock.Anything, abci.RequestApplySnapshotChunk{
Index: 2, Chunk: []byte{2},
}).Once().Return(&abci.ResponseApplySnapshotChunk{
Result: tc.result,
@@ -678,13 +678,13 @@ func TestSyncer_applyChunks_RejectSenders(t *testing.T) {
require.NoError(t, err)
// The first two chunks are accepted, before the last one asks for b sender to be rejected
rts.conn.On("ApplySnapshotChunkSync", mock.Anything, abci.RequestApplySnapshotChunk{
rts.conn.On("ApplySnapshotChunk", mock.Anything, abci.RequestApplySnapshotChunk{
Index: 0, Chunk: []byte{0}, Sender: "aa",
}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
rts.conn.On("ApplySnapshotChunkSync", mock.Anything, abci.RequestApplySnapshotChunk{
rts.conn.On("ApplySnapshotChunk", mock.Anything, abci.RequestApplySnapshotChunk{
Index: 1, Chunk: []byte{1}, Sender: "bb",
}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
rts.conn.On("ApplySnapshotChunkSync", mock.Anything, abci.RequestApplySnapshotChunk{
rts.conn.On("ApplySnapshotChunk", mock.Anything, abci.RequestApplySnapshotChunk{
Index: 2, Chunk: []byte{2}, Sender: "cc",
}).Once().Return(&abci.ResponseApplySnapshotChunk{
Result: tc.result,
@@ -693,7 +693,7 @@ func TestSyncer_applyChunks_RejectSenders(t *testing.T) {
// On retry, the last chunk will be tried again, so we just accept it then.
if tc.result == abci.ResponseApplySnapshotChunk_RETRY {
rts.conn.On("ApplySnapshotChunkSync", mock.Anything, abci.RequestApplySnapshotChunk{
rts.conn.On("ApplySnapshotChunk", mock.Anything, abci.RequestApplySnapshotChunk{
Index: 2, Chunk: []byte{2}, Sender: "cc",
}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
}
@@ -759,7 +759,7 @@ func TestSyncer_verifyApp(t *testing.T) {
rts := setup(ctx, t, nil, nil, nil, 2)
rts.connQuery.On("InfoSync", mock.Anything, proxy.RequestInfo).Return(tc.response, tc.err)
rts.connQuery.On("Info", mock.Anything, proxy.RequestInfo).Return(tc.response, tc.err)
version, err := rts.syncer.verifyApp(ctx, s)
unwrapped := errors.Unwrap(err)
if unwrapped != nil {