Compare commits

...

5 Commits

Author SHA1 Message Date
lx
fd58e6f0ac Merge pull request #1737 from bnb-chain/develop
release: draft release v1.2.8
2023-06-27 14:08:04 +08:00
lx
dda8e8da6a release: prepare for release v1.2.8 (#1736) 2023-06-27 13:01:58 +08:00
lmittmann
291cb8ab51 eth/filters, ethclient/gethclient: add fullTx option to pending tx filter (#25186) (#1626)
This PR adds a way to subscribe to the _full_ pending transactions, as opposed to just being notified about hashes.

In use cases where client subscribes to newPendingTransactions and gets txhashes only to then request the actual transaction, the caller can now shortcut that flow and obtain the transactions directly.

Co-authored-by: lx <92799281+brilliant-lx@users.noreply.github.com>
Co-authored-by: lmittmann <lmittmann@users.noreply.github.com>
Co-authored-by: Felix Lange <fjl@twurst.com>
2023-06-27 11:52:31 +08:00
NathanBSC
598c36bcea fix: avoid to block the chain when failed to send votes (#1734) 2023-06-27 08:59:08 +08:00
NathanBSC
69fd672ded feat: support password flag when handling bls keys (#1726) 2023-06-25 14:11:17 +08:00
9 changed files with 152 additions and 55 deletions

View File

@@ -1,4 +1,12 @@
# Changelog # Changelog
## v1.2.8
FEATURE
* [\#1626](https://github.com/bnb-chain/bsc/pull/1626) eth/filters, ethclient/gethclient: add fullTx option to pending tx filter
* [\#1726](https://github.com/bnb-chain/bsc/pull/1726) feat: support password flag when handling bls keys
BUGFIX
* [\#1734](https://github.com/bnb-chain/bsc/pull/1734) fix: avoid to block the chain when failed to send votes
## v1.2.7 ## v1.2.7
FEATURE FEATURE
* [\#1645](https://github.com/bnb-chain/bsc/pull/1645) lightclient: fix validator set change * [\#1645](https://github.com/bnb-chain/bsc/pull/1645) lightclient: fix validator set change

View File

@@ -46,6 +46,10 @@ var (
Name: "show-private-key", Name: "show-private-key",
Usage: "Show the BLS12-381 private key you will encrypt into a keystore file", Usage: "Show the BLS12-381 private key you will encrypt into a keystore file",
} }
BLSAccountPasswordFileFlag = cli.StringFlag{
Name: "blsaccountpassword",
Usage: "File path for the BLS account password, which contains the password to encrypt private key into keystore file for managing votes in fast_finality feature",
}
) )
var ( var (
@@ -87,6 +91,7 @@ or import a BLS account. The BLS wallet dir should be "<DATADIR>/bls/wallet".`,
Category: "BLS ACCOUNT COMMANDS", Category: "BLS ACCOUNT COMMANDS",
Flags: []cli.Flag{ Flags: []cli.Flag{
utils.DataDirFlag, utils.DataDirFlag,
utils.BLSPasswordFileFlag,
}, },
Description: ` Description: `
geth bls wallet create geth bls wallet create
@@ -128,6 +133,8 @@ Make sure you backup your BLS keys regularly.`,
utils.DataDirFlag, utils.DataDirFlag,
privateKeyFlag, privateKeyFlag,
showPrivateKeyFlag, showPrivateKeyFlag,
utils.BLSPasswordFileFlag,
BLSAccountPasswordFileFlag,
}, },
Description: ` Description: `
geth bls account new geth bls account new
@@ -147,6 +154,8 @@ You must remember this password to unlock your account in the future.`,
Category: "BLS ACCOUNT COMMANDS", Category: "BLS ACCOUNT COMMANDS",
Flags: []cli.Flag{ Flags: []cli.Flag{
utils.DataDirFlag, utils.DataDirFlag,
utils.BLSPasswordFileFlag,
BLSAccountPasswordFileFlag,
}, },
Description: ` Description: `
geth bls account import <keyFile> geth bls account import <keyFile>
@@ -163,6 +172,7 @@ If the BLS wallet not created yet, it will try to create BLS wallet first.`,
Category: "BLS ACCOUNT COMMANDS", Category: "BLS ACCOUNT COMMANDS",
Flags: []cli.Flag{ Flags: []cli.Flag{
utils.DataDirFlag, utils.DataDirFlag,
utils.BLSPasswordFileFlag,
}, },
Description: ` Description: `
geth bls account list geth bls account list
@@ -177,6 +187,7 @@ Print summary of existing BLS accounts in the current BLS wallet.`,
Category: "BLS ACCOUNT COMMANDS", Category: "BLS ACCOUNT COMMANDS",
Flags: []cli.Flag{ Flags: []cli.Flag{
utils.DataDirFlag, utils.DataDirFlag,
utils.BLSPasswordFileFlag,
}, },
Description: ` Description: `
geth bls account delete geth bls account delete
@@ -209,7 +220,7 @@ func blsWalletCreate(ctx *cli.Context) error {
utils.Fatalf("BLS wallet already exists in <DATADIR>/bls/wallet.") utils.Fatalf("BLS wallet already exists in <DATADIR>/bls/wallet.")
} }
password := utils.GetPassPhrase("Your new BLS wallet will be locked with a password. Please give a password. Do not forget this password.", true) password := utils.GetPassPhraseWithList("Your new BLS wallet will be locked with a password. Please give a password. Do not forget this password.", true, 0, GetBLSPassword(ctx))
opts := []accounts.Option{} opts := []accounts.Option{}
opts = append(opts, accounts.WithWalletDir(dir)) opts = append(opts, accounts.WithWalletDir(dir))
@@ -239,7 +250,7 @@ func openOrCreateBLSWallet(ctx *cli.Context, cfg *gethConfig) (*wallet.Wallet, e
} }
if !dirExists { if !dirExists {
fmt.Println("BLS wallet not exists, creating BLS wallet...") fmt.Println("BLS wallet not exists, creating BLS wallet...")
password := utils.GetPassPhrase("Your new BLS wallet will be locked with a password. Please give a password. Do not forget this password.", true) password := utils.GetPassPhraseWithList("Your new BLS wallet will be locked with a password. Please give a password. Do not forget this password.", true, 0, GetBLSPassword(ctx))
opts := []accounts.Option{} opts := []accounts.Option{}
opts = append(opts, accounts.WithWalletDir(walletDir)) opts = append(opts, accounts.WithWalletDir(walletDir))
@@ -259,7 +270,7 @@ func openOrCreateBLSWallet(ctx *cli.Context, cfg *gethConfig) (*wallet.Wallet, e
return w, nil return w, nil
} }
walletPassword := utils.GetPassPhrase("Enter the password for your BLS wallet.", false) walletPassword := utils.GetPassPhraseWithList("Enter the password for your BLS wallet.", false, 0, GetBLSPassword(ctx))
w, err = wallet.OpenWallet(context.Background(), &wallet.Config{ w, err = wallet.OpenWallet(context.Background(), &wallet.Config{
WalletDir: walletDir, WalletDir: walletDir,
WalletPassword: walletPassword, WalletPassword: walletPassword,
@@ -299,7 +310,7 @@ func blsAccountCreate(ctx *cli.Context) error {
if err := os.MkdirAll(keystoreDir, 0755); err != nil { if err := os.MkdirAll(keystoreDir, 0755); err != nil {
utils.Fatalf("Could not access keystore dir: %v.", err) utils.Fatalf("Could not access keystore dir: %v.", err)
} }
accountPassword := utils.GetPassPhrase("Your new BLS account will be encrypted with a password. Please give a password. Do not forget this password.", true) accountPassword := utils.GetPassPhraseWithList("Your new BLS account will be encrypted with a password. Please give a password. Do not forget this password.", true, 0, GetBLSAccountPassword(ctx))
if err := core.ValidatePasswordFormat(accountPassword); err != nil { if err := core.ValidatePasswordFormat(accountPassword); err != nil {
utils.Fatalf("Password invalid: %v.", err) utils.Fatalf("Password invalid: %v.", err)
} }
@@ -411,7 +422,8 @@ func blsAccountImport(ctx *cli.Context) error {
utils.Fatalf("The BLS keymanager cannot import keystores") utils.Fatalf("The BLS keymanager cannot import keystores")
} }
password := utils.GetPassPhrase("Enter the password for your imported account.", false) password := utils.GetPassPhraseWithList("Enter the password for your imported account.", false, 0, GetBLSAccountPassword(ctx))
fmt.Println("Importing BLS account, this may take a while...") fmt.Println("Importing BLS account, this may take a while...")
statuses, err := accounts.ImportAccounts(context.Background(), &accounts.ImportAccountsConfig{ statuses, err := accounts.ImportAccounts(context.Background(), &accounts.ImportAccountsConfig{
Importer: k, Importer: k,
@@ -447,7 +459,7 @@ func blsAccountList(ctx *cli.Context) error {
utils.Fatalf("BLS wallet not exists.") utils.Fatalf("BLS wallet not exists.")
} }
walletPassword := utils.GetPassPhrase("Enter the password for your BLS wallet.", false) walletPassword := utils.GetPassPhraseWithList("Enter the password for your BLS wallet.", false, 0, GetBLSPassword(ctx))
w, err := wallet.OpenWallet(context.Background(), &wallet.Config{ w, err := wallet.OpenWallet(context.Background(), &wallet.Config{
WalletDir: walletDir, WalletDir: walletDir,
WalletPassword: walletPassword, WalletPassword: walletPassword,
@@ -526,7 +538,7 @@ func blsAccountDelete(ctx *cli.Context) error {
utils.Fatalf("BLS wallet not exists.") utils.Fatalf("BLS wallet not exists.")
} }
walletPassword := utils.GetPassPhrase("Enter the password for your BLS wallet.", false) walletPassword := utils.GetPassPhraseWithList("Enter the password for your BLS wallet.", false, 0, GetBLSPassword(ctx))
w, err := wallet.OpenWallet(context.Background(), &wallet.Config{ w, err := wallet.OpenWallet(context.Background(), &wallet.Config{
WalletDir: walletDir, WalletDir: walletDir,
WalletPassword: walletPassword, WalletPassword: walletPassword,
@@ -587,3 +599,27 @@ func blsAccountDelete(ctx *cli.Context) error {
return nil return nil
} }
func GetBLSPassword(ctx *cli.Context) []string {
path := ctx.GlobalString(utils.BLSPasswordFileFlag.Name)
if path == "" {
return nil
}
text, err := ioutil.ReadFile(path)
if err != nil {
utils.Fatalf("Failed to read wallet password file: %v", err)
}
return []string{string(text)}
}
func GetBLSAccountPassword(ctx *cli.Context) []string {
path := ctx.String(BLSAccountPasswordFileFlag.Name)
if path == "" {
return nil
}
text, err := ioutil.ReadFile(path)
if err != nil {
utils.Fatalf("Failed to read account password file: %v", err)
}
return []string{string(text)}
}

View File

@@ -39,6 +39,7 @@ type filter struct {
typ Type typ Type
deadline *time.Timer // filter is inactiv when deadline triggers deadline *time.Timer // filter is inactiv when deadline triggers
hashes []common.Hash hashes []common.Hash
txs []*types.Transaction
crit FilterCriteria crit FilterCriteria
logs []*types.Log logs []*types.Log
s *Subscription // associated subscription in event system s *Subscription // associated subscription in event system
@@ -99,7 +100,7 @@ func (api *PublicFilterAPI) timeoutLoop(timeout time.Duration) {
} }
} }
// NewPendingTransactionFilter creates a filter that fetches pending transaction hashes // NewPendingTransactionFilter creates a filter that fetches pending transactions
// as transactions enter the pending state. // as transactions enter the pending state.
// //
// It is part of the filter package because this filter can be used through the // It is part of the filter package because this filter can be used through the
@@ -108,20 +109,20 @@ func (api *PublicFilterAPI) timeoutLoop(timeout time.Duration) {
// https://eth.wiki/json-rpc/API#eth_newpendingtransactionfilter // https://eth.wiki/json-rpc/API#eth_newpendingtransactionfilter
func (api *PublicFilterAPI) NewPendingTransactionFilter() rpc.ID { func (api *PublicFilterAPI) NewPendingTransactionFilter() rpc.ID {
var ( var (
pendingTxs = make(chan []common.Hash) pendingTxs = make(chan []*types.Transaction)
pendingTxSub = api.events.SubscribePendingTxs(pendingTxs) pendingTxSub = api.events.SubscribePendingTxs(pendingTxs)
) )
api.filtersMu.Lock() api.filtersMu.Lock()
api.filters[pendingTxSub.ID] = &filter{typ: PendingTransactionsSubscription, deadline: time.NewTimer(api.timeout), hashes: make([]common.Hash, 0), s: pendingTxSub} api.filters[pendingTxSub.ID] = &filter{typ: PendingTransactionsSubscription, deadline: time.NewTimer(api.timeout), txs: make([]*types.Transaction, 0), s: pendingTxSub}
api.filtersMu.Unlock() api.filtersMu.Unlock()
gopool.Submit(func() { gopool.Submit(func() {
for { for {
select { select {
case ph := <-pendingTxs: case pTx := <-pendingTxs:
api.filtersMu.Lock() api.filtersMu.Lock()
if f, found := api.filters[pendingTxSub.ID]; found { if f, found := api.filters[pendingTxSub.ID]; found {
f.hashes = append(f.hashes, ph...) f.txs = append(f.txs, pTx...)
} }
api.filtersMu.Unlock() api.filtersMu.Unlock()
case <-pendingTxSub.Err(): case <-pendingTxSub.Err():
@@ -136,9 +137,10 @@ func (api *PublicFilterAPI) NewPendingTransactionFilter() rpc.ID {
return pendingTxSub.ID return pendingTxSub.ID
} }
// NewPendingTransactions creates a subscription that is triggered each time a transaction // NewPendingTransactions creates a subscription that is triggered each time a
// enters the transaction pool and was signed from one of the transactions this nodes manages. // transaction enters the transaction pool. If fullTx is true the full tx is
func (api *PublicFilterAPI) NewPendingTransactions(ctx context.Context) (*rpc.Subscription, error) { // sent to the client, otherwise the hash is sent.
func (api *PublicFilterAPI) NewPendingTransactions(ctx context.Context, fullTx *bool) (*rpc.Subscription, error) {
notifier, supported := rpc.NotifierFromContext(ctx) notifier, supported := rpc.NotifierFromContext(ctx)
if !supported { if !supported {
return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
@@ -147,16 +149,20 @@ func (api *PublicFilterAPI) NewPendingTransactions(ctx context.Context) (*rpc.Su
rpcSub := notifier.CreateSubscription() rpcSub := notifier.CreateSubscription()
gopool.Submit(func() { gopool.Submit(func() {
txHashes := make(chan []common.Hash, 128) txs := make(chan []*types.Transaction, 128)
pendingTxSub := api.events.SubscribePendingTxs(txHashes) pendingTxSub := api.events.SubscribePendingTxs(txs)
for { for {
select { select {
case hashes := <-txHashes: case txs := <-txs:
// To keep the original behaviour, send a single tx hash in one notification. // To keep the original behaviour, send a single tx hash in one notification.
// TODO(rjl493456442) Send a batch of tx hashes in one notification // TODO(rjl493456442) Send a batch of tx hashes in one notification
for _, h := range hashes { for _, tx := range txs {
notifier.Notify(rpcSub.ID, h) if fullTx != nil && *fullTx {
notifier.Notify(rpcSub.ID, tx)
} else {
notifier.Notify(rpcSub.ID, tx.Hash())
}
} }
case <-rpcSub.Err(): case <-rpcSub.Err():
pendingTxSub.Unsubscribe() pendingTxSub.Unsubscribe()
@@ -551,10 +557,14 @@ func (api *PublicFilterAPI) GetFilterChanges(id rpc.ID) (interface{}, error) {
f.deadline.Reset(api.timeout) f.deadline.Reset(api.timeout)
switch f.typ { switch f.typ {
case PendingTransactionsSubscription, BlocksSubscription, FinalizedHeadersSubscription, VotesSubscription: case BlocksSubscription, FinalizedHeadersSubscription, VotesSubscription:
hashes := f.hashes hashes := f.hashes
f.hashes = nil f.hashes = nil
return returnHashes(hashes), nil return returnHashes(hashes), nil
case PendingTransactionsSubscription:
txs := f.txs
f.txs = nil
return txs, nil
case LogsSubscription, MinedAndPendingLogsSubscription: case LogsSubscription, MinedAndPendingLogsSubscription:
logs := f.logs logs := f.logs
f.logs = nil f.logs = nil

View File

@@ -47,8 +47,8 @@ const (
PendingLogsSubscription PendingLogsSubscription
// MinedAndPendingLogsSubscription queries for logs in mined and pending blocks. // MinedAndPendingLogsSubscription queries for logs in mined and pending blocks.
MinedAndPendingLogsSubscription MinedAndPendingLogsSubscription
// PendingTransactionsSubscription queries tx hashes for pending // PendingTransactionsSubscription queries for pending transactions entering
// transactions entering the pending state // the pending state
PendingTransactionsSubscription PendingTransactionsSubscription
// BlocksSubscription queries hashes for blocks that are imported // BlocksSubscription queries hashes for blocks that are imported
BlocksSubscription BlocksSubscription
@@ -83,7 +83,7 @@ type subscription struct {
created time.Time created time.Time
logsCrit ethereum.FilterQuery logsCrit ethereum.FilterQuery
logs chan []*types.Log logs chan []*types.Log
hashes chan []common.Hash txs chan []*types.Transaction
headers chan *types.Header headers chan *types.Header
finalizedHeaders chan *types.Header finalizedHeaders chan *types.Header
votes chan *types.VoteEnvelope votes chan *types.VoteEnvelope
@@ -187,7 +187,7 @@ func (sub *Subscription) Unsubscribe() {
case sub.es.uninstall <- sub.f: case sub.es.uninstall <- sub.f:
break uninstallLoop break uninstallLoop
case <-sub.f.logs: case <-sub.f.logs:
case <-sub.f.hashes: case <-sub.f.txs:
case <-sub.f.headers: case <-sub.f.headers:
case <-sub.f.votes: case <-sub.f.votes:
} }
@@ -255,7 +255,7 @@ func (es *EventSystem) subscribeMinedPendingLogs(crit ethereum.FilterQuery, logs
logsCrit: crit, logsCrit: crit,
created: time.Now(), created: time.Now(),
logs: logs, logs: logs,
hashes: make(chan []common.Hash), txs: make(chan []*types.Transaction),
headers: make(chan *types.Header), headers: make(chan *types.Header),
votes: make(chan *types.VoteEnvelope), votes: make(chan *types.VoteEnvelope),
installed: make(chan struct{}), installed: make(chan struct{}),
@@ -273,7 +273,7 @@ func (es *EventSystem) subscribeLogs(crit ethereum.FilterQuery, logs chan []*typ
logsCrit: crit, logsCrit: crit,
created: time.Now(), created: time.Now(),
logs: logs, logs: logs,
hashes: make(chan []common.Hash), txs: make(chan []*types.Transaction),
headers: make(chan *types.Header), headers: make(chan *types.Header),
votes: make(chan *types.VoteEnvelope), votes: make(chan *types.VoteEnvelope),
installed: make(chan struct{}), installed: make(chan struct{}),
@@ -291,7 +291,7 @@ func (es *EventSystem) subscribePendingLogs(crit ethereum.FilterQuery, logs chan
logsCrit: crit, logsCrit: crit,
created: time.Now(), created: time.Now(),
logs: logs, logs: logs,
hashes: make(chan []common.Hash), txs: make(chan []*types.Transaction),
headers: make(chan *types.Header), headers: make(chan *types.Header),
votes: make(chan *types.VoteEnvelope), votes: make(chan *types.VoteEnvelope),
installed: make(chan struct{}), installed: make(chan struct{}),
@@ -308,7 +308,7 @@ func (es *EventSystem) SubscribeNewHeads(headers chan *types.Header) *Subscripti
typ: BlocksSubscription, typ: BlocksSubscription,
created: time.Now(), created: time.Now(),
logs: make(chan []*types.Log), logs: make(chan []*types.Log),
hashes: make(chan []common.Hash), txs: make(chan []*types.Transaction),
headers: headers, headers: headers,
votes: make(chan *types.VoteEnvelope), votes: make(chan *types.VoteEnvelope),
installed: make(chan struct{}), installed: make(chan struct{}),
@@ -325,7 +325,7 @@ func (es *EventSystem) SubscribeNewFinalizedHeaders(headers chan *types.Header)
typ: FinalizedHeadersSubscription, typ: FinalizedHeadersSubscription,
created: time.Now(), created: time.Now(),
logs: make(chan []*types.Log), logs: make(chan []*types.Log),
hashes: make(chan []common.Hash), txs: make(chan []*types.Transaction),
headers: headers, headers: headers,
votes: make(chan *types.VoteEnvelope), votes: make(chan *types.VoteEnvelope),
installed: make(chan struct{}), installed: make(chan struct{}),
@@ -334,15 +334,15 @@ func (es *EventSystem) SubscribeNewFinalizedHeaders(headers chan *types.Header)
return es.subscribe(sub) return es.subscribe(sub)
} }
// SubscribePendingTxs creates a subscription that writes transaction hashes for // SubscribePendingTxs creates a subscription that writes transactions for
// transactions that enter the transaction pool. // transactions that enter the transaction pool.
func (es *EventSystem) SubscribePendingTxs(hashes chan []common.Hash) *Subscription { func (es *EventSystem) SubscribePendingTxs(txs chan []*types.Transaction) *Subscription {
sub := &subscription{ sub := &subscription{
id: rpc.NewID(), id: rpc.NewID(),
typ: PendingTransactionsSubscription, typ: PendingTransactionsSubscription,
created: time.Now(), created: time.Now(),
logs: make(chan []*types.Log), logs: make(chan []*types.Log),
hashes: hashes, txs: txs,
headers: make(chan *types.Header), headers: make(chan *types.Header),
votes: make(chan *types.VoteEnvelope), votes: make(chan *types.VoteEnvelope),
installed: make(chan struct{}), installed: make(chan struct{}),
@@ -359,7 +359,7 @@ func (es *EventSystem) SubscribeNewVotes(votes chan *types.VoteEnvelope) *Subscr
typ: VotesSubscription, typ: VotesSubscription,
created: time.Now(), created: time.Now(),
logs: make(chan []*types.Log), logs: make(chan []*types.Log),
hashes: make(chan []common.Hash), txs: make(chan []*types.Transaction),
headers: make(chan *types.Header), headers: make(chan *types.Header),
votes: votes, votes: votes,
installed: make(chan struct{}), installed: make(chan struct{}),
@@ -404,12 +404,8 @@ func (es *EventSystem) handleRemovedLogs(filters filterIndex, ev core.RemovedLog
} }
func (es *EventSystem) handleTxsEvent(filters filterIndex, ev core.NewTxsEvent) { func (es *EventSystem) handleTxsEvent(filters filterIndex, ev core.NewTxsEvent) {
hashes := make([]common.Hash, 0, len(ev.Txs))
for _, tx := range ev.Txs {
hashes = append(hashes, tx.Hash())
}
for _, f := range filters[PendingTransactionsSubscription] { for _, f := range filters[PendingTransactionsSubscription] {
f.hashes <- hashes f.txs <- ev.Txs
} }
} }

View File

@@ -250,7 +250,7 @@ func TestPendingTxFilter(t *testing.T) {
types.NewTransaction(4, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil), types.NewTransaction(4, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil),
} }
hashes []common.Hash txs []*types.Transaction
) )
fid0 := api.NewPendingTransactionFilter() fid0 := api.NewPendingTransactionFilter()
@@ -265,9 +265,9 @@ func TestPendingTxFilter(t *testing.T) {
t.Fatalf("Unable to retrieve logs: %v", err) t.Fatalf("Unable to retrieve logs: %v", err)
} }
h := results.([]common.Hash) tx := results.([]*types.Transaction)
hashes = append(hashes, h...) txs = append(txs, tx...)
if len(hashes) >= len(transactions) { if len(txs) >= len(transactions) {
break break
} }
// check timeout // check timeout
@@ -278,13 +278,13 @@ func TestPendingTxFilter(t *testing.T) {
time.Sleep(100 * time.Millisecond) time.Sleep(100 * time.Millisecond)
} }
if len(hashes) != len(transactions) { if len(txs) != len(transactions) {
t.Errorf("invalid number of transactions, want %d transactions(s), got %d", len(transactions), len(hashes)) t.Errorf("invalid number of transactions, want %d transactions(s), got %d", len(transactions), len(txs))
return return
} }
for i := range hashes { for i := range txs {
if hashes[i] != transactions[i].Hash() { if txs[i].Hash() != transactions[i].Hash() {
t.Errorf("hashes[%d] invalid, want %x, got %x", i, transactions[i].Hash(), hashes[i]) t.Errorf("hashes[%d] invalid, want %x, got %x", i, transactions[i].Hash(), txs[i].Hash())
} }
} }
} }
@@ -715,11 +715,11 @@ func TestPendingTxFilterDeadlock(t *testing.T) {
fids[i] = fid fids[i] = fid
// Wait for at least one tx to arrive in filter // Wait for at least one tx to arrive in filter
for { for {
hashes, err := api.GetFilterChanges(fid) txs, err := api.GetFilterChanges(fid)
if err != nil { if err != nil {
t.Fatalf("Filter should exist: %v\n", err) t.Fatalf("Filter should exist: %v\n", err)
} }
if len(hashes.([]common.Hash)) > 0 { if len(txs.([]*types.Transaction)) > 0 {
break break
} }
runtime.Gosched() runtime.Gosched()

View File

@@ -13,6 +13,9 @@ const (
// maxKnownVotes is the maximum vote hashes to keep in the known list // maxKnownVotes is the maximum vote hashes to keep in the known list
// before starting to randomly evict them. // before starting to randomly evict them.
maxKnownVotes = 5376 maxKnownVotes = 5376
// voteBufferSize is the maximum number of batch votes can be hold before sending
voteBufferSize = 21 * 2
) )
// max is a helper function which returns the larger of the two given integers. // max is a helper function which returns the larger of the two given integers.
@@ -43,7 +46,7 @@ func NewPeer(version uint, p *p2p.Peer, rw p2p.MsgReadWriter) *Peer {
peer := &Peer{ peer := &Peer{
id: id, id: id,
knownVotes: newKnownCache(maxKnownVotes), knownVotes: newKnownCache(maxKnownVotes),
voteBroadcast: make(chan []*types.VoteEnvelope), voteBroadcast: make(chan []*types.VoteEnvelope, voteBufferSize),
Peer: p, Peer: p,
rw: rw, rw: rw,
version: version, version: version,
@@ -105,7 +108,9 @@ func (p *Peer) AsyncSendVotes(votes []*types.VoteEnvelope) {
select { select {
case p.voteBroadcast <- votes: case p.voteBroadcast <- votes:
case <-p.term: case <-p.term:
p.Log().Debug("Dropping vote propagation", "count", len(votes)) p.Log().Debug("Dropping vote propagation for closed peer", "count", len(votes))
default:
p.Log().Debug("Dropping vote propagation for abnormal peer", "count", len(votes))
} }
} }

View File

@@ -175,7 +175,12 @@ func (ec *Client) GetNodeInfo(ctx context.Context) (*p2p.NodeInfo, error) {
return &result, err return &result, err
} }
// SubscribePendingTransactions subscribes to new pending transactions. // SubscribeFullPendingTransactions subscribes to new pending transactions.
func (ec *Client) SubscribeFullPendingTransactions(ctx context.Context, ch chan<- *types.Transaction) (*rpc.ClientSubscription, error) {
return ec.c.EthSubscribe(ctx, ch, "newPendingTransactions", true)
}
// SubscribePendingTransactions subscribes to new pending transaction hashes.
func (ec *Client) SubscribePendingTransactions(ctx context.Context, ch chan<- common.Hash) (*rpc.ClientSubscription, error) { func (ec *Client) SubscribePendingTransactions(ctx context.Context, ch chan<- common.Hash) (*rpc.ClientSubscription, error) {
return ec.c.EthSubscribe(ctx, ch, "newPendingTransactions") return ec.c.EthSubscribe(ctx, ch, "newPendingTransactions")
} }

View File

@@ -123,8 +123,11 @@ func TestGethClient(t *testing.T) {
"TestSetHead", "TestSetHead",
func(t *testing.T) { testSetHead(t, client) }, func(t *testing.T) { testSetHead(t, client) },
}, { }, {
"TestSubscribePendingTxs", "TestSubscribePendingTxHashes",
func(t *testing.T) { testSubscribePendingTransactions(t, client) }, func(t *testing.T) { testSubscribePendingTransactions(t, client) },
}, {
"TestSubscribePendingTxs",
func(t *testing.T) { testSubscribeFullPendingTransactions(t, client) },
}, { }, {
"TestCallContract", "TestCallContract",
func(t *testing.T) { testCallContract(t, client) }, func(t *testing.T) { testCallContract(t, client) },
@@ -298,6 +301,40 @@ func testSubscribePendingTransactions(t *testing.T, client *rpc.Client) {
} }
} }
func testSubscribeFullPendingTransactions(t *testing.T, client *rpc.Client) {
ec := New(client)
ethcl := ethclient.NewClient(client)
// Subscribe to Transactions
ch := make(chan *types.Transaction)
ec.SubscribeFullPendingTransactions(context.Background(), ch)
// Send a transaction
chainID, err := ethcl.ChainID(context.Background())
if err != nil {
t.Fatal(err)
}
// Create transaction
tx := types.NewTransaction(1, common.Address{1}, big.NewInt(1), 22000, big.NewInt(1), nil)
signer := types.LatestSignerForChainID(chainID)
signature, err := crypto.Sign(signer.Hash(tx).Bytes(), testKey)
if err != nil {
t.Fatal(err)
}
signedTx, err := tx.WithSignature(signer, signature)
if err != nil {
t.Fatal(err)
}
// Send transaction
err = ethcl.SendTransaction(context.Background(), signedTx)
if err != nil {
t.Fatal(err)
}
// Check that the transaction was send over the channel
tx = <-ch
if tx.Hash() != signedTx.Hash() {
t.Fatalf("Invalid tx hash received, got %v, want %v", tx.Hash(), signedTx.Hash())
}
}
func testCallContract(t *testing.T, client *rpc.Client) { func testCallContract(t *testing.T, client *rpc.Client) {
ec := New(client) ec := New(client)
msg := ethereum.CallMsg{ msg := ethereum.CallMsg{

View File

@@ -23,7 +23,7 @@ import (
const ( const (
VersionMajor = 1 // Major version component of the current release VersionMajor = 1 // Major version component of the current release
VersionMinor = 2 // Minor version component of the current release VersionMinor = 2 // Minor version component of the current release
VersionPatch = 7 // Patch version component of the current release VersionPatch = 8 // Patch version component of the current release
VersionMeta = "" // Version metadata to append to the version string VersionMeta = "" // Version metadata to append to the version string
) )