diff --git a/blockproducer/chain.go b/blockproducer/chain.go index b07768e23..ac737d573 100644 --- a/blockproducer/chain.go +++ b/blockproducer/chain.go @@ -311,7 +311,6 @@ func (c *Chain) produceBlock(now time.Time) (err error) { for _, s := range c.getPeers().Servers { if !s.IsEqual(&c.nodeID) { - // Bind NodeID to subroutine func(id proto.NodeID) { c.goFuncWithTimeout(func(ctx context.Context) { var ( @@ -336,7 +335,6 @@ func (c *Chain) produceBlock(now time.Time) (err error) { }(s) } } - return err } @@ -471,10 +469,31 @@ func (c *Chain) processBlocks(ctx context.Context) { } func (c *Chain) addTx(tx pi.Transaction) { - // Simple non-blocking broadcasting - for _, v := range c.getPeers().Servers { - if !v.IsEqual(&c.nodeID) { - // Bind NodeID to subroutine + select { + case c.pendingTxs <- tx: + case <-c.ctx.Done(): + log.WithError(c.ctx.Err()).Warn("add transaction aborted") + } +} + +func (c *Chain) processTx(tx pi.Transaction) { + if err := tx.Verify(); err != nil { + log.WithError(err).Errorf("failed to verify transaction with hash: %s, address: %s, tx type: %s", tx.Hash(), tx.GetAccountAddress(), tx.GetTransactionType().String()) + return + } + if ok := func() (ok bool) { + c.RLock() + defer c.RUnlock() + _, ok = c.txPool[tx.Hash()] + return + }(); ok { + log.WithFields(log.Fields{ + "tx_hash": tx.Hash().Short(4), + }).Debug("tx already exists, abort processing") + return + } + for _, s := range c.getPeers().Servers { + if !s.IsEqual(&c.nodeID) { func(id proto.NodeID) { c.goFuncWithTimeout(func(ctx context.Context) { var ( @@ -494,23 +513,10 @@ func (c *Chain) addTx(tx pi.Transaction) { "tx_hash": tx.Hash().Short(4), "tx_type": tx.GetTransactionType(), }).WithError(err).Debug("broadcasting transaction to other peers") - }, c.period) - }(v) + }, c.tick) + }(s) } } - - select { - case c.pendingTxs <- tx: - case <-c.ctx.Done(): - log.WithError(c.ctx.Err()).Warn("add transaction aborted") - } -} - -func (c *Chain) processTx(tx pi.Transaction) { - if err := tx.Verify(); err != nil { - log.WithError(err).Error("failed to verify transaction") - return - } if err := c.storeTx(tx); err != nil { log.WithError(err).Error("failed to add transaction") } @@ -562,25 +568,22 @@ func (c *Chain) mainCycle(ctx context.Context) { } func (c *Chain) syncCurrentHead(ctx context.Context) { + var h = c.getNextHeight() - 1 + if c.head().height >= h { + return + } + // Initiate blocking gossip calls to fetch block of the current height, + // with timeout of one tick. var ( - h = c.getNextHeight() - 1 - cld, ccl = context.WithTimeout(ctx, c.tick) wg = &sync.WaitGroup{} + cld, ccl = context.WithTimeout(ctx, c.tick) ) - defer func() { wg.Wait() ccl() }() - - if c.head().height >= h { - return - } - - // Initiate blocking gossip calls to fetch block of the current height, - // with timeout of one tick. - for _, v := range c.getPeers().Servers { - if !v.IsEqual(&c.nodeID) { + for _, s := range c.getPeers().Servers { + if !s.IsEqual(&c.nodeID) { wg.Add(1) go func(id proto.NodeID) { defer wg.Done() @@ -616,7 +619,7 @@ func (c *Chain) syncCurrentHead(ctx context.Context) { case <-cld.Done(): log.WithError(cld.Err()).Warn("add pending block aborted") } - }(v) + }(s) } } } @@ -763,7 +766,7 @@ func (c *Chain) replaceAndSwitchToBranch( } return fmt.Sprintf("[%04d]", i) }(), - }).Debugf("pruning branch") + }).Debug("pruning branch") } } // Replace current branches @@ -990,8 +993,8 @@ func (c *Chain) goFuncWithTimeout(f func(ctx context.Context), timeout time.Dura go func() { var ctx, ccl = context.WithTimeout(c.ctx, timeout) defer func() { - ccl() c.wg.Done() + ccl() }() f(ctx) }() diff --git a/crypto/verifier/common.go b/crypto/verifier/common.go index 41fa8ecb4..7885eb67a 100644 --- a/crypto/verifier/common.go +++ b/crypto/verifier/common.go @@ -75,7 +75,7 @@ func (i *DefaultHashSignVerifierImpl) Verify(mh MarshalHasher) (err error) { err = errors.WithStack(ErrHashValueNotMatch) return } - if !i.Signature.Verify(h[:], i.Signee) { + if i.Signature == nil || i.Signee == nil || !i.Signature.Verify(h[:], i.Signee) { err = errors.WithStack(ErrSignatureNotMatch) return } diff --git a/crypto/verifier/common_test.go b/crypto/verifier/common_test.go index 953ce61f8..c9ab5fd84 100644 --- a/crypto/verifier/common_test.go +++ b/crypto/verifier/common_test.go @@ -79,6 +79,20 @@ func TestDefaultHashSignVerifierImpl(t *testing.T) { So(err, ShouldEqual, ErrHashValueNotMatch) }) }) + Convey("When the signee is not set", func() { + obj.HSV.Signee = nil + Convey("The verifier should return signature not match error", func() { + err = errors.Cause(obj.Verify()) + So(err, ShouldEqual, ErrSignatureNotMatch) + }) + }) + Convey("When the signature is not set", func() { + obj.HSV.Signature = nil + Convey("The verifier should return signature not match error", func() { + err = errors.Cause(obj.Verify()) + So(err, ShouldEqual, ErrSignatureNotMatch) + }) + }) Convey("When the signee is modified", func() { var _, pub, err = asymmetric.GenSecp256k1KeyPair() So(err, ShouldBeNil)