diff --git a/blockproducer/branch.go b/blockproducer/branch.go index 93592fec4..fcba6e5dc 100644 --- a/blockproducer/branch.go +++ b/blockproducer/branch.go @@ -22,7 +22,7 @@ import ( "time" pi "github.com/CovenantSQL/CovenantSQL/blockproducer/interfaces" - pl "github.com/CovenantSQL/CovenantSQL/blockproducer/limits" + "github.com/CovenantSQL/CovenantSQL/conf" ca "github.com/CovenantSQL/CovenantSQL/crypto/asymmetric" "github.com/CovenantSQL/CovenantSQL/crypto/hash" "github.com/CovenantSQL/CovenantSQL/proto" @@ -57,7 +57,7 @@ func newBranch( } // Apply new blocks to view and pool for _, bn := range list { - if len(bn.block.Transactions) > pl.MaxTransactionsPerBlock { + if len(bn.block.Transactions) > conf.MaxTransactionsPerBlock { return nil, ErrTooManyTransactionsInBlock } @@ -132,7 +132,7 @@ func (b *branch) applyBlock(n *blockNode) (br *branch, err error) { } var cpy = b.makeArena() - if len(n.block.Transactions) > pl.MaxTransactionsPerBlock { + if len(n.block.Transactions) > conf.MaxTransactionsPerBlock { return nil, ErrTooManyTransactionsInBlock } @@ -185,7 +185,7 @@ func (b *branch) produceBlock( cpy = b.makeArena() txs = cpy.sortUnpackedTxs() ierr error - packCount = pl.MaxTransactionsPerBlock + packCount = conf.MaxTransactionsPerBlock ) if len(txs) < packCount { diff --git a/blockproducer/chain.go b/blockproducer/chain.go index 83e7bc381..be25dd3b2 100644 --- a/blockproducer/chain.go +++ b/blockproducer/chain.go @@ -25,7 +25,6 @@ import ( "time" pi "github.com/CovenantSQL/CovenantSQL/blockproducer/interfaces" - pl "github.com/CovenantSQL/CovenantSQL/blockproducer/limits" "github.com/CovenantSQL/CovenantSQL/chainbus" "github.com/CovenantSQL/CovenantSQL/conf" "github.com/CovenantSQL/CovenantSQL/crypto" @@ -188,7 +187,7 @@ func NewChainWithContext(ctx context.Context, cfg *Config) (c *Chain, err error) return } if t = cfg.ConfirmThreshold; t <= 0.0 { - t = float64(2) / 3.0 + t = conf.DefaultConfirmThreshold } if m = uint32(math.Ceil(float64(l)*t + 1)); m > l { m = l @@ -371,19 +370,29 @@ func (c *Chain) advanceNextHeight(now time.Time, d time.Duration) { func (c *Chain) syncHeads() { for { - var h = c.heightOfTime(c.now()) - if c.getNextHeight() > h { + var ( + now = c.now() + nowHeight uint32 + ) + if now.Before(c.genesisTime) { + log.WithFields(log.Fields{ + "local": c.getLocalBPInfo(), + }).Info("now time is before genesis time, waiting for genesis") + break + } + if nowHeight = c.heightOfTime(c.now()); c.getNextHeight() > nowHeight { break } - for c.getNextHeight() <= h { + for c.getNextHeight() <= nowHeight { // TODO(leventeliu): use the test mode flag to bypass the long-running synchronizing // on startup by now, need better solution here. if conf.GConf.StartupSyncHoles { log.WithFields(log.Fields{ + "local": c.getLocalBPInfo(), "next_height": c.getNextHeight(), - "height": h, + "now_height": nowHeight, }).Debug("synchronizing head blocks") - c.syncCurrentHead(c.ctx) + c.blockingSyncCurrentHead(c.ctx, conf.BPStartupRequiredReachableCount) } c.increaseNextHeight() } @@ -463,18 +472,18 @@ func (c *Chain) processAddTxReq(addTxReq *types.AddTxReq) { le.WithError(err).Warn("failed to load base nonce of transaction account") return } - if nonce < base || nonce >= base+pl.MaxPendingTxsPerAccount { + if nonce < base || nonce >= base+conf.MaxPendingTxsPerAccount { // TODO(leventeliu): should persist to somewhere for tx query? le.WithFields(log.Fields{ "base_nonce": base, - "pending_limit": pl.MaxPendingTxsPerAccount, + "pending_limit": conf.MaxPendingTxsPerAccount, }).Warn("invalid transaction nonce") return } // Broadcast to other block producers - if ttl > pl.MaxTxBroadcastTTL { - ttl = pl.MaxTxBroadcastTTL + if ttl > conf.MaxTxBroadcastTTL { + ttl = conf.MaxTxBroadcastTTL } if ttl > 0 { c.nonblockingBroadcastTx(ttl-1, tx) @@ -509,7 +518,7 @@ func (c *Chain) mainCycle(ctx context.Context) { select { case <-timer.C: // Try to fetch block at height `nextHeight-1` until enough peers are reachable - if err := c.blockingSyncCurrentHead(ctx); err != nil { + if err := c.blockingSyncCurrentHead(ctx, c.getRequiredConfirms()); err != nil { log.WithError(err).Info("abort main cycle") timer.Reset(0) return @@ -537,7 +546,7 @@ func (c *Chain) mainCycle(ctx context.Context) { } } -func (c *Chain) blockingSyncCurrentHead(ctx context.Context) (err error) { +func (c *Chain) blockingSyncCurrentHead(ctx context.Context, requiredReachable uint32) (err error) { var ( ticker *time.Ticker interval = 1 * time.Second @@ -548,11 +557,11 @@ func (c *Chain) blockingSyncCurrentHead(ctx context.Context) (err error) { ticker = time.NewTicker(interval) defer ticker.Stop() for { + if c.syncCurrentHead(ctx, requiredReachable) { + return + } select { case <-ticker.C: - if c.syncCurrentHead(ctx) { - return - } case <-ctx.Done(): err = ctx.Err() return @@ -561,11 +570,11 @@ func (c *Chain) blockingSyncCurrentHead(ctx context.Context) (err error) { } // syncCurrentHead synchronizes a block at the current height of the local peer from the known -// remote peers. The return value `ok` indicates that there're at least `c.confirms-1` replies -// from these gossip calls. -func (c *Chain) syncCurrentHead(ctx context.Context) (ok bool) { - var h = c.getNextHeight() - 1 - if c.head().height >= h { +// remote peers. The return value `ok` indicates that there're at least `requiredReachable-1` +// replies from these gossip calls. +func (c *Chain) syncCurrentHead(ctx context.Context, requiredReachable uint32) (ok bool) { + var currentHeight = c.getNextHeight() - 1 + if c.head().height >= currentHeight { ok = true return } @@ -573,20 +582,14 @@ func (c *Chain) syncCurrentHead(ctx context.Context) (ok bool) { // Initiate blocking gossip calls to fetch block of the current height, // with timeout of one tick. var ( - unreachable = c.blockingFetchBlock(ctx, h) - - needConfirms, serversNum = func() (cf, sn uint32) { - c.RLock() - defer c.RUnlock() - cf, sn = c.confirms, c.localBPInfo.total - return - }() + unreachable = c.blockingFetchBlock(ctx, currentHeight) + serversNum = c.getLocalBPInfo().total ) - if ok = unreachable+needConfirms <= serversNum; !ok { + if ok = unreachable+requiredReachable <= serversNum; !ok { log.WithFields(log.Fields{ "peer": c.getLocalBPInfo(), - "sync_head_height": h, + "sync_head_height": currentHeight, "unreachable_count": unreachable, }).Warn("one or more block producers are currently unreachable") } @@ -893,6 +896,12 @@ func (c *Chain) heightOfTime(t time.Time) uint32 { return uint32(t.Sub(c.genesisTime) / c.period) } +func (c *Chain) getRequiredConfirms() uint32 { + c.RLock() + defer c.RUnlock() + return c.confirms +} + func (c *Chain) getNextHeight() uint32 { c.RLock() defer c.RUnlock() diff --git a/blockproducer/chain_test.go b/blockproducer/chain_test.go index 61e41668b..58fd88bf0 100644 --- a/blockproducer/chain_test.go +++ b/blockproducer/chain_test.go @@ -145,6 +145,32 @@ func TestChain(t *testing.T) { Tick: time.Duration(300 * time.Millisecond), } + Convey("A new chain running before genesis time should be waiting for genesis", func() { + config.Genesis.SignedHeader.Timestamp = time.Now().Add(24 * time.Hour) + err = genesis.PackAndSignBlock(testingPrivateKey) + So(err, ShouldBeNil) + chain, err = NewChain(config) + So(err, ShouldBeNil) + + var sv = rpc.NewServer() + err = sv.InitRPCServer("localhost:0", testingPrivateKeyFile, []byte{}) + So(err, ShouldBeNil) + defer sv.Stop() + chain.server = sv + chain.confirms = 1 + chain.Start() + defer func() { + err = chain.Stop() + So(err, ShouldBeNil) + chain = nil + }() + time.Sleep(5 * chain.period) + var _, count, height, err = chain.fetchLastIrreversibleBlock() + So(err, ShouldBeNil) + So(count, ShouldEqual, 0) + So(height, ShouldEqual, 0) + }) + chain, err = NewChain(config) So(err, ShouldBeNil) So(chain, ShouldNotBeNil) @@ -345,7 +371,8 @@ func TestChain(t *testing.T) { chain.confirms = 1 chain.Start() defer func() { - chain.Stop() + err = chain.Stop() + So(err, ShouldBeNil) chain = nil }() chain.addTx(&types.AddTxReq{TTL: 1, Tx: t1}) diff --git a/blockproducer/config.go b/blockproducer/config.go index d9d4c3ae1..1cb6b03dc 100644 --- a/blockproducer/config.go +++ b/blockproducer/config.go @@ -24,10 +24,6 @@ import ( "github.com/CovenantSQL/CovenantSQL/types" ) -const ( - blockVersion int32 = 0x01 -) - // Config is the main chain configuration. type Config struct { Mode string diff --git a/blockproducer/limits/limits.go b/conf/limits.go similarity index 92% rename from blockproducer/limits/limits.go rename to conf/limits.go index 1e1f96405..8c12c08cb 100644 --- a/blockproducer/limits/limits.go +++ b/conf/limits.go @@ -14,8 +14,7 @@ * limitations under the License. */ -// Package limits defines limits of the CovenantSQL system. -package limits +package conf const ( // MaxTxBroadcastTTL defines the TTL limit of a AddTx request broadcasting within the diff --git a/conf/parameters.go b/conf/parameters.go index 8fbc04951..4fa2296ef 100644 --- a/conf/parameters.go +++ b/conf/parameters.go @@ -16,17 +16,12 @@ package conf -import "time" +// This parameters should be kept consistent in all BPs. +const ( + DefaultConfirmThreshold = float64(2) / 3.0 +) +// This parameters will not cause inconsistency within certain range. const ( - // BPPeriod is the block producer block produce period. - BPPeriod = 3 * time.Second - // BPTick is the block produce block fetch tick. - BPTick = 1 * time.Second - // SQLChainPeriod is the sqlchain block produce period. - SQLChainPeriod = 3 * time.Second - // SQLChainTick is the sqlchain block fetch tick. - SQLChainTick = 1 * time.Second - // SQLChainTTL is the sqlchain unack query billing ttl. - SQLChainTTL = 10 + BPStartupRequiredReachableCount = 2 // NOTE: this includes myself )