Skip to content

Commit a8c8c9f

Browse files
leventeliuleventeliu
committed
Fix bug: billing on empty blocks
1 parent de0c00a commit a8c8c9f

1 file changed

Lines changed: 52 additions & 46 deletions

File tree

‎sqlchain/chain.go‎

Lines changed: 52 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -625,6 +625,11 @@ func (c *Chain) mainCycle(ctx context.Context) {
625625
func (c *Chain) sync() {
626626
le := c.logEntry()
627627
le.Debug("synchronizing chain state")
628+
defer func() {
629+
c.stat()
630+
c.pruneBlockCache()
631+
c.ai.advance(c.rt.getMinValidHeight())
632+
}()
628633
for {
629634
now := c.rt.now()
630635
height := c.rt.getHeightFromTime(now)
@@ -637,11 +642,7 @@ func (c *Chain) sync() {
637642
}
638643
for c.rt.getNextTurn() <= height {
639644
c.syncHead()
640-
c.stat()
641-
c.pruneBlockCache()
642645
c.rt.setNextTurn()
643-
c.ai.advance(c.rt.getMinValidHeight())
644-
c.heights <- c.rt.getHead().Height
645646
}
646647
}
647648
}
@@ -673,8 +674,38 @@ func (c *Chain) processBlocks(ctx context.Context) {
673674

674675
var stash []*types.Block
675676
for {
677+
le := c.logEntryWithHeadState()
676678
select {
677679
case h := <-c.heights:
680+
// Trigger billing
681+
head := c.rt.getHead()
682+
if uint64(h)%c.updatePeriod == 0 {
683+
ub, err := c.billing(h, head.node)
684+
if err != nil {
685+
le.WithError(err).Error("billing failed")
686+
}
687+
// allocate nonce
688+
nonceReq := &types.NextAccountNonceReq{}
689+
nonceResp := &types.NextAccountNonceResp{}
690+
nonceReq.Addr = *c.addr
691+
if err = rpc.RequestBP(route.MCCNextAccountNonce.String(), nonceReq, nonceResp); err != nil {
692+
// allocate nonce failed
693+
le.WithError(err).Warning("allocate nonce for transaction failed")
694+
}
695+
ub.Nonce = nonceResp.Nonce
696+
if err = ub.Sign(c.pk); err != nil {
697+
le.WithError(err).Warning("sign tx failed")
698+
}
699+
700+
addTxReq := &types.AddTxReq{TTL: 1}
701+
addTxResp := &types.AddTxResp{}
702+
addTxReq.Tx = ub
703+
le.Debugf("nonce in processBlocks: %d, addr: %s",
704+
addTxReq.Tx.GetAccountNonce(), addTxReq.Tx.GetAccountAddress())
705+
if err = rpc.RequestBP(route.MCCAddTx.String(), addTxReq, addTxResp); err != nil {
706+
le.WithError(err).Warning("send tx failed")
707+
}
708+
}
678709
// Return all stashed blocks to pending channel
679710
c.logEntryWithHeadState().WithFields(log.Fields{
680711
"height": h,
@@ -687,11 +718,10 @@ func (c *Chain) processBlocks(ctx context.Context) {
687718
}
688719
case block := <-c.blocks:
689720
height := c.rt.getHeightFromTime(block.Timestamp())
690-
le := c.logEntryWithHeadState().WithFields(log.Fields{
721+
le.WithFields(log.Fields{
691722
"block_height": height,
692723
"block_hash": block.BlockHash().String(),
693-
})
694-
le.Debug("processing new block")
724+
}).Debug("processing new block")
695725

696726
if height > c.rt.getNextTurn()-1 {
697727
// Stash newer blocks for later check
@@ -704,35 +734,6 @@ func (c *Chain) processBlocks(ctx context.Context) {
704734
if err := c.CheckAndPushNewBlock(block); err != nil {
705735
le.WithError(err).Error("failed to check and push new block")
706736
} else {
707-
head := c.rt.getHead()
708-
currentCount := uint64(head.node.count)
709-
if currentCount%c.updatePeriod == 0 {
710-
ub, err := c.billing(head.node)
711-
if err != nil {
712-
le.WithError(err).Error("billing failed")
713-
}
714-
// allocate nonce
715-
nonceReq := &types.NextAccountNonceReq{}
716-
nonceResp := &types.NextAccountNonceResp{}
717-
nonceReq.Addr = *c.addr
718-
if err = rpc.RequestBP(route.MCCNextAccountNonce.String(), nonceReq, nonceResp); err != nil {
719-
// allocate nonce failed
720-
le.WithError(err).Warning("allocate nonce for transaction failed")
721-
}
722-
ub.Nonce = nonceResp.Nonce
723-
if err = ub.Sign(c.pk); err != nil {
724-
le.WithError(err).Warning("sign tx failed")
725-
}
726-
727-
addTxReq := &types.AddTxReq{TTL: 1}
728-
addTxResp := &types.AddTxResp{}
729-
addTxReq.Tx = ub
730-
le.Debugf("nonce in processBlocks: %d, addr: %s",
731-
addTxReq.Tx.GetAccountNonce(), addTxReq.Tx.GetAccountAddress())
732-
if err = rpc.RequestBP(route.MCCAddTx.String(), addTxReq, addTxResp); err != nil {
733-
le.WithError(err).Warning("send tx failed")
734-
}
735-
}
736737
}
737738
}
738739
}
@@ -983,28 +984,33 @@ func (c *Chain) stat() {
983984
c.st.Stat(c.databaseID)
984985
}
985986

986-
func (c *Chain) billing(node *blockNode) (ub *types.UpdateBilling, err error) {
987-
log.WithField("db", c.databaseID).Debugf("begin to billing from count %d", node.count)
987+
func (c *Chain) billing(h int32, node *blockNode) (ub *types.UpdateBilling, err error) {
988+
le := c.logEntryWithHeadState()
989+
le.WithFields(log.Fields{"given_height": h}).Debug("begin to billing")
988990
var (
989991
i, j uint64
992+
iter *blockNode
990993
minerAddr proto.AccountAddress
991994
userAddr proto.AccountAddress
995+
minHeight = h - int32(c.updatePeriod)
992996
usersMap = make(map[proto.AccountAddress]uint64)
993997
minersMap = make(map[proto.AccountAddress]map[proto.AccountAddress]uint64)
994998
)
995999

996-
for i = 0; i < c.updatePeriod && node != nil; i++ {
997-
var block = node.block
1000+
for iter = node; iter != nil && iter.height > h; iter = iter.parent {
1001+
}
1002+
for i = 0; i < c.updatePeriod && iter != nil && iter.height > minHeight; i++ {
1003+
var block = iter.block
9981004
// Not cached, recover from storage
9991005
if block == nil {
1000-
if block, err = c.FetchBlock(node.height); err != nil {
1006+
if block, err = c.FetchBlock(iter.height); err != nil {
10011007
return
10021008
}
10031009
}
10041010
for _, tx := range block.QueryTxs {
10051011
minerAddr = tx.Response.ResponseAccount
10061012
if userAddr, err = crypto.PubKeyHash(tx.Request.Header.Signee); err != nil {
1007-
log.WithError(err).WithField("db", c.databaseID).Warning("billing fail: miner addr")
1013+
le.WithError(err).Warning("billing fail: miner addr")
10081014
return
10091015
}
10101016

@@ -1022,11 +1028,11 @@ func (c *Chain) billing(node *blockNode) (ub *types.UpdateBilling, err error) {
10221028

10231029
for _, req := range block.FailedReqs {
10241030
if minerAddr, err = crypto.PubKeyHash(block.Signee()); err != nil {
1025-
log.WithError(err).WithField("db", c.databaseID).Warning("billing fail: miner addr")
1031+
le.WithError(err).Warning("billing fail: miner addr")
10261032
return
10271033
}
10281034
if userAddr, err = crypto.PubKeyHash(req.Header.Signee); err != nil {
1029-
log.WithError(err).WithField("db", c.databaseID).Warning("billing fail: user addr")
1035+
le.WithError(err).Warning("billing fail: user addr")
10301036
return
10311037
}
10321038
if _, ok := minersMap[userAddr][minerAddr]; !ok {
@@ -1036,7 +1042,7 @@ func (c *Chain) billing(node *blockNode) (ub *types.UpdateBilling, err error) {
10361042
minersMap[userAddr][minerAddr] += uint64(len(req.Payload.Queries))
10371043
usersMap[userAddr] += uint64(len(req.Payload.Queries))
10381044
}
1039-
node = node.parent
1045+
iter = iter.parent
10401046
}
10411047

10421048
ub = types.NewUpdateBilling(&types.UpdateBillingHeader{
@@ -1046,7 +1052,7 @@ func (c *Chain) billing(node *blockNode) (ub *types.UpdateBilling, err error) {
10461052
i = 0
10471053
j = 0
10481054
for userAddr, cost := range usersMap {
1049-
log.WithField("db", c.databaseID).Debugf("user %s, cost %d", userAddr.String(), cost)
1055+
le.Debugf("user %s, cost %d", userAddr.String(), cost)
10501056
ub.Users[i] = &types.UserCost{
10511057
User: userAddr,
10521058
Cost: cost,

0 commit comments

Comments
 (0)