This repository has been archived by the owner on Mar 18, 2021. It is now read-only.
forked from tendermint/tendermint
-
Notifications
You must be signed in to change notification settings - Fork 0
/
execution.go
402 lines (341 loc) · 13.6 KB
/
execution.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
package state
import (
"fmt"
fail "github.com/ebuchman/fail-test"
abci "github.com/tendermint/tendermint/abci/types"
dbm "github.com/tendermint/tendermint/libs/db"
"github.com/tendermint/tendermint/libs/log"
"github.com/tendermint/tendermint/proxy"
"github.com/tendermint/tendermint/types"
)
//-----------------------------------------------------------------------------
// BlockExecutor handles block execution and state updates.
// It exposes ApplyBlock(), which validates & executes the block, updates state w/ ABCI responses,
// then commits and updates the mempool atomically, then saves state.
// BlockExecutor provides the context and accessories for properly executing a block.
type BlockExecutor struct {
// save state, validators, consensus params, abci responses here
db dbm.DB
// execute the app against this
proxyApp proxy.AppConnConsensus
// events
eventBus types.BlockEventPublisher
// update these with block results after commit
mempool Mempool
evpool EvidencePool
logger log.Logger
}
// NewBlockExecutor returns a new BlockExecutor with a NopEventBus.
// Call SetEventBus to provide one.
func NewBlockExecutor(db dbm.DB, logger log.Logger, proxyApp proxy.AppConnConsensus,
mempool Mempool, evpool EvidencePool) *BlockExecutor {
return &BlockExecutor{
db: db,
proxyApp: proxyApp,
eventBus: types.NopEventBus{},
mempool: mempool,
evpool: evpool,
logger: logger,
}
}
// SetEventBus - sets the event bus for publishing block related events.
// If not called, it defaults to types.NopEventBus.
func (blockExec *BlockExecutor) SetEventBus(eventBus types.BlockEventPublisher) {
blockExec.eventBus = eventBus
}
// ValidateBlock validates the given block against the given state.
// If the block is invalid, it returns an error.
// Validation does not mutate state, but does require historical information from the stateDB,
// ie. to verify evidence from a validator at an old height.
func (blockExec *BlockExecutor) ValidateBlock(state State, block *types.Block) error {
return validateBlock(blockExec.db, state, block)
}
// ApplyBlock validates the block against the state, executes it against the app,
// fires the relevant events, commits the app, and saves the new state and responses.
// It's the only function that needs to be called
// from outside this package to process and commit an entire block.
// It takes a blockID to avoid recomputing the parts hash.
func (blockExec *BlockExecutor) ApplyBlock(state State, blockID types.BlockID, block *types.Block) (State, error) {
if err := blockExec.ValidateBlock(state, block); err != nil {
return state, ErrInvalidBlock(err)
}
abciResponses, err := execBlockOnProxyApp(blockExec.logger, blockExec.proxyApp, block, state.LastValidators, blockExec.db)
if err != nil {
return state, ErrProxyAppConn(err)
}
fail.Fail() // XXX
// save the results before we commit
saveABCIResponses(blockExec.db, block.Height, abciResponses)
fail.Fail() // XXX
// update the state with the block and responses
state, err = updateState(state, blockID, block.Header, abciResponses)
if err != nil {
return state, fmt.Errorf("Commit failed for application: %v", err)
}
// lock mempool, commit app state, update mempoool
appHash, err := blockExec.Commit(block)
if err != nil {
return state, fmt.Errorf("Commit failed for application: %v", err)
}
// Update evpool with the block and state.
blockExec.evpool.Update(block, state)
fail.Fail() // XXX
// update the app hash and save the state
state.AppHash = appHash
SaveState(blockExec.db, state)
fail.Fail() // XXX
// events are fired after everything else
// NOTE: if we crash between Commit and Save, events wont be fired during replay
fireEvents(blockExec.logger, blockExec.eventBus, block, abciResponses)
return state, nil
}
// Commit locks the mempool, runs the ABCI Commit message, and updates the mempool.
// It returns the result of calling abci.Commit (the AppHash), and an error.
// The Mempool must be locked during commit and update because state is typically reset on Commit and old txs must be replayed
// against committed state before new txs are run in the mempool, lest they be invalid.
func (blockExec *BlockExecutor) Commit(block *types.Block) ([]byte, error) {
blockExec.mempool.Lock()
defer blockExec.mempool.Unlock()
// while mempool is Locked, flush to ensure all async requests have completed
// in the ABCI app before Commit.
err := blockExec.mempool.FlushAppConn()
if err != nil {
blockExec.logger.Error("Client error during mempool.FlushAppConn", "err", err)
return nil, err
}
// Commit block, get hash back
res, err := blockExec.proxyApp.CommitSync()
if err != nil {
blockExec.logger.Error("Client error during proxyAppConn.CommitSync", "err", err)
return nil, err
}
// ResponseCommit has no error code - just data
blockExec.logger.Info("Committed state",
"height", block.Height,
"txs", block.NumTxs,
"appHash", fmt.Sprintf("%X", res.Data))
// Update mempool.
if err := blockExec.mempool.Update(block.Height, block.Txs); err != nil {
return nil, err
}
return res.Data, nil
}
//---------------------------------------------------------
// Helper functions for executing blocks and updating state
// Executes block's transactions on proxyAppConn.
// Returns a list of transaction results and updates to the validator set
func execBlockOnProxyApp(logger log.Logger, proxyAppConn proxy.AppConnConsensus,
block *types.Block, lastValSet *types.ValidatorSet, stateDB dbm.DB) (*ABCIResponses, error) {
var validTxs, invalidTxs = 0, 0
txIndex := 0
abciResponses := NewABCIResponses(block)
// Execute transactions and get hash
proxyCb := func(req *abci.Request, res *abci.Response) {
switch r := res.Value.(type) {
case *abci.Response_DeliverTx:
// 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.DeliverTx[txIndex] = txRes
txIndex++
}
}
proxyAppConn.SetResponseCallback(proxyCb)
signVals, byzVals := getBeginBlockValidatorInfo(block, lastValSet, stateDB)
// Begin block
_, err := proxyAppConn.BeginBlockSync(abci.RequestBeginBlock{
Hash: block.Hash(),
Header: types.TM2PB.Header(block.Header),
Validators: signVals,
ByzantineValidators: byzVals,
})
if err != nil {
logger.Error("Error in proxyAppConn.BeginBlock", "err", err)
return nil, err
}
// Run txs of block
for _, tx := range block.Txs {
proxyAppConn.DeliverTxAsync(tx)
if err := proxyAppConn.Error(); err != nil {
return nil, err
}
}
// End block
abciResponses.EndBlock, err = proxyAppConn.EndBlockSync(abci.RequestEndBlock{block.Height})
if err != nil {
logger.Error("Error in proxyAppConn.EndBlock", "err", err)
return nil, err
}
logger.Info("Executed block", "height", block.Height, "validTxs", validTxs, "invalidTxs", invalidTxs)
valUpdates := abciResponses.EndBlock.ValidatorUpdates
if len(valUpdates) > 0 {
logger.Info("Updates to validators", "updates", abci.ValidatorsString(valUpdates))
}
return abciResponses, nil
}
func getBeginBlockValidatorInfo(block *types.Block, lastValSet *types.ValidatorSet, stateDB dbm.DB) ([]abci.SigningValidator, []abci.Evidence) {
// Sanity check that commit length matches validator set size -
// only applies after first block
if block.Height > 1 {
precommitLen := len(block.LastCommit.Precommits)
valSetLen := len(lastValSet.Validators)
if precommitLen != valSetLen {
// sanity check
panic(fmt.Sprintf("precommit length (%d) doesn't match valset length (%d) at height %d\n\n%v\n\n%v",
precommitLen, valSetLen, block.Height, block.LastCommit.Precommits, lastValSet.Validators))
}
}
// determine which validators did not sign last block.
signVals := make([]abci.SigningValidator, len(lastValSet.Validators))
for i, val := range lastValSet.Validators {
var vote *types.Vote
if i < len(block.LastCommit.Precommits) {
vote = block.LastCommit.Precommits[i]
}
val := abci.SigningValidator{
Validator: types.TM2PB.Validator(val),
SignedLastBlock: vote != nil,
}
signVals[i] = val
}
byzVals := make([]abci.Evidence, len(block.Evidence.Evidence))
for i, ev := range block.Evidence.Evidence {
// We need the validator set. We already did this in validateBlock.
// TODO: Should we instead cache the valset in the evidence itself and add
// `SetValidatorSet()` and `ToABCI` methods ?
valset, err := LoadValidators(stateDB, ev.Height())
if err != nil {
panic(err) // shouldn't happen
}
byzVals[i] = types.TM2PB.Evidence(ev, valset, block.Time)
}
return signVals, byzVals
}
// If more or equal than 1/3 of total voting power changed in one block, then
// a light client could never prove the transition externally. See
// ./lite/doc.go for details on how a light client tracks validators.
func updateValidators(currentSet *types.ValidatorSet, abciUpdates []abci.Validator) error {
updates, err := types.PB2TM.Validators(abciUpdates)
if err != nil {
return err
}
// these are tendermint types now
for _, valUpdate := range updates {
if valUpdate.VotingPower < 0 {
return fmt.Errorf("Voting power can't be negative %v", valUpdate)
}
address := valUpdate.Address
_, val := currentSet.GetByAddress(address)
if valUpdate.VotingPower == 0 {
// remove val
_, removed := currentSet.Remove(address)
if !removed {
return fmt.Errorf("Failed to remove validator %X", address)
}
} else if val == nil {
// add val
added := currentSet.Add(valUpdate)
if !added {
return fmt.Errorf("Failed to add new validator %v", valUpdate)
}
} else {
// update val
updated := currentSet.Update(valUpdate)
if !updated {
return fmt.Errorf("Failed to update validator %X to %v", address, valUpdate)
}
}
}
return nil
}
// updateState returns a new State updated according to the header and responses.
func updateState(state State, blockID types.BlockID, header *types.Header,
abciResponses *ABCIResponses) (State, error) {
// copy the valset so we can apply changes from EndBlock
// and update s.LastValidators and s.Validators
prevValSet := state.Validators.Copy()
nextValSet := prevValSet.Copy()
// update the validator set with the latest abciResponses
lastHeightValsChanged := state.LastHeightValidatorsChanged
if len(abciResponses.EndBlock.ValidatorUpdates) > 0 {
err := updateValidators(nextValSet, abciResponses.EndBlock.ValidatorUpdates)
if err != nil {
return state, fmt.Errorf("Error changing validator set: %v", err)
}
// change results from this height but only applies to the next height
lastHeightValsChanged = header.Height + 1
}
// Update validator accums and set state variables
nextValSet.IncrementAccum(1)
// update the params with the latest abciResponses
nextParams := state.ConsensusParams
lastHeightParamsChanged := state.LastHeightConsensusParamsChanged
if abciResponses.EndBlock.ConsensusParamUpdates != nil {
// NOTE: must not mutate s.ConsensusParams
nextParams = state.ConsensusParams.Update(abciResponses.EndBlock.ConsensusParamUpdates)
err := nextParams.Validate()
if err != nil {
return state, fmt.Errorf("Error updating consensus params: %v", err)
}
// change results from this height but only applies to the next height
lastHeightParamsChanged = header.Height + 1
}
// NOTE: the AppHash has not been populated.
// It will be filled on state.Save.
return State{
ChainID: state.ChainID,
LastBlockHeight: header.Height,
LastBlockTotalTx: state.LastBlockTotalTx + header.NumTxs,
LastBlockID: blockID,
LastBlockTime: header.Time,
Validators: nextValSet,
LastValidators: state.Validators.Copy(),
LastHeightValidatorsChanged: lastHeightValsChanged,
ConsensusParams: nextParams,
LastHeightConsensusParamsChanged: lastHeightParamsChanged,
LastResultsHash: abciResponses.ResultsHash(),
AppHash: nil,
}, nil
}
// Fire NewBlock, NewBlockHeader.
// Fire TxEvent for every tx.
// NOTE: if Tendermint crashes before commit, some or all of these events may be published again.
func fireEvents(logger log.Logger, eventBus types.BlockEventPublisher, block *types.Block, abciResponses *ABCIResponses) {
eventBus.PublishEventNewBlock(types.EventDataNewBlock{block})
eventBus.PublishEventNewBlockHeader(types.EventDataNewBlockHeader{block.Header})
for i, tx := range block.Data.Txs {
eventBus.PublishEventTx(types.EventDataTx{types.TxResult{
Height: block.Height,
Index: uint32(i),
Tx: tx,
Result: *(abciResponses.DeliverTx[i]),
}})
}
}
//----------------------------------------------------------------------------------------------------
// Execute block without state. TODO: eliminate
// ExecCommitBlock executes and commits a block on the proxyApp without validating or mutating the state.
// It returns the application root hash (result of abci.Commit).
func ExecCommitBlock(appConnConsensus proxy.AppConnConsensus, block *types.Block,
logger log.Logger, lastValSet *types.ValidatorSet, stateDB dbm.DB) ([]byte, error) {
_, err := execBlockOnProxyApp(logger, appConnConsensus, block, lastValSet, stateDB)
if err != nil {
logger.Error("Error executing block on proxy app", "height", block.Height, "err", err)
return nil, err
}
// Commit block, get hash back
res, err := appConnConsensus.CommitSync()
if err != nil {
logger.Error("Client error during proxyAppConn.CommitSync", "err", res)
return nil, err
}
// ResponseCommit has no error or log, just data
return res.Data, nil
}