mirror of
https://github.com/fluencelabs/tendermint
synced 2025-07-25 01:01:56 +00:00
Compare commits
7 Commits
v0.31.2-rc
...
release/v0
Author | SHA1 | Date | |
---|---|---|---|
|
3cfd9757a7 | ||
|
882622ec10 | ||
|
1ecf814838 | ||
|
79e9f20578 | ||
|
ab24925c94 | ||
|
0ae41cc663 | ||
|
422d04c8ba |
51
CHANGELOG.md
51
CHANGELOG.md
@@ -1,5 +1,56 @@
|
||||
# Changelog
|
||||
|
||||
## v0.31.3
|
||||
|
||||
*April 1st, 2019*
|
||||
|
||||
This release includes two security sensitive fixes: it ensures generated private
|
||||
keys are valid, and it prevents certain DNS lookups that would cause the node to
|
||||
panic if the lookup failed.
|
||||
|
||||
### BUG FIXES:
|
||||
|
||||
- [crypto/secp256k1] [\#3439](https://github.com/tendermint/tendermint/issues/3439)
|
||||
Ensure generated private keys are valid by randomly sampling until a valid key is found.
|
||||
Previously, it was possible (though rare!) to generate keys that exceeded the curve order.
|
||||
Such keys would lead to invalid signatures.
|
||||
- [p2p] [\#3522](https://github.com/tendermint/tendermint/issues/3522) Memoize
|
||||
socket address in peer connections to avoid DNS lookups. Previously, failed
|
||||
DNS lookups could cause the node to panic.
|
||||
|
||||
## v0.31.2
|
||||
|
||||
*March 30th, 2019*
|
||||
|
||||
This release fixes a regression from v0.31.1 where Tendermint panics under
|
||||
mempool load for external ABCI apps.
|
||||
|
||||
Special thanks to external contributors on this release:
|
||||
@guagualvcha
|
||||
|
||||
### BREAKING CHANGES:
|
||||
|
||||
* CLI/RPC/Config
|
||||
|
||||
* Apps
|
||||
|
||||
* Go API
|
||||
- [libs/autofile] [\#3504](https://github.com/tendermint/tendermint/issues/3504) Remove unused code in autofile package. Deleted functions: `Group.Search`, `Group.FindLast`, `GroupReader.ReadLine`, `GroupReader.PushLine`, `MakeSimpleSearchFunc` (@guagualvcha)
|
||||
|
||||
* Blockchain Protocol
|
||||
|
||||
* P2P Protocol
|
||||
|
||||
### FEATURES:
|
||||
|
||||
### IMPROVEMENTS:
|
||||
|
||||
- [circle] [\#3497](https://github.com/tendermint/tendermint/issues/3497) Move release management to CircleCI
|
||||
|
||||
### BUG FIXES:
|
||||
|
||||
- [mempool] [\#3512](https://github.com/tendermint/tendermint/issues/3512) Fix panic from concurrent access to txsMap, a regression for external ABCI apps introduced in v0.31.1
|
||||
|
||||
## v0.31.1
|
||||
|
||||
*March 27th, 2019*
|
||||
|
@@ -1,4 +1,4 @@
|
||||
## v0.32.0
|
||||
## v0.31.2
|
||||
|
||||
**
|
||||
|
||||
@@ -9,7 +9,6 @@
|
||||
* Apps
|
||||
|
||||
* Go API
|
||||
- [libs/autofile] \#3504 Remove unused code in autofile package. Deleted functions: `Group.Search`, `Group.FindLast`, `GroupReader.ReadLine`, `GroupReader.PushLine`, `MakeSimpleSearchFunc` (@guagualvcha)
|
||||
|
||||
* Blockchain Protocol
|
||||
|
||||
@@ -19,6 +18,5 @@
|
||||
|
||||
### IMPROVEMENTS:
|
||||
|
||||
- [CircleCI] \#3497 Move release management to CircleCI
|
||||
|
||||
### BUG FIXES:
|
||||
|
||||
|
@@ -26,16 +26,17 @@ var _ Client = (*socketClient)(nil)
|
||||
type socketClient struct {
|
||||
cmn.BaseService
|
||||
|
||||
reqQueue chan *ReqRes
|
||||
flushTimer *cmn.ThrottleTimer
|
||||
addr string
|
||||
mustConnect bool
|
||||
conn net.Conn
|
||||
|
||||
reqQueue chan *ReqRes
|
||||
flushTimer *cmn.ThrottleTimer
|
||||
|
||||
mtx sync.Mutex
|
||||
addr string
|
||||
conn net.Conn
|
||||
err error
|
||||
reqSent *list.List
|
||||
resCb func(*types.Request, *types.Response) // listens to all callbacks
|
||||
reqSent *list.List // list of requests sent, waiting for response
|
||||
resCb func(*types.Request, *types.Response) // called on all requests, if set.
|
||||
|
||||
}
|
||||
|
||||
@@ -86,6 +87,7 @@ func (cli *socketClient) OnStop() {
|
||||
cli.mtx.Lock()
|
||||
defer cli.mtx.Unlock()
|
||||
if cli.conn != nil {
|
||||
// does this really need a mutex?
|
||||
cli.conn.Close()
|
||||
}
|
||||
|
||||
@@ -207,12 +209,15 @@ func (cli *socketClient) didRecvResponse(res *types.Response) error {
|
||||
reqres.Done() // Release waiters
|
||||
cli.reqSent.Remove(next) // Pop first item from linked list
|
||||
|
||||
// Notify reqRes listener if set
|
||||
// Notify reqRes listener if set (request specific callback).
|
||||
// NOTE: it is possible this callback isn't set on the reqres object.
|
||||
// at this point, in which case it will be called after, when it is set.
|
||||
// TODO: should we move this after the resCb call so the order is always consistent?
|
||||
if cb := reqres.GetCallback(); cb != nil {
|
||||
cb(res)
|
||||
}
|
||||
|
||||
// Notify client listener if set
|
||||
// Notify client listener if set (global callback).
|
||||
if cli.resCb != nil {
|
||||
cli.resCb(reqres.Request, res)
|
||||
}
|
||||
|
@@ -6,6 +6,7 @@ import (
|
||||
"crypto/subtle"
|
||||
"fmt"
|
||||
"io"
|
||||
"math/big"
|
||||
|
||||
"golang.org/x/crypto/ripemd160"
|
||||
|
||||
@@ -65,32 +66,61 @@ func (privKey PrivKeySecp256k1) Equals(other crypto.PrivKey) bool {
|
||||
}
|
||||
|
||||
// GenPrivKey generates a new ECDSA private key on curve secp256k1 private key.
|
||||
// It uses OS randomness in conjunction with the current global random seed
|
||||
// in tendermint/libs/common to generate the private key.
|
||||
// It uses OS randomness to generate the private key.
|
||||
func GenPrivKey() PrivKeySecp256k1 {
|
||||
return genPrivKey(crypto.CReader())
|
||||
}
|
||||
|
||||
// genPrivKey generates a new secp256k1 private key using the provided reader.
|
||||
func genPrivKey(rand io.Reader) PrivKeySecp256k1 {
|
||||
privKeyBytes := [32]byte{}
|
||||
_, err := io.ReadFull(rand, privKeyBytes[:])
|
||||
if err != nil {
|
||||
panic(err)
|
||||
var privKeyBytes [32]byte
|
||||
d := new(big.Int)
|
||||
for {
|
||||
privKeyBytes = [32]byte{}
|
||||
_, err := io.ReadFull(rand, privKeyBytes[:])
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
d.SetBytes(privKeyBytes[:])
|
||||
// break if we found a valid point (i.e. > 0 and < N == curverOrder)
|
||||
isValidFieldElement := 0 < d.Sign() && d.Cmp(secp256k1.S256().N) < 0
|
||||
if isValidFieldElement {
|
||||
break
|
||||
}
|
||||
}
|
||||
// crypto.CRandBytes is guaranteed to be 32 bytes long, so it can be
|
||||
// casted to PrivKeySecp256k1.
|
||||
|
||||
return PrivKeySecp256k1(privKeyBytes)
|
||||
}
|
||||
|
||||
var one = new(big.Int).SetInt64(1)
|
||||
|
||||
// GenPrivKeySecp256k1 hashes the secret with SHA2, and uses
|
||||
// that 32 byte output to create the private key.
|
||||
//
|
||||
// It makes sure the private key is a valid field element by setting:
|
||||
//
|
||||
// c = sha256(secret)
|
||||
// k = (c mod (n − 1)) + 1, where n = curve order.
|
||||
//
|
||||
// NOTE: secret should be the output of a KDF like bcrypt,
|
||||
// if it's derived from user input.
|
||||
func GenPrivKeySecp256k1(secret []byte) PrivKeySecp256k1 {
|
||||
privKey32 := sha256.Sum256(secret)
|
||||
// sha256.Sum256() is guaranteed to be 32 bytes long, so it can be
|
||||
// casted to PrivKeySecp256k1.
|
||||
secHash := sha256.Sum256(secret)
|
||||
// to guarantee that we have a valid field element, we use the approach of:
|
||||
// "Suite B Implementer’s Guide to FIPS 186-3", A.2.1
|
||||
// https://apps.nsa.gov/iaarchive/library/ia-guidance/ia-solutions-for-classified/algorithm-guidance/suite-b-implementers-guide-to-fips-186-3-ecdsa.cfm
|
||||
// see also https://github.com/golang/go/blob/0380c9ad38843d523d9c9804fe300cb7edd7cd3c/src/crypto/ecdsa/ecdsa.go#L89-L101
|
||||
fe := new(big.Int).SetBytes(secHash[:])
|
||||
n := new(big.Int).Sub(secp256k1.S256().N, one)
|
||||
fe.Mod(fe, n)
|
||||
fe.Add(fe, one)
|
||||
|
||||
feB := fe.Bytes()
|
||||
var privKey32 [32]byte
|
||||
// copy feB over to fixed 32 byte privKey32 and pad (if necessary)
|
||||
copy(privKey32[32-len(feB):32], feB)
|
||||
|
||||
return PrivKeySecp256k1(privKey32)
|
||||
}
|
||||
|
||||
|
39
crypto/secp256k1/secp256k1_cgo_test.go
Normal file
39
crypto/secp256k1/secp256k1_cgo_test.go
Normal file
@@ -0,0 +1,39 @@
|
||||
// +build libsecp256k1
|
||||
|
||||
package secp256k1
|
||||
|
||||
import (
|
||||
"github.com/magiconair/properties/assert"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestPrivKeySecp256k1SignVerify(t *testing.T) {
|
||||
msg := []byte("A.1.2 ECC Key Pair Generation by Testing Candidates")
|
||||
priv := GenPrivKey()
|
||||
tests := []struct {
|
||||
name string
|
||||
privKey PrivKeySecp256k1
|
||||
wantSignErr bool
|
||||
wantVerifyPasses bool
|
||||
}{
|
||||
{name: "valid sign-verify round", privKey: priv, wantSignErr: false, wantVerifyPasses: true},
|
||||
{name: "invalid private key", privKey: [32]byte{}, wantSignErr: true, wantVerifyPasses: false},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
got, err := tt.privKey.Sign(msg)
|
||||
if tt.wantSignErr {
|
||||
require.Error(t, err)
|
||||
t.Logf("Got error: %s", err)
|
||||
return
|
||||
}
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, got)
|
||||
|
||||
pub := tt.privKey.PubKey()
|
||||
assert.Equal(t, tt.wantVerifyPasses, pub.VerifyBytes(msg, got))
|
||||
})
|
||||
}
|
||||
}
|
45
crypto/secp256k1/secp256k1_internal_test.go
Normal file
45
crypto/secp256k1/secp256k1_internal_test.go
Normal file
@@ -0,0 +1,45 @@
|
||||
package secp256k1
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"math/big"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
underlyingSecp256k1 "github.com/btcsuite/btcd/btcec"
|
||||
)
|
||||
|
||||
func Test_genPrivKey(t *testing.T) {
|
||||
|
||||
empty := make([]byte, 32)
|
||||
oneB := big.NewInt(1).Bytes()
|
||||
onePadded := make([]byte, 32)
|
||||
copy(onePadded[32-len(oneB):32], oneB)
|
||||
t.Logf("one padded: %v, len=%v", onePadded, len(onePadded))
|
||||
|
||||
validOne := append(empty, onePadded...)
|
||||
tests := []struct {
|
||||
name string
|
||||
notSoRand []byte
|
||||
shouldPanic bool
|
||||
}{
|
||||
{"empty bytes (panics because 1st 32 bytes are zero and 0 is not a valid field element)", empty, true},
|
||||
{"curve order: N", underlyingSecp256k1.S256().N.Bytes(), true},
|
||||
{"valid because 0 < 1 < N", validOne, false},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
if tt.shouldPanic {
|
||||
require.Panics(t, func() {
|
||||
genPrivKey(bytes.NewReader(tt.notSoRand))
|
||||
})
|
||||
return
|
||||
}
|
||||
got := genPrivKey(bytes.NewReader(tt.notSoRand))
|
||||
fe := new(big.Int).SetBytes(got[:])
|
||||
require.True(t, fe.Cmp(underlyingSecp256k1.S256().N) < 0)
|
||||
require.True(t, fe.Sign() > 0)
|
||||
})
|
||||
}
|
||||
}
|
@@ -6,7 +6,6 @@ import (
|
||||
"testing"
|
||||
|
||||
secp256k1 "github.com/btcsuite/btcd/btcec"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
|
@@ -2,6 +2,7 @@ package secp256k1_test
|
||||
|
||||
import (
|
||||
"encoding/hex"
|
||||
"math/big"
|
||||
"testing"
|
||||
|
||||
"github.com/btcsuite/btcutil/base58"
|
||||
@@ -84,3 +85,28 @@ func TestSecp256k1LoadPrivkeyAndSerializeIsIdentity(t *testing.T) {
|
||||
require.Equal(t, privKeyBytes[:], serializedBytes)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGenPrivKeySecp256k1(t *testing.T) {
|
||||
// curve oder N
|
||||
N := underlyingSecp256k1.S256().N
|
||||
tests := []struct {
|
||||
name string
|
||||
secret []byte
|
||||
}{
|
||||
{"empty secret", []byte{}},
|
||||
{"some long secret", []byte("We live in a society exquisitely dependent on science and technology, in which hardly anyone knows anything about science and technology.")},
|
||||
{"another seed used in cosmos tests #1", []byte{0}},
|
||||
{"another seed used in cosmos tests #2", []byte("mySecret")},
|
||||
{"another seed used in cosmos tests #3", []byte("")},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
gotPrivKey := secp256k1.GenPrivKeySecp256k1(tt.secret)
|
||||
require.NotNil(t, gotPrivKey)
|
||||
// interpret as a big.Int and make sure it is a valid field element:
|
||||
fe := new(big.Int).SetBytes(gotPrivKey[:])
|
||||
require.True(t, fe.Cmp(N) < 0)
|
||||
require.True(t, fe.Sign() > 0)
|
||||
})
|
||||
}
|
||||
}
|
@@ -19,7 +19,7 @@ func TestCacheRemove(t *testing.T) {
|
||||
for i := 0; i < numTxs; i++ {
|
||||
// probability of collision is 2**-256
|
||||
txBytes := make([]byte, 32)
|
||||
rand.Read(txBytes)
|
||||
rand.Read(txBytes) // nolint: gosec
|
||||
txs[i] = txBytes
|
||||
cache.Push(txBytes)
|
||||
// make sure its added to both the linked list and the map
|
||||
|
@@ -149,6 +149,11 @@ func TxID(tx []byte) string {
|
||||
return fmt.Sprintf("%X", types.Tx(tx).Hash())
|
||||
}
|
||||
|
||||
// txKey is the fixed length array sha256 hash used as the key in maps.
|
||||
func txKey(tx types.Tx) [sha256.Size]byte {
|
||||
return sha256.Sum256(tx)
|
||||
}
|
||||
|
||||
// Mempool is an ordered in-memory pool for transactions before they are proposed in a consensus
|
||||
// round. Transaction validity is checked using the CheckTx abci message before the transaction is
|
||||
// added to the pool. The Mempool uses a concurrent list structure for storing transactions that
|
||||
@@ -159,23 +164,27 @@ type Mempool struct {
|
||||
proxyMtx sync.Mutex
|
||||
proxyAppConn proxy.AppConnMempool
|
||||
txs *clist.CList // concurrent linked-list of good txs
|
||||
// map for quick access to txs
|
||||
// Used in CheckTx to record the tx sender.
|
||||
txsMap map[[sha256.Size]byte]*clist.CElement
|
||||
height int64 // the last block Update()'d to
|
||||
rechecking int32 // for re-checking filtered txs on Update()
|
||||
recheckCursor *clist.CElement // next expected response
|
||||
recheckEnd *clist.CElement // re-checking stops here
|
||||
preCheck PreCheckFunc
|
||||
postCheck PostCheckFunc
|
||||
|
||||
// Track whether we're rechecking txs.
|
||||
// These are not protected by a mutex and are expected to be mutated
|
||||
// in serial (ie. by abci responses which are called in serial).
|
||||
recheckCursor *clist.CElement // next expected response
|
||||
recheckEnd *clist.CElement // re-checking stops here
|
||||
|
||||
// notify listeners (ie. consensus) when txs are available
|
||||
notifiedTxsAvailable bool
|
||||
txsAvailable chan struct{} // fires once for each height, when the mempool is not empty
|
||||
preCheck PreCheckFunc
|
||||
postCheck PostCheckFunc
|
||||
|
||||
// Map for quick access to txs to record sender in CheckTx.
|
||||
// txsMap: txKey -> CElement
|
||||
txsMap sync.Map
|
||||
|
||||
// Atomic integers
|
||||
|
||||
// Used to check if the mempool size is bigger than the allowed limit.
|
||||
// See TxsBytes
|
||||
txsBytes int64
|
||||
height int64 // the last block Update()'d to
|
||||
rechecking int32 // for re-checking filtered txs on Update()
|
||||
txsBytes int64 // total size of mempool, in bytes
|
||||
|
||||
// Keep a cache of already-seen txs.
|
||||
// This reduces the pressure on the proxyApp.
|
||||
@@ -203,7 +212,6 @@ func NewMempool(
|
||||
config: config,
|
||||
proxyAppConn: proxyAppConn,
|
||||
txs: clist.New(),
|
||||
txsMap: make(map[[sha256.Size]byte]*clist.CElement),
|
||||
height: height,
|
||||
rechecking: 0,
|
||||
recheckCursor: nil,
|
||||
@@ -216,7 +224,7 @@ func NewMempool(
|
||||
} else {
|
||||
mempool.cache = nopTxCache{}
|
||||
}
|
||||
proxyAppConn.SetResponseCallback(mempool.resCb)
|
||||
proxyAppConn.SetResponseCallback(mempool.globalCb)
|
||||
for _, option := range options {
|
||||
option(mempool)
|
||||
}
|
||||
@@ -319,7 +327,7 @@ func (mem *Mempool) Flush() {
|
||||
e.DetachPrev()
|
||||
}
|
||||
|
||||
mem.txsMap = make(map[[sha256.Size]byte]*clist.CElement)
|
||||
mem.txsMap = sync.Map{}
|
||||
_ = atomic.SwapInt64(&mem.txsBytes, 0)
|
||||
}
|
||||
|
||||
@@ -380,13 +388,12 @@ func (mem *Mempool) CheckTxWithInfo(tx types.Tx, cb func(*abci.Response), txInfo
|
||||
|
||||
// CACHE
|
||||
if !mem.cache.Push(tx) {
|
||||
// record the sender
|
||||
e, ok := mem.txsMap[sha256.Sum256(tx)]
|
||||
// The check is needed because tx may be in cache, but not in the mempool.
|
||||
// E.g. after we've committed a block, txs are removed from the mempool,
|
||||
// but not from the cache.
|
||||
if ok {
|
||||
memTx := e.Value.(*mempoolTx)
|
||||
// Record a new sender for a tx we've already seen.
|
||||
// Note it's possible a tx is still in the cache but no longer in the mempool
|
||||
// (eg. after committing a block, txs are removed from mempool but not cache),
|
||||
// so we only record the sender for txs still in the mempool.
|
||||
if e, ok := mem.txsMap.Load(txKey(tx)); ok {
|
||||
memTx := e.(*clist.CElement).Value.(*mempoolTx)
|
||||
if _, loaded := memTx.senders.LoadOrStore(txInfo.PeerID, true); loaded {
|
||||
// TODO: consider punishing peer for dups,
|
||||
// its non-trivial since invalid txs can become valid,
|
||||
@@ -416,25 +423,21 @@ func (mem *Mempool) CheckTxWithInfo(tx types.Tx, cb func(*abci.Response), txInfo
|
||||
if err = mem.proxyAppConn.Error(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
reqRes := mem.proxyAppConn.CheckTxAsync(tx)
|
||||
if cb != nil {
|
||||
composedCallback := func(res *abci.Response) {
|
||||
mem.reqResCb(tx, txInfo.PeerID)(res)
|
||||
cb(res)
|
||||
}
|
||||
reqRes.SetCallback(composedCallback)
|
||||
} else {
|
||||
reqRes.SetCallback(mem.reqResCb(tx, txInfo.PeerID))
|
||||
}
|
||||
reqRes.SetCallback(mem.reqResCb(tx, txInfo.PeerID, cb))
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Global callback, which is called in the absence of the specific callback.
|
||||
//
|
||||
// In recheckTxs because no reqResCb (specific) callback is set, this callback
|
||||
// will be called.
|
||||
func (mem *Mempool) resCb(req *abci.Request, res *abci.Response) {
|
||||
// Global callback that will be called after every ABCI response.
|
||||
// Having a single global callback avoids needing to set a callback for each request.
|
||||
// However, processing the checkTx response requires the peerID (so we can track which txs we heard from who),
|
||||
// and peerID is not included in the ABCI request, so we have to set request-specific callbacks that
|
||||
// include this information. If we're not in the midst of a recheck, this function will just return,
|
||||
// so the request specific callback can do the work.
|
||||
// When rechecking, we don't need the peerID, so the recheck callback happens here.
|
||||
func (mem *Mempool) globalCb(req *abci.Request, res *abci.Response) {
|
||||
if mem.recheckCursor == nil {
|
||||
return
|
||||
}
|
||||
@@ -446,35 +449,50 @@ func (mem *Mempool) resCb(req *abci.Request, res *abci.Response) {
|
||||
mem.metrics.Size.Set(float64(mem.Size()))
|
||||
}
|
||||
|
||||
// Specific callback, which allows us to incorporate local information, like
|
||||
// the peer that sent us this tx, so we can avoid sending it back to the same
|
||||
// peer.
|
||||
// Request specific callback that should be set on individual reqRes objects
|
||||
// to incorporate local information when processing the response.
|
||||
// This allows us to track the peer that sent us this tx, so we can avoid sending it back to them.
|
||||
// NOTE: alternatively, we could include this information in the ABCI request itself.
|
||||
//
|
||||
// External callers of CheckTx, like the RPC, can also pass an externalCb through here that is called
|
||||
// when all other response processing is complete.
|
||||
//
|
||||
// Used in CheckTxWithInfo to record PeerID who sent us the tx.
|
||||
func (mem *Mempool) reqResCb(tx []byte, peerID uint16) func(res *abci.Response) {
|
||||
func (mem *Mempool) reqResCb(tx []byte, peerID uint16, externalCb func(*abci.Response)) func(res *abci.Response) {
|
||||
return func(res *abci.Response) {
|
||||
if mem.recheckCursor != nil {
|
||||
return
|
||||
// this should never happen
|
||||
panic("recheck cursor is not nil in reqResCb")
|
||||
}
|
||||
|
||||
mem.resCbFirstTime(tx, peerID, res)
|
||||
|
||||
// update metrics
|
||||
mem.metrics.Size.Set(float64(mem.Size()))
|
||||
|
||||
// passed in by the caller of CheckTx, eg. the RPC
|
||||
if externalCb != nil {
|
||||
externalCb(res)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Called from:
|
||||
// - resCbFirstTime (lock not held) if tx is valid
|
||||
func (mem *Mempool) addTx(memTx *mempoolTx) {
|
||||
e := mem.txs.PushBack(memTx)
|
||||
mem.txsMap[sha256.Sum256(memTx.tx)] = e
|
||||
mem.txsMap.Store(txKey(memTx.tx), e)
|
||||
atomic.AddInt64(&mem.txsBytes, int64(len(memTx.tx)))
|
||||
mem.metrics.TxSizeBytes.Observe(float64(len(memTx.tx)))
|
||||
}
|
||||
|
||||
// Called from:
|
||||
// - Update (lock held) if tx was committed
|
||||
// - resCbRecheck (lock not held) if tx was invalidated
|
||||
func (mem *Mempool) removeTx(tx types.Tx, elem *clist.CElement, removeFromCache bool) {
|
||||
mem.txs.Remove(elem)
|
||||
elem.DetachPrev()
|
||||
delete(mem.txsMap, sha256.Sum256(tx))
|
||||
mem.txsMap.Delete(txKey(tx))
|
||||
atomic.AddInt64(&mem.txsBytes, int64(-len(tx)))
|
||||
|
||||
if removeFromCache {
|
||||
@@ -733,7 +751,7 @@ func (mem *Mempool) recheckTxs(txs []types.Tx) {
|
||||
mem.recheckEnd = mem.txs.Back()
|
||||
|
||||
// Push txs to proxyAppConn
|
||||
// NOTE: reqResCb may be called concurrently.
|
||||
// NOTE: globalCb may be called concurrently.
|
||||
for _, tx := range txs {
|
||||
mem.proxyAppConn.CheckTxAsync(tx)
|
||||
}
|
||||
@@ -746,8 +764,11 @@ func (mem *Mempool) recheckTxs(txs []types.Tx) {
|
||||
type mempoolTx struct {
|
||||
height int64 // height that this tx had been validated in
|
||||
gasWanted int64 // amount of gas this tx states it will require
|
||||
senders sync.Map // ids of peers who've sent us this tx (as a map for quick lookups)
|
||||
tx types.Tx //
|
||||
|
||||
// ids of peers who've sent us this tx (as a map for quick lookups).
|
||||
// senders: PeerID -> bool
|
||||
senders sync.Map
|
||||
}
|
||||
|
||||
// Height returns the height for this transaction
|
||||
@@ -798,7 +819,7 @@ func (cache *mapTxCache) Push(tx types.Tx) bool {
|
||||
defer cache.mtx.Unlock()
|
||||
|
||||
// Use the tx hash in the cache
|
||||
txHash := sha256.Sum256(tx)
|
||||
txHash := txKey(tx)
|
||||
if moved, exists := cache.map_[txHash]; exists {
|
||||
cache.list.MoveToBack(moved)
|
||||
return false
|
||||
@@ -820,7 +841,7 @@ func (cache *mapTxCache) Push(tx types.Tx) bool {
|
||||
// Remove removes the given tx from the cache.
|
||||
func (cache *mapTxCache) Remove(tx types.Tx) {
|
||||
cache.mtx.Lock()
|
||||
txHash := sha256.Sum256(tx)
|
||||
txHash := txKey(tx)
|
||||
popped := cache.map_[txHash]
|
||||
delete(cache.map_, txHash)
|
||||
if popped != nil {
|
||||
|
@@ -6,6 +6,7 @@ import (
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
mrand "math/rand"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
@@ -18,6 +19,7 @@ import (
|
||||
|
||||
"github.com/tendermint/tendermint/abci/example/counter"
|
||||
"github.com/tendermint/tendermint/abci/example/kvstore"
|
||||
abciserver "github.com/tendermint/tendermint/abci/server"
|
||||
abci "github.com/tendermint/tendermint/abci/types"
|
||||
cfg "github.com/tendermint/tendermint/config"
|
||||
cmn "github.com/tendermint/tendermint/libs/common"
|
||||
@@ -510,6 +512,54 @@ func TestMempoolTxsBytes(t *testing.T) {
|
||||
assert.EqualValues(t, 0, mempool.TxsBytes())
|
||||
}
|
||||
|
||||
// This will non-deterministically catch some concurrency failures like
|
||||
// https://github.com/tendermint/tendermint/issues/3509
|
||||
// TODO: all of the tests should probably also run using the remote proxy app
|
||||
// since otherwise we're not actually testing the concurrency of the mempool here!
|
||||
func TestMempoolRemoteAppConcurrency(t *testing.T) {
|
||||
sockPath := fmt.Sprintf("unix:///tmp/echo_%v.sock", cmn.RandStr(6))
|
||||
app := kvstore.NewKVStoreApplication()
|
||||
cc, server := newRemoteApp(t, sockPath, app)
|
||||
defer server.Stop()
|
||||
config := cfg.ResetTestRoot("mempool_test")
|
||||
mempool, cleanup := newMempoolWithAppAndConfig(cc, config)
|
||||
defer cleanup()
|
||||
|
||||
// generate small number of txs
|
||||
nTxs := 10
|
||||
txLen := 200
|
||||
txs := make([]types.Tx, nTxs)
|
||||
for i := 0; i < nTxs; i++ {
|
||||
txs[i] = cmn.RandBytes(txLen)
|
||||
}
|
||||
|
||||
// simulate a group of peers sending them over and over
|
||||
N := config.Mempool.Size
|
||||
maxPeers := 5
|
||||
for i := 0; i < N; i++ {
|
||||
peerID := mrand.Intn(maxPeers)
|
||||
txNum := mrand.Intn(nTxs)
|
||||
tx := txs[int(txNum)]
|
||||
|
||||
// this will err with ErrTxInCache many times ...
|
||||
mempool.CheckTxWithInfo(tx, nil, TxInfo{PeerID: uint16(peerID)})
|
||||
}
|
||||
err := mempool.FlushAppConn()
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
// caller must close server
|
||||
func newRemoteApp(t *testing.T, addr string, app abci.Application) (clientCreator proxy.ClientCreator, server cmn.Service) {
|
||||
clientCreator = proxy.NewRemoteClientCreator(addr, "socket", true)
|
||||
|
||||
// Start server
|
||||
server = abciserver.NewSocketServer(addr, app)
|
||||
server.SetLogger(log.TestingLogger().With("module", "abci-server"))
|
||||
if err := server.Start(); err != nil {
|
||||
t.Fatalf("Error starting socket server: %v", err.Error())
|
||||
}
|
||||
return clientCreator, server
|
||||
}
|
||||
func checksumIt(data []byte) string {
|
||||
h := sha256.New()
|
||||
h.Write(data)
|
||||
|
@@ -489,7 +489,7 @@ func NewNode(config *cfg.Config,
|
||||
addrBook := pex.NewAddrBook(config.P2P.AddrBookFile(), config.P2P.AddrBookStrict)
|
||||
|
||||
// Add ourselves to addrbook to prevent dialing ourselves
|
||||
addrBook.AddOurAddress(nodeInfo.NetAddress())
|
||||
addrBook.AddOurAddress(sw.NetAddress())
|
||||
|
||||
addrBook.SetLogger(p2pLogger.With("book", config.P2P.AddrBookFile()))
|
||||
if config.P2P.PexReactor {
|
||||
|
@@ -62,7 +62,7 @@ func (mp *Peer) Get(key string) interface{} {
|
||||
func (mp *Peer) Set(key string, value interface{}) {
|
||||
mp.kv[key] = value
|
||||
}
|
||||
func (mp *Peer) RemoteIP() net.IP { return mp.ip }
|
||||
func (mp *Peer) OriginalAddr() *p2p.NetAddress { return mp.addr }
|
||||
func (mp *Peer) RemoteAddr() net.Addr { return &net.TCPAddr{IP: mp.ip, Port: 8800} }
|
||||
func (mp *Peer) CloseConn() error { return nil }
|
||||
func (mp *Peer) RemoteIP() net.IP { return mp.ip }
|
||||
func (mp *Peer) SocketAddr() *p2p.NetAddress { return mp.addr }
|
||||
func (mp *Peer) RemoteAddr() net.Addr { return &net.TCPAddr{IP: mp.ip, Port: 8800} }
|
||||
func (mp *Peer) CloseConn() error { return nil }
|
||||
|
@@ -23,14 +23,8 @@ func MaxNodeInfoSize() int {
|
||||
// NodeInfo exposes basic info of a node
|
||||
// and determines if we're compatible.
|
||||
type NodeInfo interface {
|
||||
nodeInfoAddress
|
||||
nodeInfoTransport
|
||||
}
|
||||
|
||||
// nodeInfoAddress exposes just the core info of a node.
|
||||
type nodeInfoAddress interface {
|
||||
ID() ID
|
||||
NetAddress() *NetAddress
|
||||
nodeInfoTransport
|
||||
}
|
||||
|
||||
// nodeInfoTransport validates a nodeInfo and checks
|
||||
@@ -221,7 +215,7 @@ func (info DefaultNodeInfo) NetAddress() *NetAddress {
|
||||
if err != nil {
|
||||
switch err.(type) {
|
||||
case ErrNetAddressLookup:
|
||||
// XXX If the peer provided a host name and the lookup fails here
|
||||
// XXX If the peer provided a host name and the lookup fails here
|
||||
// we're out of luck.
|
||||
// TODO: use a NetAddress in DefaultNodeInfo
|
||||
default:
|
||||
|
27
p2p/peer.go
27
p2p/peer.go
@@ -29,7 +29,7 @@ type Peer interface {
|
||||
|
||||
NodeInfo() NodeInfo // peer's info
|
||||
Status() tmconn.ConnectionStatus
|
||||
OriginalAddr() *NetAddress // original address for outbound peers
|
||||
SocketAddr() *NetAddress // actual address of the socket
|
||||
|
||||
Send(byte, []byte) bool
|
||||
TrySend(byte, []byte) bool
|
||||
@@ -46,7 +46,7 @@ type peerConn struct {
|
||||
persistent bool
|
||||
conn net.Conn // source connection
|
||||
|
||||
originalAddr *NetAddress // nil for inbound connections
|
||||
socketAddr *NetAddress
|
||||
|
||||
// cached RemoteIP()
|
||||
ip net.IP
|
||||
@@ -55,14 +55,14 @@ type peerConn struct {
|
||||
func newPeerConn(
|
||||
outbound, persistent bool,
|
||||
conn net.Conn,
|
||||
originalAddr *NetAddress,
|
||||
socketAddr *NetAddress,
|
||||
) peerConn {
|
||||
|
||||
return peerConn{
|
||||
outbound: outbound,
|
||||
persistent: persistent,
|
||||
conn: conn,
|
||||
originalAddr: originalAddr,
|
||||
outbound: outbound,
|
||||
persistent: persistent,
|
||||
conn: conn,
|
||||
socketAddr: socketAddr,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -223,13 +223,12 @@ func (p *peer) NodeInfo() NodeInfo {
|
||||
return p.nodeInfo
|
||||
}
|
||||
|
||||
// OriginalAddr returns the original address, which was used to connect with
|
||||
// the peer. Returns nil for inbound peers.
|
||||
func (p *peer) OriginalAddr() *NetAddress {
|
||||
if p.peerConn.outbound {
|
||||
return p.peerConn.originalAddr
|
||||
}
|
||||
return nil
|
||||
// SocketAddr returns the address of the socket.
|
||||
// For outbound peers, it's the address dialed (after DNS resolution).
|
||||
// For inbound peers, it's the address returned by the underlying connection
|
||||
// (not what's reported in the peer's NodeInfo).
|
||||
func (p *peer) SocketAddr() *NetAddress {
|
||||
return p.peerConn.socketAddr
|
||||
}
|
||||
|
||||
// Status returns the peer's ConnectionStatus.
|
||||
|
@@ -29,7 +29,7 @@ func (mp *mockPeer) IsPersistent() bool { return true }
|
||||
func (mp *mockPeer) Get(s string) interface{} { return s }
|
||||
func (mp *mockPeer) Set(string, interface{}) {}
|
||||
func (mp *mockPeer) RemoteIP() net.IP { return mp.ip }
|
||||
func (mp *mockPeer) OriginalAddr() *NetAddress { return nil }
|
||||
func (mp *mockPeer) SocketAddr() *NetAddress { return nil }
|
||||
func (mp *mockPeer) RemoteAddr() net.Addr { return &net.TCPAddr{IP: mp.ip, Port: 8800} }
|
||||
func (mp *mockPeer) CloseConn() error { return nil }
|
||||
|
||||
|
@@ -109,25 +109,27 @@ func testOutboundPeerConn(
|
||||
persistent bool,
|
||||
ourNodePrivKey crypto.PrivKey,
|
||||
) (peerConn, error) {
|
||||
|
||||
var pc peerConn
|
||||
conn, err := testDial(addr, config)
|
||||
if err != nil {
|
||||
return peerConn{}, cmn.ErrorWrap(err, "Error creating peer")
|
||||
return pc, cmn.ErrorWrap(err, "Error creating peer")
|
||||
}
|
||||
|
||||
pc, err := testPeerConn(conn, config, true, persistent, ourNodePrivKey, addr)
|
||||
pc, err = testPeerConn(conn, config, true, persistent, ourNodePrivKey, addr)
|
||||
if err != nil {
|
||||
if cerr := conn.Close(); cerr != nil {
|
||||
return peerConn{}, cmn.ErrorWrap(err, cerr.Error())
|
||||
return pc, cmn.ErrorWrap(err, cerr.Error())
|
||||
}
|
||||
return peerConn{}, err
|
||||
return pc, err
|
||||
}
|
||||
|
||||
// ensure dialed ID matches connection ID
|
||||
if addr.ID != pc.ID() {
|
||||
if cerr := conn.Close(); cerr != nil {
|
||||
return peerConn{}, cmn.ErrorWrap(err, cerr.Error())
|
||||
return pc, cmn.ErrorWrap(err, cerr.Error())
|
||||
}
|
||||
return peerConn{}, ErrSwitchAuthenticationFailure{addr, pc.ID()}
|
||||
return pc, ErrSwitchAuthenticationFailure{addr, pc.ID()}
|
||||
}
|
||||
|
||||
return pc, nil
|
||||
|
@@ -167,7 +167,7 @@ func (r *PEXReactor) AddPeer(p Peer) {
|
||||
}
|
||||
} else {
|
||||
// inbound peer is its own source
|
||||
addr := p.NodeInfo().NetAddress()
|
||||
addr := p.SocketAddr()
|
||||
src := addr
|
||||
|
||||
// add to book. dont RequestAddrs right away because
|
||||
@@ -309,7 +309,7 @@ func (r *PEXReactor) ReceiveAddrs(addrs []*p2p.NetAddress, src Peer) error {
|
||||
}
|
||||
r.requestsSent.Delete(id)
|
||||
|
||||
srcAddr := src.NodeInfo().NetAddress()
|
||||
srcAddr := src.SocketAddr()
|
||||
for _, netAddr := range addrs {
|
||||
// Validate netAddr. Disconnect from a peer if it sends us invalid data.
|
||||
if netAddr == nil {
|
||||
|
@@ -96,7 +96,7 @@ func TestPEXReactorRunning(t *testing.T) {
|
||||
}
|
||||
|
||||
addOtherNodeAddrToAddrBook := func(switchIndex, otherSwitchIndex int) {
|
||||
addr := switches[otherSwitchIndex].NodeInfo().NetAddress()
|
||||
addr := switches[otherSwitchIndex].NetAddress()
|
||||
books[switchIndex].AddAddress(addr, addr)
|
||||
}
|
||||
|
||||
@@ -127,7 +127,7 @@ func TestPEXReactorReceive(t *testing.T) {
|
||||
r.RequestAddrs(peer)
|
||||
|
||||
size := book.Size()
|
||||
addrs := []*p2p.NetAddress{peer.NodeInfo().NetAddress()}
|
||||
addrs := []*p2p.NetAddress{peer.SocketAddr()}
|
||||
msg := cdc.MustMarshalBinaryBare(&pexAddrsMessage{Addrs: addrs})
|
||||
r.Receive(PexChannel, peer, msg)
|
||||
assert.Equal(t, size+1, book.Size())
|
||||
@@ -184,7 +184,7 @@ func TestPEXReactorAddrsMessageAbuse(t *testing.T) {
|
||||
assert.True(t, r.requestsSent.Has(id))
|
||||
assert.True(t, sw.Peers().Has(peer.ID()))
|
||||
|
||||
addrs := []*p2p.NetAddress{peer.NodeInfo().NetAddress()}
|
||||
addrs := []*p2p.NetAddress{peer.SocketAddr()}
|
||||
msg := cdc.MustMarshalBinaryBare(&pexAddrsMessage{Addrs: addrs})
|
||||
|
||||
// receive some addrs. should clear the request
|
||||
@@ -229,7 +229,7 @@ func TestCheckSeeds(t *testing.T) {
|
||||
badPeerConfig = &PEXReactorConfig{
|
||||
Seeds: []string{"ed3dfd27bfc4af18f67a49862f04cc100696e84d@bad.network.addr:26657",
|
||||
"d824b13cb5d40fa1d8a614e089357c7eff31b670@anotherbad.network.addr:26657",
|
||||
seed.NodeInfo().NetAddress().String()},
|
||||
seed.NetAddress().String()},
|
||||
}
|
||||
peer = testCreatePeerWithConfig(dir, 2, badPeerConfig)
|
||||
require.Nil(t, peer.Start())
|
||||
@@ -263,12 +263,13 @@ func TestConnectionSpeedForPeerReceivedFromSeed(t *testing.T) {
|
||||
defer os.RemoveAll(dir) // nolint: errcheck
|
||||
|
||||
// 1. create peer
|
||||
peer := testCreateDefaultPeer(dir, 1)
|
||||
require.Nil(t, peer.Start())
|
||||
defer peer.Stop()
|
||||
peerSwitch := testCreateDefaultPeer(dir, 1)
|
||||
require.Nil(t, peerSwitch.Start())
|
||||
defer peerSwitch.Stop()
|
||||
|
||||
// 2. Create seed which knows about the peer
|
||||
seed := testCreateSeed(dir, 2, []*p2p.NetAddress{peer.NodeInfo().NetAddress()}, []*p2p.NetAddress{peer.NodeInfo().NetAddress()})
|
||||
peerAddr := peerSwitch.NetAddress()
|
||||
seed := testCreateSeed(dir, 2, []*p2p.NetAddress{peerAddr}, []*p2p.NetAddress{peerAddr})
|
||||
require.Nil(t, seed.Start())
|
||||
defer seed.Stop()
|
||||
|
||||
@@ -295,7 +296,7 @@ func TestPEXReactorCrawlStatus(t *testing.T) {
|
||||
// Create a peer, add it to the peer set and the addrbook.
|
||||
peer := p2p.CreateRandomPeer(false)
|
||||
p2p.AddPeerToSwitch(pexR.Switch, peer)
|
||||
addr1 := peer.NodeInfo().NetAddress()
|
||||
addr1 := peer.SocketAddr()
|
||||
pexR.book.AddAddress(addr1, addr1)
|
||||
|
||||
// Add a non-connected address to the book.
|
||||
@@ -359,7 +360,7 @@ func TestPEXReactorSeedModeFlushStop(t *testing.T) {
|
||||
reactor := switches[0].Reactors()["pex"].(*PEXReactor)
|
||||
peerID := switches[1].NodeInfo().ID()
|
||||
|
||||
err = switches[1].DialPeerWithAddress(switches[0].NodeInfo().NetAddress(), false)
|
||||
err = switches[1].DialPeerWithAddress(switches[0].NetAddress(), false)
|
||||
assert.NoError(t, err)
|
||||
|
||||
// sleep up to a second while waiting for the peer to send us a message.
|
||||
@@ -397,7 +398,7 @@ func TestPEXReactorDoesNotAddPrivatePeersToAddrBook(t *testing.T) {
|
||||
pexR.RequestAddrs(peer)
|
||||
|
||||
size := book.Size()
|
||||
addrs := []*p2p.NetAddress{peer.NodeInfo().NetAddress()}
|
||||
addrs := []*p2p.NetAddress{peer.SocketAddr()}
|
||||
msg := cdc.MustMarshalBinaryBare(&pexAddrsMessage{Addrs: addrs})
|
||||
pexR.Receive(PexChannel, peer, msg)
|
||||
assert.Equal(t, size, book.Size())
|
||||
@@ -414,7 +415,7 @@ func TestPEXReactorDialPeer(t *testing.T) {
|
||||
sw.SetAddrBook(book)
|
||||
|
||||
peer := mock.NewPeer(nil)
|
||||
addr := peer.NodeInfo().NetAddress()
|
||||
addr := peer.SocketAddr()
|
||||
|
||||
assert.Equal(t, 0, pexR.AttemptsToDial(addr))
|
||||
|
||||
@@ -547,7 +548,7 @@ func testCreateSeed(dir string, id int, knownAddrs, srcAddrs []*p2p.NetAddress)
|
||||
// Starting and stopping the peer is left to the caller
|
||||
func testCreatePeerWithSeed(dir string, id int, seed *p2p.Switch) *p2p.Switch {
|
||||
conf := &PEXReactorConfig{
|
||||
Seeds: []string{seed.NodeInfo().NetAddress().String()},
|
||||
Seeds: []string{seed.NetAddress().String()},
|
||||
}
|
||||
return testCreatePeerWithConfig(dir, id, conf)
|
||||
}
|
||||
|
@@ -86,6 +86,12 @@ type Switch struct {
|
||||
metrics *Metrics
|
||||
}
|
||||
|
||||
// NetAddress returns the address the switch is listening on.
|
||||
func (sw *Switch) NetAddress() *NetAddress {
|
||||
addr := sw.transport.NetAddress()
|
||||
return &addr
|
||||
}
|
||||
|
||||
// SwitchOption sets an optional parameter on the Switch.
|
||||
type SwitchOption func(*Switch)
|
||||
|
||||
@@ -289,13 +295,7 @@ func (sw *Switch) StopPeerForError(peer Peer, reason interface{}) {
|
||||
sw.stopAndRemovePeer(peer, reason)
|
||||
|
||||
if peer.IsPersistent() {
|
||||
addr := peer.OriginalAddr()
|
||||
if addr == nil {
|
||||
// FIXME: persistent peers can't be inbound right now.
|
||||
// self-reported address for inbound persistent peers
|
||||
addr = peer.NodeInfo().NetAddress()
|
||||
}
|
||||
go sw.reconnectToPeer(addr)
|
||||
go sw.reconnectToPeer(peer.SocketAddr())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -383,7 +383,7 @@ func (sw *Switch) SetAddrBook(addrBook AddrBook) {
|
||||
// like contributed to consensus.
|
||||
func (sw *Switch) MarkPeerAsGood(peer Peer) {
|
||||
if sw.addrBook != nil {
|
||||
sw.addrBook.MarkGood(peer.NodeInfo().NetAddress())
|
||||
sw.addrBook.MarkGood(peer.SocketAddr())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -400,7 +400,7 @@ func (sw *Switch) DialPeersAsync(addrBook AddrBook, peers []string, persistent b
|
||||
sw.Logger.Error("Error in peer's address", "err", err)
|
||||
}
|
||||
|
||||
ourAddr := sw.nodeInfo.NetAddress()
|
||||
ourAddr := sw.NetAddress()
|
||||
|
||||
// TODO: this code feels like it's in the wrong place.
|
||||
// The integration tests depend on the addrBook being saved
|
||||
@@ -536,7 +536,7 @@ func (sw *Switch) acceptRoutine() {
|
||||
if in >= sw.config.MaxNumInboundPeers {
|
||||
sw.Logger.Info(
|
||||
"Ignoring inbound connection: already have enough inbound peers",
|
||||
"address", p.NodeInfo().NetAddress().String(),
|
||||
"address", p.SocketAddr(),
|
||||
"have", in,
|
||||
"max", sw.config.MaxNumInboundPeers,
|
||||
)
|
||||
@@ -653,7 +653,7 @@ func (sw *Switch) addPeer(p Peer) error {
|
||||
return err
|
||||
}
|
||||
|
||||
p.SetLogger(sw.Logger.With("peer", p.NodeInfo().NetAddress()))
|
||||
p.SetLogger(sw.Logger.With("peer", p.SocketAddr()))
|
||||
|
||||
// Handle the shut down case where the switch has stopped but we're
|
||||
// concurrently trying to add a peer.
|
||||
|
@@ -161,10 +161,6 @@ func assertMsgReceivedWithTimeout(t *testing.T, msgBytes []byte, channel byte, r
|
||||
|
||||
func TestSwitchFiltersOutItself(t *testing.T) {
|
||||
s1 := MakeSwitch(cfg, 1, "127.0.0.1", "123.123.123", initSwitchFunc)
|
||||
// addr := s1.NodeInfo().NetAddress()
|
||||
|
||||
// // add ourselves like we do in node.go#427
|
||||
// s1.addrBook.AddOurAddress(addr)
|
||||
|
||||
// simulate s1 having a public IP by creating a remote peer with the same ID
|
||||
rp := &remotePeer{PrivKey: s1.nodeKey.PrivKey, Config: cfg}
|
||||
@@ -498,7 +494,7 @@ func TestSwitchAcceptRoutine(t *testing.T) {
|
||||
rp := &remotePeer{PrivKey: ed25519.GenPrivKey(), Config: cfg}
|
||||
remotePeers = append(remotePeers, rp)
|
||||
rp.Start()
|
||||
c, err := rp.Dial(sw.NodeInfo().NetAddress())
|
||||
c, err := rp.Dial(sw.NetAddress())
|
||||
require.NoError(t, err)
|
||||
// spawn a reading routine to prevent connection from closing
|
||||
go func(c net.Conn) {
|
||||
@@ -517,7 +513,7 @@ func TestSwitchAcceptRoutine(t *testing.T) {
|
||||
// 2. check we close new connections if we already have MaxNumInboundPeers peers
|
||||
rp := &remotePeer{PrivKey: ed25519.GenPrivKey(), Config: cfg}
|
||||
rp.Start()
|
||||
conn, err := rp.Dial(sw.NodeInfo().NetAddress())
|
||||
conn, err := rp.Dial(sw.NetAddress())
|
||||
require.NoError(t, err)
|
||||
// check conn is closed
|
||||
one := make([]byte, 1)
|
||||
@@ -537,6 +533,10 @@ type errorTransport struct {
|
||||
acceptErr error
|
||||
}
|
||||
|
||||
func (et errorTransport) NetAddress() NetAddress {
|
||||
panic("not implemented")
|
||||
}
|
||||
|
||||
func (et errorTransport) Accept(c peerConfig) (Peer, error) {
|
||||
return nil, et.acceptErr
|
||||
}
|
||||
|
@@ -35,7 +35,8 @@ func CreateRandomPeer(outbound bool) *peer {
|
||||
addr, netAddr := CreateRoutableAddr()
|
||||
p := &peer{
|
||||
peerConn: peerConn{
|
||||
outbound: outbound,
|
||||
outbound: outbound,
|
||||
socketAddr: netAddr,
|
||||
},
|
||||
nodeInfo: mockNodeInfo{netAddr},
|
||||
mconn: &conn.MConnection{},
|
||||
@@ -174,10 +175,15 @@ func MakeSwitch(
|
||||
PrivKey: ed25519.GenPrivKey(),
|
||||
}
|
||||
nodeInfo := testNodeInfo(nodeKey.ID(), fmt.Sprintf("node%d", i))
|
||||
addr, err := NewNetAddressString(
|
||||
IDAddressString(nodeKey.ID(), nodeInfo.(DefaultNodeInfo).ListenAddr),
|
||||
)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
t := NewMultiplexTransport(nodeInfo, nodeKey, MConnConfig(cfg))
|
||||
|
||||
addr := nodeInfo.NetAddress()
|
||||
if err := t.Listen(*addr); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
@@ -214,7 +220,7 @@ func testPeerConn(
|
||||
cfg *config.P2PConfig,
|
||||
outbound, persistent bool,
|
||||
ourNodePrivKey crypto.PrivKey,
|
||||
originalAddr *NetAddress,
|
||||
socketAddr *NetAddress,
|
||||
) (pc peerConn, err error) {
|
||||
conn := rawConn
|
||||
|
||||
@@ -231,12 +237,7 @@ func testPeerConn(
|
||||
}
|
||||
|
||||
// Only the information we already have
|
||||
return peerConn{
|
||||
outbound: outbound,
|
||||
persistent: persistent,
|
||||
conn: conn,
|
||||
originalAddr: originalAddr,
|
||||
}, nil
|
||||
return newPeerConn(outbound, persistent, conn, socketAddr), nil
|
||||
}
|
||||
|
||||
//----------------------------------------------------------------
|
||||
|
@@ -24,6 +24,7 @@ type IPResolver interface {
|
||||
// accept is the container to carry the upgraded connection and NodeInfo from an
|
||||
// asynchronously running routine to the Accept method.
|
||||
type accept struct {
|
||||
netAddr *NetAddress
|
||||
conn net.Conn
|
||||
nodeInfo NodeInfo
|
||||
err error
|
||||
@@ -47,6 +48,9 @@ type peerConfig struct {
|
||||
// the transport. Each transport is also responsible to filter establishing
|
||||
// peers specific to its domain.
|
||||
type Transport interface {
|
||||
// Listening address.
|
||||
NetAddress() NetAddress
|
||||
|
||||
// Accept returns a newly connected Peer.
|
||||
Accept(peerConfig) (Peer, error)
|
||||
|
||||
@@ -115,6 +119,7 @@ func MultiplexTransportResolver(resolver IPResolver) MultiplexTransportOption {
|
||||
// MultiplexTransport accepts and dials tcp connections and upgrades them to
|
||||
// multiplexed peers.
|
||||
type MultiplexTransport struct {
|
||||
netAddr NetAddress
|
||||
listener net.Listener
|
||||
|
||||
acceptc chan accept
|
||||
@@ -161,6 +166,11 @@ func NewMultiplexTransport(
|
||||
}
|
||||
}
|
||||
|
||||
// NetAddress implements Transport.
|
||||
func (mt *MultiplexTransport) NetAddress() NetAddress {
|
||||
return mt.netAddr
|
||||
}
|
||||
|
||||
// Accept implements Transport.
|
||||
func (mt *MultiplexTransport) Accept(cfg peerConfig) (Peer, error) {
|
||||
select {
|
||||
@@ -173,7 +183,7 @@ func (mt *MultiplexTransport) Accept(cfg peerConfig) (Peer, error) {
|
||||
|
||||
cfg.outbound = false
|
||||
|
||||
return mt.wrapPeer(a.conn, a.nodeInfo, cfg, nil), nil
|
||||
return mt.wrapPeer(a.conn, a.nodeInfo, cfg, a.netAddr), nil
|
||||
case <-mt.closec:
|
||||
return nil, ErrTransportClosed{}
|
||||
}
|
||||
@@ -224,6 +234,7 @@ func (mt *MultiplexTransport) Listen(addr NetAddress) error {
|
||||
return err
|
||||
}
|
||||
|
||||
mt.netAddr = addr
|
||||
mt.listener = ln
|
||||
|
||||
go mt.acceptPeers()
|
||||
@@ -258,15 +269,21 @@ func (mt *MultiplexTransport) acceptPeers() {
|
||||
var (
|
||||
nodeInfo NodeInfo
|
||||
secretConn *conn.SecretConnection
|
||||
netAddr *NetAddress
|
||||
)
|
||||
|
||||
err := mt.filterConn(c)
|
||||
if err == nil {
|
||||
secretConn, nodeInfo, err = mt.upgrade(c, nil)
|
||||
if err == nil {
|
||||
addr := c.RemoteAddr()
|
||||
id := PubKeyToID(secretConn.RemotePubKey())
|
||||
netAddr = NewNetAddress(id, addr)
|
||||
}
|
||||
}
|
||||
|
||||
select {
|
||||
case mt.acceptc <- accept{secretConn, nodeInfo, err}:
|
||||
case mt.acceptc <- accept{netAddr, secretConn, nodeInfo, err}:
|
||||
// Make the upgraded peer available.
|
||||
case <-mt.closec:
|
||||
// Give up if the transport was closed.
|
||||
@@ -426,14 +443,14 @@ func (mt *MultiplexTransport) wrapPeer(
|
||||
c net.Conn,
|
||||
ni NodeInfo,
|
||||
cfg peerConfig,
|
||||
dialedAddr *NetAddress,
|
||||
socketAddr *NetAddress,
|
||||
) Peer {
|
||||
|
||||
peerConn := newPeerConn(
|
||||
cfg.outbound,
|
||||
cfg.persistent,
|
||||
c,
|
||||
dialedAddr,
|
||||
socketAddr,
|
||||
)
|
||||
|
||||
p := newPeer(
|
||||
|
@@ -8,6 +8,8 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/tendermint/tendermint/crypto/ed25519"
|
||||
"github.com/tendermint/tendermint/p2p/conn"
|
||||
)
|
||||
@@ -142,43 +144,23 @@ func TestTransportMultiplexConnFilterTimeout(t *testing.T) {
|
||||
|
||||
func TestTransportMultiplexAcceptMultiple(t *testing.T) {
|
||||
mt := testSetupMultiplexTransport(t)
|
||||
id, addr := mt.nodeKey.ID(), mt.listener.Addr().String()
|
||||
laddr, err := NewNetAddressStringWithOptionalID(IDAddressString(id, addr))
|
||||
require.NoError(t, err)
|
||||
|
||||
var (
|
||||
seed = rand.New(rand.NewSource(time.Now().UnixNano()))
|
||||
errc = make(chan error, seed.Intn(64)+64)
|
||||
seed = rand.New(rand.NewSource(time.Now().UnixNano()))
|
||||
nDialers = seed.Intn(64) + 64
|
||||
errc = make(chan error, nDialers)
|
||||
)
|
||||
|
||||
// Setup dialers.
|
||||
for i := 0; i < cap(errc); i++ {
|
||||
go func() {
|
||||
var (
|
||||
pv = ed25519.GenPrivKey()
|
||||
dialer = newMultiplexTransport(
|
||||
testNodeInfo(PubKeyToID(pv.PubKey()), defaultNodeName),
|
||||
NodeKey{
|
||||
PrivKey: pv,
|
||||
},
|
||||
)
|
||||
)
|
||||
addr, err := NewNetAddressStringWithOptionalID(IDAddressString(mt.nodeKey.ID(), mt.listener.Addr().String()))
|
||||
if err != nil {
|
||||
errc <- err
|
||||
return
|
||||
}
|
||||
|
||||
_, err = dialer.Dial(*addr, peerConfig{})
|
||||
if err != nil {
|
||||
errc <- err
|
||||
return
|
||||
}
|
||||
|
||||
// Signal that the connection was established.
|
||||
errc <- nil
|
||||
}()
|
||||
for i := 0; i < nDialers; i++ {
|
||||
go testDialer(*laddr, errc)
|
||||
}
|
||||
|
||||
// Catch connection errors.
|
||||
for i := 0; i < cap(errc); i++ {
|
||||
for i := 0; i < nDialers; i++ {
|
||||
if err := <-errc; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -216,6 +198,27 @@ func TestTransportMultiplexAcceptMultiple(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func testDialer(dialAddr NetAddress, errc chan error) {
|
||||
var (
|
||||
pv = ed25519.GenPrivKey()
|
||||
dialer = newMultiplexTransport(
|
||||
testNodeInfo(PubKeyToID(pv.PubKey()), defaultNodeName),
|
||||
NodeKey{
|
||||
PrivKey: pv,
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
_, err := dialer.Dial(dialAddr, peerConfig{})
|
||||
if err != nil {
|
||||
errc <- err
|
||||
return
|
||||
}
|
||||
|
||||
// Signal that the connection was established.
|
||||
errc <- nil
|
||||
}
|
||||
|
||||
func TestTransportMultiplexAcceptNonBlocking(t *testing.T) {
|
||||
mt := testSetupMultiplexTransport(t)
|
||||
|
||||
@@ -591,6 +594,7 @@ func TestTransportHandshake(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// create listener
|
||||
func testSetupMultiplexTransport(t *testing.T) *MultiplexTransport {
|
||||
var (
|
||||
pv = ed25519.GenPrivKey()
|
||||
|
@@ -218,7 +218,7 @@ func DumpConsensusState(ctx *rpctypes.Context) (*ctypes.ResultDumpConsensusState
|
||||
}
|
||||
peerStates[i] = ctypes.PeerStateInfo{
|
||||
// Peer basic info.
|
||||
NodeAddress: peer.NodeInfo().NetAddress().String(),
|
||||
NodeAddress: peer.SocketAddr().String(),
|
||||
// Peer consensus state.
|
||||
PeerState: peerStateJSON,
|
||||
}
|
||||
|
@@ -29,10 +29,10 @@ def request(org, repo, data):
|
||||
return json.loads(responsedata)
|
||||
|
||||
|
||||
def create_draft(org,repo,version):
|
||||
def create_draft(org,repo,branch,version):
|
||||
draft = {
|
||||
'tag_name': version,
|
||||
'target_commitish': 'master',
|
||||
'target_commitish': '{0}'.format(branch),
|
||||
'name': '{0} (WARNING: ALPHA SOFTWARE)'.format(version),
|
||||
'body': '<a href=https://github.com/{0}/{1}/blob/master/CHANGELOG.md#{2}>https://github.com/{0}/{1}/blob/master/CHANGELOG.md#{2}</a>'.format(org,repo,version.replace('v','').replace('.','')),
|
||||
'draft': True,
|
||||
@@ -45,6 +45,7 @@ if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--org", default="tendermint", help="GitHub organization")
|
||||
parser.add_argument("--repo", default="tendermint", help="GitHub repository")
|
||||
parser.add_argument("--branch", default=os.environ.get('CIRCLE_BRANCH'), help="Branch to build from, e.g.: v1.0")
|
||||
parser.add_argument("--version", default=os.environ.get('CIRCLE_TAG'), help="Version number for binary, e.g.: v1.0.0")
|
||||
args = parser.parse_args()
|
||||
|
||||
@@ -54,7 +55,7 @@ if __name__ == "__main__":
|
||||
if not os.environ.has_key('GITHUB_TOKEN'):
|
||||
raise parser.error('environment variable GITHUB_TOKEN is required')
|
||||
|
||||
release = create_draft(args.org,args.repo,args.version)
|
||||
release = create_draft(args.org,args.repo,args.branch,args.version)
|
||||
|
||||
print(release["id"])
|
||||
|
||||
|
@@ -20,7 +20,7 @@ const (
|
||||
// Must be a string because scripts like dist.sh read this file.
|
||||
// XXX: Don't change the name of this variable or you will break
|
||||
// automation :)
|
||||
TMCoreSemVer = "0.31.1"
|
||||
TMCoreSemVer = "0.31.3"
|
||||
|
||||
// ABCISemVer is the semantic version of the ABCI library
|
||||
ABCISemVer = "0.16.0"
|
||||
|
Reference in New Issue
Block a user