diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index 4865840e5..5b9abed7c 100644 --- a/.gitlab-ci.yml +++ b/.gitlab-ci.yml @@ -69,6 +69,9 @@ old-client-compatibility: - set -o errexit - set -o pipefail - set -x + - commit=$(git rev-parse --short HEAD) + - branch=$(git branch -rv |grep $commit | awk '{print $1}') + - if [[ $branch =~ "/beta_" ]]; then exit 0; fi - make clean - cp ${BIN_CACHE}/* bin/ - ./test/compatibility/specific_old.sh client @@ -79,6 +82,9 @@ old-bp-compatibility: - set -o errexit - set -o pipefail - set -x + - commit=$(git rev-parse --short HEAD) + - branch=$(git branch -rv |grep $commit | awk '{print $1}') + - if [[ $branch =~ "/beta_" ]]; then exit 0; fi - make clean - cp ${BIN_CACHE}/* bin/ - ./test/compatibility/specific_old.sh bp @@ -89,6 +95,9 @@ old-miner-compatibility: - set -o errexit - set -o pipefail - set -x + - commit=$(git rev-parse --short HEAD) + - branch=$(git branch -rv |grep $commit | awk '{print $1}') + - if [[ $branch =~ "/beta_" ]]; then exit 0; fi - make clean - cp ${BIN_CACHE}/* bin/ - ./test/compatibility/specific_old.sh miner diff --git a/alltest.sh b/alltest.sh index e362edbfb..459e60d1b 100755 --- a/alltest.sh +++ b/alltest.sh @@ -5,17 +5,17 @@ set -o pipefail set -o nounset main() { - go test -tags "$UNITTESTTAGS" -race -failfast -parallel 16 -cpu 16 -coverprofile main.cover.out $(go list ./... | grep -v CovenantSQL/api) - go test -tags "$UNITTESTTAGS" -race -failfast -parallel 16 -cpu 16 -coverpkg ./api/...,./rpc/jsonrpc -coverprofile api.cover.out ./api/... + go test -tags "${UNITTESTTAGS:-}" -race -failfast -parallel 16 -cpu 16 -coverprofile main.cover.out $(go list ./... | grep -v CovenantSQL/api) + go test -tags "${UNITTESTTAGS:-}" -race -failfast -parallel 16 -cpu 16 -coverpkg ./api/...,./rpc/jsonrpc -coverprofile api.cover.out ./api/... set -x gocovmerge main.cover.out api.cover.out $(find cmd -name "*.cover.out") | grep -F -v '_gen.go' > coverage.txt && rm -f *.cover.out bash <(curl -s https://codecov.io/bash) # some benchmarks - go test -tags "$UNITTESTTAGS" -bench=^BenchmarkPersistentCaller_Call$ -run ^$ ./rpc/ + go test -tags "${UNITTESTTAGS:-}" -bench=^BenchmarkPersistentCaller_Call$ -run ^$ ./rpc/ bash cleanupDB.sh || true - go test -tags "$UNITTESTTAGS" -bench=^BenchmarkMiner$ -benchtime=5s -run ^$ ./cmd/cql-minerd/ -bench-miner-count=2 + go test -tags "${UNITTESTTAGS:-}" -bench=^BenchmarkMiner$ -benchtime=5s -run ^$ ./cmd/cql-minerd/ -bench-miner-count=2 bash cleanupDB.sh || true } diff --git a/blockproducer/errors.go b/blockproducer/errors.go index 2a6cc74cb..f8a92e0d8 100644 --- a/blockproducer/errors.go +++ b/blockproducer/errors.go @@ -54,6 +54,8 @@ var ( ErrUnknownTransactionType = errors.New("unknown transaction type") // ErrInvalidSender indicates that tx.Signee != tx.Sender. ErrInvalidSender = errors.New("invalid sender") + // ErrInvalidRange indicates that the billing range is invalid. + ErrInvalidRange = errors.New("invalid billing range") // ErrNoSuchMiner indicates that this miner does not exist or register. ErrNoSuchMiner = errors.New("no such miner") // ErrNoEnoughMiner indicates that there is not enough miners diff --git a/blockproducer/interfaces/transaction.go b/blockproducer/interfaces/transaction.go index 085a1d74b..e074e6ea3 100644 --- a/blockproducer/interfaces/transaction.go +++ b/blockproducer/interfaces/transaction.go @@ -46,8 +46,8 @@ func FromBytes(b []byte) TransactionType { } const ( - // TransactionTypeBilling defines billing transaction type. - TransactionTypeBilling TransactionType = iota + // TransactionTypeDeprecated is a deprecated transaction type, do NOT use it. + TransactionTypeDeprecated TransactionType = iota // TransactionTypeTransfer defines transfer transaction type. TransactionTypeTransfer // TransactionTypeCreateAccount defines account creation transaction type. @@ -78,8 +78,6 @@ const ( func (t TransactionType) String() string { switch t { - case TransactionTypeBilling: - return "Billing" case TransactionTypeTransfer: return "Transfer" case TransactionTypeCreateAccount: diff --git a/blockproducer/interfaces/transaction_test.go b/blockproducer/interfaces/transaction_test.go index 7a80ac53b..c7c436e43 100644 --- a/blockproducer/interfaces/transaction_test.go +++ b/blockproducer/interfaces/transaction_test.go @@ -55,7 +55,7 @@ func TestTypes(t *testing.T) { } }) Convey("test string", t, func() { - for i := TransactionTypeBilling; i != TransactionTypeNumber+1; i++ { + for i := TransactionTypeTransfer; i != TransactionTypeNumber+1; i++ { So(i.String(), ShouldNotBeEmpty) } }) diff --git a/blockproducer/interfaces/transaction_wrapper_test.go b/blockproducer/interfaces/transaction_wrapper_test.go index 8084c4453..276cefaf9 100644 --- a/blockproducer/interfaces/transaction_wrapper_test.go +++ b/blockproducer/interfaces/transaction_wrapper_test.go @@ -63,7 +63,7 @@ func (e *TestTransactionEncode) Msgsize() int { } func init() { - pi.RegisterTransaction(pi.TransactionTypeBilling, (*TestTransactionEncode)(nil)) + pi.RegisterTransaction(pi.TransactionTypeTransfer, (*TestTransactionEncode)(nil)) } func TestTransactionWrapper(t *testing.T) { @@ -81,13 +81,13 @@ func TestTransactionWrapper(t *testing.T) { // encode test e := &TestTransactionEncode{} - e.SetTransactionType(pi.TransactionTypeBilling) + e.SetTransactionType(pi.TransactionTypeTransfer) buf, err = utils.EncodeMsgPack(e) So(err, ShouldBeNil) var v2 pi.Transaction err = utils.DecodeMsgPack(buf.Bytes(), &v2) So(err, ShouldBeNil) - So(v2.GetTransactionType(), ShouldEqual, pi.TransactionTypeBilling) + So(v2.GetTransactionType(), ShouldEqual, pi.TransactionTypeTransfer) // encode with wrapper test e2 := pi.WrapTransaction(e) @@ -96,14 +96,14 @@ func TestTransactionWrapper(t *testing.T) { var v3 pi.Transaction err = utils.DecodeMsgPack(buf.Bytes(), &v3) So(err, ShouldBeNil) - So(v3.GetTransactionType(), ShouldEqual, pi.TransactionTypeBilling) + So(v3.GetTransactionType(), ShouldEqual, pi.TransactionTypeTransfer) tw, ok := v3.(*pi.TransactionWrapper) So(ok, ShouldBeTrue) - So(tw.Unwrap().GetTransactionType(), ShouldEqual, pi.TransactionTypeBilling) + So(tw.Unwrap().GetTransactionType(), ShouldEqual, pi.TransactionTypeTransfer) // test encode non-existence type e3 := &TestTransactionEncode{} - e3.SetTransactionType(pi.TransactionTypeTransfer) + e3.SetTransactionType(pi.TransactionTypeCreateAccount) buf, err = utils.EncodeMsgPack(e3) So(err, ShouldBeNil) var v4 pi.Transaction @@ -132,21 +132,21 @@ func TestTransactionWrapper(t *testing.T) { So(err, ShouldNotBeNil) // test invalid decode, nil payload - buf, err = utils.EncodeMsgPack([]interface{}{pi.TransactionTypeBilling, nil}) + buf, err = utils.EncodeMsgPack([]interface{}{pi.TransactionTypeTransfer, nil}) So(err, ShouldBeNil) var v8 pi.Transaction err = utils.DecodeMsgPack(buf.Bytes(), &v8) So(err, ShouldNotBeNil) // test invalid decode, invalid payload container type - buf, err = utils.EncodeMsgPack([]interface{}{pi.TransactionTypeBilling, []uint64{}}) + buf, err = utils.EncodeMsgPack([]interface{}{pi.TransactionTypeTransfer, []uint64{}}) So(err, ShouldBeNil) var v9 pi.Transaction err = utils.DecodeMsgPack(buf.Bytes(), &v9) So(err, ShouldNotBeNil) // extra payload - buf, err = utils.EncodeMsgPack([]interface{}{pi.TransactionTypeBilling, e, 1, 2}) + buf, err = utils.EncodeMsgPack([]interface{}{pi.TransactionTypeTransfer, e, 1, 2}) So(err, ShouldBeNil) var v10 pi.Transaction err = utils.DecodeMsgPack(buf.Bytes(), &v10) @@ -174,14 +174,14 @@ func TestTransactionWrapper(t *testing.T) { So(err, ShouldNotBeNil) // test tx data - buf, err = utils.EncodeMsgPack(map[string]interface{}{"TxType": pi.TransactionTypeBilling, "TestField": 1}) + buf, err = utils.EncodeMsgPack(map[string]interface{}{"TxType": pi.TransactionTypeTransfer, "TestField": 1}) So(err, ShouldBeNil) var v14 pi.Transaction err = utils.DecodeMsgPack(buf.Bytes(), &v14) So(err, ShouldBeNil) // test invalid tx data - buf, err = utils.EncodeMsgPack(map[string]interface{}{"TxType": pi.TransactionTypeBilling, "TestField": "happy"}) + buf, err = utils.EncodeMsgPack(map[string]interface{}{"TxType": pi.TransactionTypeTransfer, "TestField": "happy"}) So(err, ShouldBeNil) var v15 pi.Transaction err = utils.DecodeMsgPack(buf.Bytes(), &v15) @@ -189,7 +189,7 @@ func TestTransactionWrapper(t *testing.T) { // test json marshal and unmarshal v16 := &TestTransactionEncode{TestField: 10} - v16.SetTransactionType(pi.TransactionTypeBilling) + v16.SetTransactionType(pi.TransactionTypeTransfer) var v17 pi.Transaction = v16 var jsonData []byte jsonData, err = json.Marshal(v17) @@ -200,17 +200,17 @@ func TestTransactionWrapper(t *testing.T) { err = json.Unmarshal(jsonData, &v18) So(err, ShouldBeNil) So(v18.(*pi.TransactionWrapper).Unwrap(), ShouldNotBeNil) - So(v18.GetTransactionType(), ShouldEqual, pi.TransactionTypeBilling) + So(v18.GetTransactionType(), ShouldEqual, pi.TransactionTypeTransfer) So(v18.(*pi.TransactionWrapper).Unwrap().(*TestTransactionEncode).TestField, ShouldEqual, 10) jsonData, err = json.Marshal(v18) So(string(jsonData), ShouldContainSubstring, "TestField") v18.(*pi.TransactionWrapper).Transaction = nil - jsonData = []byte(`{"TxType": 0, "TestField": 11}`) + jsonData = []byte(`{"TxType": 1, "TestField": 11}`) err = json.Unmarshal(jsonData, &v18) So(err, ShouldBeNil) - So(v18.GetTransactionType(), ShouldEqual, pi.TransactionTypeBilling) + So(v18.GetTransactionType(), ShouldEqual, pi.TransactionTypeTransfer) So(v18.(*pi.TransactionWrapper).Unwrap().(*TestTransactionEncode).TestField, ShouldEqual, 11) // unmarshal fail cases @@ -225,7 +225,7 @@ func TestTransactionWrapper(t *testing.T) { So(err, ShouldNotBeNil) v18.(*pi.TransactionWrapper).Transaction = nil - jsonData = []byte(fmt.Sprintf(`{"TxType": %d, "TestField": 11}`, pi.TransactionTypeTransfer)) + jsonData = []byte(fmt.Sprintf(`{"TxType": %d, "TestField": 11}`, pi.TransactionTypeCreateAccount)) err = json.Unmarshal(jsonData, &v18) So(err, ShouldNotBeNil) }) diff --git a/blockproducer/metastate.go b/blockproducer/metastate.go index c21c68e49..43fcb8252 100644 --- a/blockproducer/metastate.go +++ b/blockproducer/metastate.go @@ -509,21 +509,6 @@ func (s *metaState) increaseNonce(addr proto.AccountAddress) (err error) { return } -func (s *metaState) applyBilling(tx *types.Billing) (err error) { - for i, v := range tx.Receivers { - // Create empty receiver account if not found - s.loadOrStoreAccountObject(*v, &types.Account{Address: *v}) - - if err = s.increaseAccountCovenantBalance(*v, tx.Fees[i]); err != nil { - return - } - if err = s.increaseAccountStableBalance(*v, tx.Rewards[i]); err != nil { - return - } - } - return -} - func (s *metaState) updateProviderList(tx *types.ProvideService) (err error) { sender, err := crypto.PubKeyHash(tx.Signee) if err != nil { @@ -915,6 +900,13 @@ func (s *metaState) updateBilling(tx *types.UpdateBilling) (err error) { err = errors.Wrap(ErrDatabaseNotFound, "update billing failed") return } + + if tx.Version > 0 && (tx.Range.From >= tx.Range.To || newProfile.LastUpdatedHeight != tx.Range.From) { + err = errors.Wrapf(ErrInvalidRange, + "update billing within range %d:(%d, %d]", + newProfile.LastUpdatedHeight, tx.Range.From, tx.Range.To) + return + } log.Debugf("update billing addr: %s, user: %d, tx: %v", tx.GetAccountAddress(), len(tx.Users), tx) if newProfile.GasPrice == 0 { @@ -988,6 +980,7 @@ func (s *metaState) updateBilling(tx *types.UpdateBilling) (err error) { } } } + newProfile.LastUpdatedHeight = tx.Range.To s.dirty.databases[tx.Receiver.DatabaseID()] = newProfile return } @@ -1128,8 +1121,6 @@ func (s *metaState) applyTransaction(tx pi.Transaction) (err error) { err = s.transferAccountToken(t) } return - case *types.Billing: - err = s.applyBilling(t) case *types.BaseAccount: err = s.storeBaseAccount(t.Address, &t.Account) case *types.ProvideService: diff --git a/blockproducer/metastate_test.go b/blockproducer/metastate_test.go index c290fc14d..0978dc5f1 100644 --- a/blockproducer/metastate_test.go +++ b/blockproducer/metastate_test.go @@ -489,13 +489,12 @@ func TestMetaState(t *testing.T) { Amount: 0, }, ) - t2 = types.NewBilling( - &types.BillingHeader{ - Nonce: 2, - Producer: addr1, - Receivers: []*proto.AccountAddress{&addr2}, - Fees: []uint64{1}, - Rewards: []uint64{1}, + t2 = types.NewTransfer( + &types.TransferHeader{ + Sender: addr1, + Receiver: addr2, + Nonce: 2, + Amount: 0, }, ) ) @@ -557,29 +556,11 @@ func TestMetaState(t *testing.T) { Amount: 10, }, ), - types.NewBilling( - &types.BillingHeader{ - Nonce: 2, - Producer: addr1, - Receivers: []*proto.AccountAddress{&addr2}, - Fees: []uint64{1}, - Rewards: []uint64{1}, - }, - ), - types.NewBilling( - &types.BillingHeader{ - Nonce: 1, - Producer: addr2, - Receivers: []*proto.AccountAddress{&addr1}, - Fees: []uint64{1}, - Rewards: []uint64{1}, - }, - ), types.NewTransfer( &types.TransferHeader{ Sender: addr2, Receiver: addr1, - Nonce: 2, + Nonce: 1, Amount: 1, }, ), @@ -587,7 +568,7 @@ func TestMetaState(t *testing.T) { &types.TransferHeader{ Sender: addr1, Receiver: addr2, - Nonce: 3, + Nonce: 2, Amount: 10, }, ), @@ -595,7 +576,7 @@ func TestMetaState(t *testing.T) { &types.TransferHeader{ Sender: addr2, Receiver: addr1, - Nonce: 3, + Nonce: 2, Amount: 1, }, ), @@ -603,7 +584,7 @@ func TestMetaState(t *testing.T) { &types.TransferHeader{ Sender: addr2, Receiver: addr1, - Nonce: 4, + Nonce: 3, Amount: 1, }, ), @@ -612,12 +593,10 @@ func TestMetaState(t *testing.T) { txs[0].Sign(privKey1) txs[1].Sign(privKey2) txs[2].Sign(privKey1) - txs[3].Sign(privKey1) - txs[4].Sign(privKey2) + txs[3].Sign(privKey2) + txs[4].Sign(privKey1) txs[5].Sign(privKey2) - txs[6].Sign(privKey1) - txs[7].Sign(privKey2) - txs[8].Sign(privKey2) + txs[6].Sign(privKey2) for _, tx := range txs { err = ms.apply(tx) So(err, ShouldBeNil) @@ -626,10 +605,10 @@ func TestMetaState(t *testing.T) { Convey("The state should match the update result", func() { bl, loaded = ms.loadAccountTokenBalance(addr1, types.Particle) So(loaded, ShouldBeTrue) - So(bl, ShouldEqual, 84) + So(bl, ShouldEqual, 83) bl, loaded = ms.loadAccountTokenBalance(addr2, types.Particle) So(loaded, ShouldBeTrue) - So(bl, ShouldEqual, 118) + So(bl, ShouldEqual, 117) }) }) Convey("When SQLChain are created", func() { @@ -1185,7 +1164,12 @@ func TestMetaState(t *testing.T) { }, }, }, + Range: types.Range{ + From: 0, + To: 10, + }, }) + ub.Version = int32(ub.HSPDefaultVersion()) nonce, err = ms.nextNonce(addr2) So(err, ShouldBeNil) ub.Nonce = nonce @@ -1363,8 +1347,13 @@ func TestMetaState(t *testing.T) { UpdateBillingHeader: types.UpdateBillingHeader{ Receiver: addr1, Nonce: up.Nonce, + Range: types.Range{ + From: 0, + To: 10, + }, }, } + ub1.Version = int32(ub1.HSPDefaultVersion()) err = ub1.Sign(privKey1) So(err, ShouldBeNil) err = ms.apply(ub1) @@ -1449,8 +1438,13 @@ func TestMetaState(t *testing.T) { Receiver: dbAccount, Users: users[:], Nonce: 2, + Range: types.Range{ + From: 0, + To: 10, + }, }, } + ub2.Version = int32(ub2.HSPDefaultVersion()) err = ub2.Sign(privKey2) So(err, ShouldBeNil) err = ms.apply(ub2) @@ -1496,8 +1490,13 @@ func TestMetaState(t *testing.T) { Receiver: dbAccount, Users: users[:], Nonce: 3, + Range: types.Range{ + From: 10, + To: 20, + }, }, } + ub3.Version = int32(ub3.HSPDefaultVersion()) err = ub3.Sign(privKey2) So(err, ShouldBeNil) err = ms.apply(ub3) diff --git a/cmd/cql-minerd/integration_test.go b/cmd/cql-minerd/integration_test.go index 44745276f..4ab6a6ca0 100644 --- a/cmd/cql-minerd/integration_test.go +++ b/cmd/cql-minerd/integration_test.go @@ -735,7 +735,7 @@ func benchDB(b *testing.B, db *sql.DB, createDB bool) { }) routineCount := runtime.NumGoroutine() - if routineCount > 100 { + if routineCount > 150 { b.Errorf("go routine count: %d", routineCount) } else { log.Infof("go routine count: %d", routineCount) @@ -777,7 +777,7 @@ func benchDB(b *testing.B, db *sql.DB, createDB bool) { }) routineCount = runtime.NumGoroutine() - if routineCount > 100 { + if routineCount > 150 { b.Errorf("go routine count: %d", routineCount) } else { log.Infof("go routine count: %d", routineCount) diff --git a/sqlchain/blockindex.go b/sqlchain/blockindex.go index 4256b675f..6abe47010 100644 --- a/sqlchain/blockindex.go +++ b/sqlchain/blockindex.go @@ -33,31 +33,16 @@ type blockNode struct { } func newBlockNode(height int32, block *types.Block, parent *blockNode) *blockNode { + var count int32 + if parent != nil { + count = parent.count + 1 + } return &blockNode{ - hash: *block.BlockHash(), + hash: block.SignedHeader.HSV.DataHash, parent: parent, block: block, height: height, - count: func() int32 { - if parent != nil { - return parent.count + 1 - } - - return 0 - }(), - } -} - -func (n *blockNode) initBlockNode(height int32, block *types.Block, parent *blockNode) { - n.block = block - n.hash = *block.BlockHash() - n.parent = nil - n.height = height - n.count = 0 - - if parent != nil { - n.parent = parent - n.count = parent.count + 1 + count: count, } } diff --git a/sqlchain/blockindex_test.go b/sqlchain/blockindex_test.go index e372980c8..0d486347b 100644 --- a/sqlchain/blockindex_test.go +++ b/sqlchain/blockindex_test.go @@ -19,7 +19,6 @@ package sqlchain import ( "testing" - "github.com/CovenantSQL/CovenantSQL/crypto/hash" "github.com/CovenantSQL/CovenantSQL/types" ) @@ -74,20 +73,7 @@ func TestNewBlockNode(t *testing.T) { } func TestInitBlockNode(t *testing.T) { - parent := &blockNode{ - parent: nil, - hash: hash.Hash{}, - count: -1, - } - - child := &blockNode{ - parent: nil, - hash: hash.Hash{}, - count: -1, - } - - parent.initBlockNode(0, testBlocks[0], nil) - + parent := newBlockNode(0, testBlocks[0], nil) if parent == nil { t.Fatal("unexpected result: nil") } else if parent.parent != nil { @@ -96,8 +82,7 @@ func TestInitBlockNode(t *testing.T) { t.Fatalf("unexpected height: %d", parent.count) } - child.initBlockNode(1, testBlocks[1], parent) - + child := newBlockNode(1, testBlocks[1], parent) if child == nil { t.Fatal("unexpected result: nil") } else if child.parent != parent { diff --git a/sqlchain/chain.go b/sqlchain/chain.go index 6f421c372..dba3dff1d 100644 --- a/sqlchain/chain.go +++ b/sqlchain/chain.go @@ -22,7 +22,6 @@ import ( "database/sql" "encoding/binary" "fmt" - "os" rt "runtime" "sync" "sync/atomic" @@ -52,21 +51,22 @@ const ( ) var ( - metaState = [4]byte{'S', 'T', 'A', 'T'} metaBlockIndex = [4]byte{'B', 'L', 'C', 'K'} metaResponseIndex = [4]byte{'R', 'E', 'S', 'P'} metaAckIndex = [4]byte{'Q', 'A', 'C', 'K'} - leveldbConf = opt.Options{} + + leveldbConf = opt.Options{ + Compression: opt.SnappyCompression, + } + leveldbInit sync.Once + blkDB *leveldb.DB + txDB *leveldb.DB // Atomic counters for stats cachedBlockCount int32 ) -func init() { - leveldbConf.Compression = opt.SnappyCompression -} - -func statBlock(b *types.Block) { +func trackBlock(b *types.Block) { atomic.AddInt32(&cachedBlockCount, 1) rt.SetFinalizer(b, func(_ *types.Block) { atomic.AddInt32(&cachedBlockCount, -1) @@ -98,16 +98,11 @@ func keyWithSymbolToHeight(k []byte) int32 { // Chain represents a sql-chain. type Chain struct { - // bdb stores state, profile and block - bdb *leveldb.DB - // tdb stores ack/request/response - tdb *leveldb.DB - bi *blockIndex - ai *ackIndex - st *x.State - cl *rpc.Caller - rt *runtime - ctx context.Context // ctx is the root context of Chain + bi *blockIndex + ai *ackIndex + st *x.State + cl *rpc.Caller + rt *runtime blocks chan *types.Block heights chan int32 @@ -126,6 +121,10 @@ type Chain struct { pk *asymmetric.PrivateKey // addr is the AccountAddress generate from public key. addr *proto.AccountAddress + // key prefixes + metaBlockIndex []byte + metaResponseIndex []byte + metaAckIndex []byte } // NewChain creates a new sql-chain struct. @@ -135,41 +134,35 @@ func NewChain(c *Config) (chain *Chain, err error) { // NewChainWithContext creates a new sql-chain struct with context. func NewChainWithContext(ctx context.Context, c *Config) (chain *Chain, err error) { - // TODO(leventeliu): this is a rough solution, you may also want to clean database file and - // force rebuilding. - var fi os.FileInfo - if fi, err = os.Stat(c.ChainFilePrefix + "-block-state.ldb"); err == nil && fi.Mode().IsDir() { - return LoadChain(c) - } - - err = c.Genesis.VerifyAsGenesis() - if err != nil { - return - } - - // Open LevelDB for block and state - bdbFile := c.ChainFilePrefix + "-block-state.ldb" - bdb, err := leveldb.OpenFile(bdbFile, &leveldbConf) - if err != nil { - err = errors.Wrapf(err, "open leveldb %s", bdbFile) - return - } + le := log.WithField("db", c.DatabaseID) - log.WithField("db", c.DatabaseID).Debugf("create new chain bdb %s", bdbFile) + leveldbInit.Do(func() { + // Open LevelDB for block and state + bdbFile := c.ChainFilePrefix + "-block-state.ldb" + blkDB, err = leveldb.OpenFile(bdbFile, &leveldbConf) + if err != nil { + err = errors.Wrapf(err, "open leveldb %s", bdbFile) + return + } + le.Debugf("opened chain bdb %s", bdbFile) - // Open LevelDB for ack/request/response - tdbFile := c.ChainFilePrefix + "-ack-req-resp.ldb" - tdb, err := leveldb.OpenFile(tdbFile, &leveldbConf) + // Open LevelDB for ack/request/response + tdbFile := c.ChainFilePrefix + "-ack-req-resp.ldb" + txDB, err = leveldb.OpenFile(tdbFile, &leveldbConf) + if err != nil { + err = errors.Wrapf(err, "open leveldb %s", tdbFile) + return + } + le.Debugf("opened chain tdb %s", tdbFile) + }) if err != nil { - err = errors.Wrapf(err, "open leveldb %s", tdbFile) return } - log.WithField("db", c.DatabaseID).Debugf("create new chain tdb %s", tdbFile) - // Open storage var strg xi.Storage if strg, err = xs.NewSqlite(c.DataFile); err != nil { + err = errors.Wrapf(err, "open data file %s", c.DataFile) return } @@ -184,95 +177,23 @@ func NewChainWithContext(ctx context.Context, c *Config) (chain *Chain, err erro } addr, err = crypto.PubKeyHash(pk.PubKey()) if err != nil { - log.WithError(err).WithField("db", c.DatabaseID).Warning("failed to generate addr in NewChain") - return - } - - // Create chain state - chain = &Chain{ - bdb: bdb, - tdb: tdb, - bi: newBlockIndex(), - ai: newAckIndex(), - st: x.NewState(sql.IsolationLevel(c.IsolationLevel), c.Server, strg), - cl: rpc.NewCaller(), - rt: newRunTime(ctx, c), - ctx: ctx, - blocks: make(chan *types.Block), - heights: make(chan int32, 1), - responses: make(chan *types.ResponseHeader), - acks: make(chan *types.AckHeader), - tokenType: c.TokenType, - gasPrice: c.GasPrice, - updatePeriod: c.UpdatePeriod, - databaseID: c.DatabaseID, - - pk: pk, - addr: &addr, - } - - if err = chain.pushBlock(c.Genesis); err != nil { - return nil, err - } - - return -} - -// LoadChain loads the chain state from the specified database and rebuilds a memory index. -func LoadChain(c *Config) (chain *Chain, err error) { - return LoadChainWithContext(context.Background(), c) -} - -// LoadChainWithContext loads the chain state from the specified database and rebuilds -// a memory index with context. -func LoadChainWithContext(ctx context.Context, c *Config) (chain *Chain, err error) { - // Open LevelDB for block and state - bdbFile := c.ChainFilePrefix + "-block-state.ldb" - bdb, err := leveldb.OpenFile(bdbFile, &leveldbConf) - if err != nil { - err = errors.Wrapf(err, "open leveldb %s", bdbFile) - return - } - - // Open LevelDB for ack/request/response - tdbFile := c.ChainFilePrefix + "-ack-req-resp.ldb" - tdb, err := leveldb.OpenFile(tdbFile, &leveldbConf) - if err != nil { - err = errors.Wrapf(err, "open leveldb %s", tdbFile) - return - } - - // Open x.State - var strg xi.Storage - if strg, err = xs.NewSqlite(c.DataFile); err != nil { + err = errors.Wrap(err, "failed to generate address") return } - // Cache local private key - var ( - pk *asymmetric.PrivateKey - addr proto.AccountAddress - ) - if pk, err = kms.GetLocalPrivateKey(); err != nil { - err = errors.Wrap(err, "failed to cache private key") - return - } - addr, err = crypto.PubKeyHash(pk.PubKey()) + metaKeyPrefix, err := c.DatabaseID.AccountAddress() if err != nil { - log.WithError(err).WithField("db", c.DatabaseID).Warning("failed to generate addr in LoadChain") + err = errors.Wrap(err, "failed to generate database meta prefix") return } // Create chain state chain = &Chain{ - bdb: bdb, - tdb: tdb, bi: newBlockIndex(), ai: newAckIndex(), st: x.NewState(sql.IsolationLevel(c.IsolationLevel), c.Server, strg), cl: rpc.NewCaller(), rt: newRunTime(ctx, c), - ctx: ctx, blocks: make(chan *types.Block), heights: make(chan int32, 1), responses: make(chan *types.ResponseHeader), @@ -282,41 +203,26 @@ func LoadChainWithContext(ctx context.Context, c *Config) (chain *Chain, err err updatePeriod: c.UpdatePeriod, databaseID: c.DatabaseID, - pk: pk, - addr: &addr, + pk: pk, + addr: &addr, + metaBlockIndex: utils.ConcatAll(metaKeyPrefix[:], metaBlockIndex[:]), + metaResponseIndex: utils.ConcatAll(metaKeyPrefix[:], metaResponseIndex[:]), + metaAckIndex: utils.ConcatAll(metaKeyPrefix[:], metaAckIndex[:]), } - - // Read state struct - stateEnc, err := chain.bdb.Get(metaState[:], nil) - if err != nil { - return nil, err - } - st := &state{} - if err = utils.DecodeMsgPack(stateEnc, st); err != nil { - return nil, err - } - - log.WithFields(log.Fields{ - "peer": chain.rt.getPeerInfoString(), - "state": st, - "db": c.DatabaseID, - }).Debug("loading state from database") + le = le.WithField("peer", chain.rt.getPeerInfoString()) // Read blocks and rebuild memory index var ( - id uint64 - index int32 - last *blockNode - blockIter = chain.bdb.NewIterator(util.BytesPrefix(metaBlockIndex[:]), nil) + id uint64 + last, parent *blockNode + blockIter = blkDB.NewIterator(util.BytesPrefix(chain.metaBlockIndex), nil) ) defer blockIter.Release() - for index = 0; blockIter.Next(); index++ { + for blockIter.Next() { var ( k = blockIter.Key() v = blockIter.Value() block = &types.Block{} - - current, parent *blockNode ) if err = utils.DecodeMsgPack(v, block); err != nil { @@ -324,11 +230,7 @@ func LoadChainWithContext(ctx context.Context, c *Config) (chain *Chain, err err keyWithSymbolToHeight(k), string(k)) return } - log.WithFields(log.Fields{ - "peer": chain.rt.getPeerInfoString(), - "block": block.BlockHash().String(), - "db": c.DatabaseID, - }).Debug("loading block from database") + le.WithField("block", block.BlockHash().String()).Debug("loading block from database") if last == nil { if err = block.VerifyAsGenesis(); err != nil { @@ -355,24 +257,34 @@ func LoadChainWithContext(ctx context.Context, c *Config) (chain *Chain, err err id = nid } - current = &blockNode{} - current.initBlockNode(chain.rt.getHeightFromTime(block.Timestamp()), block, parent) - chain.bi.addBlock(current) - last = current + last = newBlockNode(chain.rt.getHeightFromTime(block.Timestamp()), block, parent) + chain.bi.addBlock(last) } if err = blockIter.Error(); err != nil { - err = errors.Wrap(err, "load block") + err = errors.Wrap(err, "accumulated error of iterator") + return + } + + // Initiate chain Genesis if block list is empty + if last == nil { + if err = chain.genesis(c.Genesis); err != nil { + return nil, err + } return } // Set chain state - st.node = last - chain.rt.setHead(st) + var head = &state{ + node: last, + Head: last.hash, + Height: last.height, + } + chain.rt.setHead(head) chain.st.SetSeq(id) chain.pruneBlockCache() // Read queries and rebuild memory index - respIter := chain.tdb.NewIterator(util.BytesPrefix(metaResponseIndex[:]), nil) + respIter := txDB.NewIterator(util.BytesPrefix(chain.metaResponseIndex), nil) defer respIter.Release() for respIter.Next() { k := respIter.Key() @@ -394,7 +306,7 @@ func LoadChainWithContext(ctx context.Context, c *Config) (chain *Chain, err err return } - ackIter := chain.tdb.NewIterator(util.BytesPrefix(metaAckIndex[:]), nil) + ackIter := txDB.NewIterator(util.BytesPrefix(chain.metaAckIndex), nil) defer ackIter.Release() for ackIter.Next() { k := ackIter.Key() @@ -419,92 +331,86 @@ func LoadChainWithContext(ctx context.Context, c *Config) (chain *Chain, err err return } +func (c *Chain) genesis(b *types.Block) (err error) { + if b == nil { + err = errors.New("genesis block not provided") + return + } + if err = b.VerifyAsGenesis(); err != nil { + err = errors.Wrap(err, "initialize chain state") + return + } + return c.pushBlock(b) +} + // pushBlock pushes the signed block header to extend the current main chain. func (c *Chain) pushBlock(b *types.Block) (err error) { // Prepare and encode - h := c.rt.getHeightFromTime(b.Timestamp()) - node := newBlockNode(h, b, c.rt.getHead().node) - st := &state{ - node: node, - Head: node.hash, - Height: node.height, - } - var encBlock, encState *bytes.Buffer + var ( + h = c.rt.getHeightFromTime(b.Timestamp()) + node = newBlockNode(h, b, c.rt.getHead().node) + head = &state{ + node: node, + Head: node.hash, + Height: node.height, + } + blockKey = utils.ConcatAll(c.metaBlockIndex, node.indexKey()) + encBlock *bytes.Buffer + ) if encBlock, err = utils.EncodeMsgPack(b); err != nil { return } - if encState, err = utils.EncodeMsgPack(st); err != nil { - return - } - - // Update in transaction - t, err := c.bdb.OpenTransaction() - if err = t.Put(metaState[:], encState.Bytes(), nil); err != nil { - err = errors.Wrapf(err, "put %s", string(metaState[:])) - t.Discard() - return - } - blockKey := utils.ConcatAll(metaBlockIndex[:], node.indexKey()) - if err = t.Put(blockKey, encBlock.Bytes(), nil); err != nil { + // Put block + err = blkDB.Put(blockKey, encBlock.Bytes(), nil) + if err != nil { err = errors.Wrapf(err, "put %s", string(node.indexKey())) - t.Discard() - return - } - if err = t.Commit(); err != nil { - err = errors.Wrapf(err, "commit error") - t.Discard() return } - c.rt.setHead(st) + c.rt.setHead(head) c.bi.addBlock(node) // Keep track of the queries from the new block - var ierr error + var ( + ierr error + le = log.WithFields(log.Fields{ + "db": c.databaseID, + "producer": b.Producer(), + "block_hash": b.BlockHash(), + }) + ) for i, v := range b.QueryTxs { if ierr = c.AddResponse(v.Response); ierr != nil { - log.WithFields(log.Fields{ - "index": i, - "producer": b.Producer(), - "block_hash": b.BlockHash(), - "db": c.databaseID, - }).WithError(ierr).Warn("failed to add response to ackIndex") + le.WithFields(log.Fields{ + "index": i, + }).WithError(ierr).Warn("failed to add Response to ackIndex") } } for i, v := range b.Acks { if ierr = c.remove(v); ierr != nil { - log.WithFields(log.Fields{ - "index": i, - "producer": b.Producer(), - "block_hash": b.BlockHash(), - "db": c.databaseID, + le.WithFields(log.Fields{ + "index": i, }).WithError(ierr).Warn("failed to remove Ack from ackIndex") } } - if err == nil { - log.WithFields(log.Fields{ - "peer": c.rt.getPeerInfoString()[:14], - "time": c.rt.getChainTimeString(), - "block": b.BlockHash().String()[:8], - "producer": b.Producer()[:8], - "queryCount": len(b.QueryTxs), - "ackCount": len(b.Acks), - "blockTime": b.Timestamp().Format(time.RFC3339Nano), - "height": c.rt.getHeightFromTime(b.Timestamp()), - "head": fmt.Sprintf("%s <- %s", - func() string { - if st.node.parent != nil { - return st.node.parent.hash.String()[:8] - } - return "|" - }(), st.Head.String()[:8]), - "headHeight": c.rt.getHead().Height, - "db": c.databaseID, - }).Info("pushed new block") - } - + c.logEntry().WithFields(log.Fields{ + "block": b.BlockHash().String()[:8], + "producer": b.Producer()[:8], + "queryCount": len(b.QueryTxs), + "ackCount": len(b.Acks), + "blockTime": b.Timestamp().Format(time.RFC3339Nano), + "height": c.rt.getHeightFromTime(b.Timestamp()), + "head": fmt.Sprintf("%s <- %s", + func() string { + if head.node.parent != nil { + return head.node.parent.hash.String()[:8] + } + return "|" + }(), head.Head.String()[:8]), + "headHeight": c.rt.getHead().Height, + }).Info("pushed new block") return } @@ -519,14 +425,14 @@ func (c *Chain) pushAckedQuery(ack *types.SignedAckHeader) (err error) { return } - tdbKey := utils.ConcatAll(metaAckIndex[:], k, ack.Hash().AsBytes()) + tdbKey := utils.ConcatAll(c.metaAckIndex, k, ack.Hash().AsBytes()) if err = c.register(ack); err != nil { err = errors.Wrapf(err, "register ack %v at height %d", ack.Hash(), h) return } - if err = c.tdb.Put(tdbKey, enc.Bytes(), nil); err != nil { + if err = txDB.Put(tdbKey, enc.Bytes(), nil); err != nil { err = errors.Wrapf(err, "put ack %d %s", h, ack.Hash().String()) return } @@ -541,6 +447,11 @@ func (c *Chain) produceBlock(now time.Time) (err error) { qts []*x.QueryTracker ) if frs, qts, err = c.st.CommitEx(); err != nil { + err = errors.Wrap(err, "failed to fetch query list from db state") + return + } + if len(frs) == 0 && len(qts) == 0 { + c.logEntryWithHeadState().Debug("no query found in current period, skip block producing") return } var block = &types.Block{ @@ -558,11 +469,11 @@ func (c *Chain) produceBlock(now time.Time) (err error) { QueryTxs: make([]*types.QueryAsTx, len(qts)), Acks: c.ai.acks(c.rt.getHeightFromTime(now)), } - statBlock(block) + trackBlock(block) for i, v := range qts { // TODO(leventeliu): maybe block waiting at a ready channel instead? for !v.Ready() { - time.Sleep(1 * time.Millisecond) + time.Sleep(c.rt.period / 10) if c.rt.ctx.Err() != nil { err = c.rt.ctx.Err() return @@ -579,20 +490,18 @@ func (c *Chain) produceBlock(now time.Time) (err error) { return } // Send to pending list + le := c.logEntryWithHeadState().WithFields(log.Fields{ + "using_timestamp": now.Format(time.RFC3339Nano), + "block_hash": block.BlockHash().String(), + }) select { case c.blocks <- block: case <-c.rt.ctx.Done(): err = c.rt.ctx.Err() + le.WithError(err).Info("abort block producing") return } - log.WithFields(log.Fields{ - "peer": c.rt.getPeerInfoString(), - "time": c.rt.getChainTimeString(), - "curr_turn": c.rt.getNextTurn(), - "using_timestamp": now.Format(time.RFC3339Nano), - "block_hash": block.BlockHash().String(), - "db": c.databaseID, - }).Debug("produced new block") + le.Debug("produced new block") // Advise new block to the other peers var ( req = &MuxAdviseNewBlockReq{ @@ -625,14 +534,7 @@ func (c *Chain) produceBlock(now time.Time) (err error) { if err := c.cl.CallNodeWithContext( c.rt.ctx, id, route.SQLCAdviseNewBlock.String(), req, resp, ); err != nil { - log.WithFields(log.Fields{ - "peer": c.rt.getPeerInfoString(), - "time": c.rt.getChainTimeString(), - "curr_turn": c.rt.getNextTurn(), - "using_timestamp": now.Format(time.RFC3339Nano), - "block_hash": block.BlockHash().String(), - "db": c.databaseID, - }).WithError(err).Error("failed to advise new block") + le.WithError(err).Error("failed to advise new block") } }(s) } @@ -657,41 +559,27 @@ func (c *Chain) syncHead() { } resp := &MuxFetchBlockResp{} peers := c.rt.getPeers() + l := len(peers.Servers) succ := false + le := c.logEntryWithHeadState() for i, s := range peers.Servers { + ile := le.WithFields(log.Fields{"remote": fmt.Sprintf("[%d/%d] %s", i, l, s)}) if s != c.rt.getServer() { if err = c.cl.CallNode( s, route.SQLCFetchBlock.String(), req, resp, ); err != nil || resp.Block == nil { - log.WithFields(log.Fields{ - "peer": c.rt.getPeerInfoString(), - "time": c.rt.getChainTimeString(), - "remote": fmt.Sprintf("[%d/%d] %s", i, len(peers.Servers), s), - "curr_turn": c.rt.getNextTurn(), - "head_height": c.rt.getHead().Height, - "head_block": c.rt.getHead().Head.String(), - "db": c.databaseID, - }).WithError(err).Debug( - "Failed to fetch block from peer") + ile.WithError(err).Debug("failed to fetch block from peer") } else { - statBlock(resp.Block) + trackBlock(resp.Block) select { case c.blocks <- resp.Block: case <-c.rt.ctx.Done(): err = c.rt.ctx.Err() + le.WithError(err).Info("abort head block synchronizing") return } - log.WithFields(log.Fields{ - "peer": c.rt.getPeerInfoString(), - "time": c.rt.getChainTimeString(), - "remote": fmt.Sprintf("[%d/%d] %s", i, len(peers.Servers), s), - "curr_turn": c.rt.getNextTurn(), - "head_height": c.rt.getHead().Height, - "head_block": c.rt.getHead().Head.String(), - "db": c.databaseID, - }).Debug( - "Fetch block from remote peer successfully") + ile.Debug("fetch block from remote peer successfully") succ = true break } @@ -699,21 +587,18 @@ func (c *Chain) syncHead() { } if !succ { - log.WithFields(log.Fields{ - "peer": c.rt.getPeerInfoString(), - "time": c.rt.getChainTimeString(), - "curr_turn": c.rt.getNextTurn(), - "head_height": c.rt.getHead().Height, - "head_block": c.rt.getHead().Head.String(), - "db": c.databaseID, - }).Debug( - "Cannot get block from any peer") + le.Debug("cannot get block from any peer") } } } // runCurrentTurn does the check and runs block producing if its my turn. func (c *Chain) runCurrentTurn(now time.Time) { + h := c.rt.getNextTurn() + le := c.logEntryWithHeadState().WithFields(log.Fields{ + "using_timestamp": now.Format(time.RFC3339Nano), + }) + defer func() { c.stat() c.pruneBlockCache() @@ -721,44 +606,22 @@ func (c *Chain) runCurrentTurn(now time.Time) { c.ai.advance(c.rt.getMinValidHeight()) // Info the block processing goroutine that the chain height has grown, so please return // any stashed blocks for further check. - c.heights <- c.rt.getHead().Height + select { + case c.heights <- h: + case <-c.rt.ctx.Done(): + le.Debug("abort publishing height") + } }() - log.WithFields(log.Fields{ - "peer": c.rt.getPeerInfoString(), - "time": c.rt.getChainTimeString(), - "curr_turn": c.rt.getNextTurn(), - "head_height": c.rt.getHead().Height, - "head_block": c.rt.getHead().Head.String(), - "using_timestamp": now.Format(time.RFC3339Nano), - "db": c.databaseID, - }).Debug("run current turn") - + le.Debug("run current turn") if c.rt.getHead().Height < c.rt.getNextTurn()-1 { - log.WithFields(log.Fields{ - "peer": c.rt.getPeerInfoString(), - "time": c.rt.getChainTimeString(), - "curr_turn": c.rt.getNextTurn(), - "head_height": c.rt.getHead().Height, - "head_block": c.rt.getHead().Head.String(), - "using_timestamp": now.Format(time.RFC3339Nano), - "db": c.databaseID, - }).Error("A block will be skipped") + le.Error("a block will be skipped") } - if !c.rt.isMyTurn() { return } - if err := c.produceBlock(now); err != nil { - log.WithFields(log.Fields{ - "peer": c.rt.getPeerInfoString(), - "time": c.rt.getChainTimeString(), - "curr_turn": c.rt.getNextTurn(), - "using_timestamp": now.Format(time.RFC3339Nano), - "db": c.databaseID, - }).WithError(err).Error( - "Failed to produce block") + le.WithError(err).Error("failed to produce block") } } @@ -767,21 +630,11 @@ func (c *Chain) mainCycle(ctx context.Context) { for { select { case <-ctx.Done(): + c.logEntry().WithError(ctx.Err()).Info("abort main cycle") return default: c.syncHead() - if t, d := c.rt.nextTick(); d > 0 { - //log.WithFields(log.Fields{ - // "peer": c.rt.getPeerInfoString(), - // "time": c.rt.getChainTimeString(), - // "next_turn": c.rt.getNextTurn(), - // "head_height": c.rt.getHead().Height, - // "head_block": c.rt.getHead().Head.String(), - // "using_timestamp": t.Format(time.RFC3339Nano), - // "duration": d, - // "db": c.databaseID, - //}).Debug("main cycle") time.Sleep(d) } else { c.runCurrentTurn(t) @@ -791,28 +644,29 @@ func (c *Chain) mainCycle(ctx context.Context) { } // sync synchronizes blocks and queries from the other peers. -func (c *Chain) sync() (err error) { - log.WithFields(log.Fields{ - "peer": c.rt.getPeerInfoString(), - "time": c.rt.getChainTimeString(), - "db": c.databaseID, - }).Debug("synchronizing chain state") - +func (c *Chain) sync() { + le := c.logEntry() + le.Debug("synchronizing chain state") + defer func() { + c.stat() + c.pruneBlockCache() + c.ai.advance(c.rt.getMinValidHeight()) + }() for { now := c.rt.now() height := c.rt.getHeightFromTime(now) - - if c.rt.getNextTurn() >= height { + if now.Before(c.rt.chainInitTime) { + le.Debug("now time is before genesis time, waiting for genesis") + return + } + if c.rt.getNextTurn() > height { break } - for c.rt.getNextTurn() <= height { - // TODO(leventeliu): fetch blocks and queries. + c.syncHead() c.rt.setNextTurn() } } - - return } func (c *Chain) processBlocks(ctx context.Context) { @@ -823,10 +677,13 @@ func (c *Chain) processBlocks(ctx context.Context) { returnStash := func(stash []*types.Block) { defer wg.Done() - for _, block := range stash { + for i, block := range stash { select { case c.blocks <- block: case <-cld.Done(): + c.logEntry().WithFields(log.Fields{ + "remaining": len(stash) - i, + }).WithError(cld.Err()).Debug("abort stash returning") return } } @@ -837,17 +694,43 @@ func (c *Chain) processBlocks(ctx context.Context) { wg.Wait() }() - var ( - stash []*types.Block - ) + var stash []*types.Block for { + le := c.logEntryWithHeadState() select { case h := <-c.heights: + // Trigger billing + if uint64(h)%c.updatePeriod == 0 { + ub, err := c.billing(h, c.rt.getHead().node) + if err != nil { + le.WithError(err).Error("billing failed") + } + // allocate nonce + nonceReq := &types.NextAccountNonceReq{} + nonceResp := &types.NextAccountNonceResp{} + nonceReq.Addr = *c.addr + if err = rpc.RequestBP(route.MCCNextAccountNonce.String(), nonceReq, nonceResp); err != nil { + // allocate nonce failed + le.WithError(err).Warning("allocate nonce for transaction failed") + } + ub.Nonce = nonceResp.Nonce + if err = ub.Sign(c.pk); err != nil { + le.WithError(err).Warning("sign tx failed") + } + + addTxReq := &types.AddTxReq{TTL: 1} + addTxResp := &types.AddTxResp{} + addTxReq.Tx = ub + le.Debugf("nonce in processBlocks: %d, addr: %s", + addTxReq.Tx.GetAccountNonce(), addTxReq.Tx.GetAccountAddress()) + if err = rpc.RequestBP(route.MCCAddTx.String(), addTxReq, addTxResp); err != nil { + le.WithError(err).Warning("send tx failed") + } + } // Return all stashed blocks to pending channel - log.WithFields(log.Fields{ + c.logEntryWithHeadState().WithFields(log.Fields{ "height": h, "stashs": len(stash), - "db": c.databaseID, }).Debug("read new height from channel") if stash != nil { wg.Add(1) @@ -856,15 +739,9 @@ func (c *Chain) processBlocks(ctx context.Context) { } case block := <-c.blocks: height := c.rt.getHeightFromTime(block.Timestamp()) - log.WithFields(log.Fields{ - "peer": c.rt.getPeerInfoString(), - "time": c.rt.getChainTimeString(), - "curr_turn": c.rt.getNextTurn(), - "head_height": c.rt.getHead().Height, - "head_block": c.rt.getHead().Head.String(), + le.WithFields(log.Fields{ "block_height": height, "block_hash": block.BlockHash().String(), - "db": c.databaseID, }).Debug("processing new block") if height > c.rt.getNextTurn()-1 { @@ -876,50 +753,12 @@ func (c *Chain) processBlocks(ctx context.Context) { // TODO(leventeliu): check and add to fork list. } else { if err := c.CheckAndPushNewBlock(block); err != nil { - log.WithFields(log.Fields{ - "peer": c.rt.getPeerInfoString(), - "time": c.rt.getChainTimeString(), - "curr_turn": c.rt.getNextTurn(), - "head_height": c.rt.getHead().Height, - "head_block": c.rt.getHead().Head.String(), - "block_height": height, - "block_hash": block.BlockHash().String(), - "db": c.databaseID, - }).WithError(err).Error("Failed to check and push new block") - } else { - head := c.rt.getHead() - currentCount := uint64(head.node.count) - if currentCount%c.updatePeriod == 0 { - ub, err := c.billing(head.node) - if err != nil { - log.WithError(err).WithField("db", c.databaseID).Error("billing failed") - } - // allocate nonce - nonceReq := &types.NextAccountNonceReq{} - nonceResp := &types.NextAccountNonceResp{} - nonceReq.Addr = *c.addr - if err = rpc.RequestBP(route.MCCNextAccountNonce.String(), nonceReq, nonceResp); err != nil { - // allocate nonce failed - log.WithError(err).WithField("db", c.databaseID).Warning("allocate nonce for transaction failed") - } - ub.Nonce = nonceResp.Nonce - if err = ub.Sign(c.pk); err != nil { - log.WithError(err).WithField("db", c.databaseID).Warning("sign tx failed") - } - - addTxReq := &types.AddTxReq{TTL: 1} - addTxResp := &types.AddTxResp{} - addTxReq.Tx = ub - log.WithField("db", c.databaseID).Debugf("nonce in processBlocks: %d, addr: %s", - addTxReq.Tx.GetAccountNonce(), addTxReq.Tx.GetAccountAddress()) - if err = rpc.RequestBP(route.MCCAddTx.String(), addTxReq, addTxResp); err != nil { - log.WithError(err).WithField("db", c.databaseID).Warning("send tx failed") - } - } + le.WithError(err).Error("failed to check and push new block") } } } case <-ctx.Done(): + c.logEntryWithHeadState().WithError(ctx.Err()).Debug("abort block processing") return } } @@ -927,11 +766,8 @@ func (c *Chain) processBlocks(ctx context.Context) { // Start starts the main process of the sql-chain. func (c *Chain) Start() (err error) { - if err = c.sync(); err != nil { - return - } - c.rt.goFunc(c.processBlocks) + c.sync() c.rt.goFunc(c.mainCycle) c.rt.startService(c) return @@ -940,44 +776,16 @@ func (c *Chain) Start() (err error) { // Stop stops the main process of the sql-chain. func (c *Chain) Stop() (err error) { // Stop main process - log.WithFields(log.Fields{ - "peer": c.rt.getPeerInfoString(), - "time": c.rt.getChainTimeString(), - "db": c.databaseID, - }).Debug("stopping chain") + le := c.logEntry() + le.Debug("stopping chain") c.rt.stop(c.databaseID) - log.WithFields(log.Fields{ - "peer": c.rt.getPeerInfoString(), - "time": c.rt.getChainTimeString(), - "db": c.databaseID, - }).Debug("chain service and workers stopped") - // Close LevelDB file - var ierr error - if ierr = c.bdb.Close(); ierr != nil && err == nil { - err = ierr - } - log.WithFields(log.Fields{ - "peer": c.rt.getPeerInfoString(), - "time": c.rt.getChainTimeString(), - "db": c.databaseID, - }).WithError(ierr).Debug("chain database closed") - if ierr = c.tdb.Close(); ierr != nil && err == nil { - err = ierr - } - log.WithFields(log.Fields{ - "peer": c.rt.getPeerInfoString(), - "time": c.rt.getChainTimeString(), - "db": c.databaseID, - }).WithError(ierr).Debug("chain database closed") + le.Debug("chain service and workers stopped") // Close state + var ierr error if ierr = c.st.Close(false); ierr != nil && err == nil { err = ierr } - log.WithFields(log.Fields{ - "peer": c.rt.getPeerInfoString(), - "time": c.rt.getChainTimeString(), - "db": c.databaseID, - }).WithError(ierr).Debug("chain state storage closed") + le.WithError(ierr).Debug("chain state storage closed") return } @@ -1017,16 +825,16 @@ func (c *Chain) FetchBlockByCount(count int32) (b *types.Block, realCount int32, } func (c *Chain) fetchBlockByIndexKey(indexKey []byte) (b *types.Block, err error) { - k := utils.ConcatAll(metaBlockIndex[:], indexKey) + k := utils.ConcatAll(c.metaBlockIndex, indexKey) var v []byte - v, err = c.bdb.Get(k, nil) + v, err = blkDB.Get(k, nil) if err != nil { err = errors.Wrapf(err, "fetch block %s", string(k)) return } b = &types.Block{} - statBlock(b) + trackBlock(b) err = utils.DecodeMsgPack(v, b) if err != nil { err = errors.Wrapf(err, "fetch block %s", string(k)) @@ -1048,29 +856,27 @@ func (c *Chain) CheckAndPushNewBlock(block *types.Block) (err error) { } return -1 }() - log.WithFields(log.Fields{ - "peer": c.rt.getPeerInfoString(), - "time": c.rt.getChainTimeString(), + le := c.logEntryWithHeadState().WithFields(log.Fields{ "block": block.BlockHash().String(), "producer": block.Producer(), "blocktime": block.Timestamp().Format(time.RFC3339Nano), "blockheight": height, "blockparent": block.ParentHash().String(), - "headblock": head.Head.String(), - "headheight": head.Height, - "db": c.databaseID, - }).WithError(err).Debug("checking new block from other peer") + }) + le.Debug("checking new block from other peer") if head.Height == height && head.Head.IsEqual(block.BlockHash()) { // Maybe already set by FetchBlock return nil } else if !block.ParentHash().IsEqual(&head.Head) { - // Pushed block must extend the best chain + err = ErrInvalidBlock + le.WithError(err).Error("invalid new block for the current chain") return ErrInvalidBlock } // Verify block signatures if err = block.Verify(); err != nil { + le.WithError(err).Error("failed to verify block") return } @@ -1080,21 +886,19 @@ func (c *Chain) CheckAndPushNewBlock(block *types.Block) (err error) { } // Check block producer index, found := peers.Find(block.Producer()) - if !found { + err = ErrUnknownProducer + le.WithError(err).Error("unknown producer of new block") return ErrUnknownProducer } if index != next { - log.WithFields(log.Fields{ - "peer": c.rt.getPeerInfoString(), - "time": c.rt.getChainTimeString(), + err = ErrInvalidProducer + le.WithFields(log.Fields{ "expected": next, "actual": index, - "db": c.databaseID, - }).WithError(err).Error( - "Failed to check new block") - return ErrInvalidProducer + }).WithError(err).Error("invalid producer of new block") + return } // TODO(leventeliu): check if too many periods are skipped or store block for future use. @@ -1104,6 +908,7 @@ func (c *Chain) CheckAndPushNewBlock(block *types.Block) (err error) { // Replicate local state from the new block if err = c.st.ReplayBlockWithContext(c.rt.ctx, block); err != nil { + le.WithError(err).Error("failed to replay new block") return } @@ -1114,7 +919,9 @@ func (c *Chain) CheckAndPushNewBlock(block *types.Block) (err error) { func (c *Chain) VerifyAndPushAckedQuery(ack *types.SignedAckHeader) (err error) { // TODO(leventeliu): check ack. if c.rt.queryTimeIsExpired(ack.GetResponseTimestamp()) { - err = errors.Wrapf(ErrQueryExpired, "Verify ack query, min valid height %d, ack height %d", c.rt.getMinValidHeight(), c.rt.getHeightFromTime(ack.Timestamp)) + err = errors.Wrapf(ErrQueryExpired, + "Verify ack query, min valid height %d, ack height %d", + c.rt.getMinValidHeight(), c.rt.getHeightFromTime(ack.Timestamp)) return } @@ -1178,40 +985,43 @@ func (c *Chain) stat() { bc = atomic.LoadInt32(&cachedBlockCount) ) // Print chain stats - log.WithFields(log.Fields{ - "database_id": c.databaseID, + c.logEntry().WithFields(log.Fields{ "multiIndex_count": ic, "response_header_count": rc, "query_tracker_count": tc, "cached_block_count": bc, - "db": c.databaseID, }).Info("chain mem stats") // Print xeno stats c.st.Stat(c.databaseID) } -func (c *Chain) billing(node *blockNode) (ub *types.UpdateBilling, err error) { - log.WithField("db", c.databaseID).Debugf("begin to billing from count %d", node.count) +func (c *Chain) billing(h int32, node *blockNode) (ub *types.UpdateBilling, err error) { + le := c.logEntryWithHeadState() + le.WithFields(log.Fields{"given_height": h}).Info("begin to billing") var ( i, j uint64 + iter *blockNode minerAddr proto.AccountAddress userAddr proto.AccountAddress + minHeight = c.rt.getLastBillingHeight() usersMap = make(map[proto.AccountAddress]uint64) minersMap = make(map[proto.AccountAddress]map[proto.AccountAddress]uint64) ) - for i = 0; i < c.updatePeriod && node != nil; i++ { - var block = node.block + for iter = node; iter != nil && iter.height > h; iter = iter.parent { + } + for iter != nil && iter.height > minHeight { + var block = iter.block // Not cached, recover from storage if block == nil { - if block, err = c.FetchBlock(node.height); err != nil { + if block, err = c.FetchBlock(iter.height); err != nil { return } } for _, tx := range block.QueryTxs { minerAddr = tx.Response.ResponseAccount if userAddr, err = crypto.PubKeyHash(tx.Request.Header.Signee); err != nil { - log.WithError(err).WithField("db", c.databaseID).Warning("billing fail: miner addr") + le.WithError(err).Warning("billing fail: miner addr") return } @@ -1229,11 +1039,11 @@ func (c *Chain) billing(node *blockNode) (ub *types.UpdateBilling, err error) { for _, req := range block.FailedReqs { if minerAddr, err = crypto.PubKeyHash(block.Signee()); err != nil { - log.WithError(err).WithField("db", c.databaseID).Warning("billing fail: miner addr") + le.WithError(err).Warning("billing fail: miner addr") return } if userAddr, err = crypto.PubKeyHash(req.Header.Signee); err != nil { - log.WithError(err).WithField("db", c.databaseID).Warning("billing fail: user addr") + le.WithError(err).Warning("billing fail: user addr") return } if _, ok := minersMap[userAddr][minerAddr]; !ok { @@ -1243,17 +1053,18 @@ func (c *Chain) billing(node *blockNode) (ub *types.UpdateBilling, err error) { minersMap[userAddr][minerAddr] += uint64(len(req.Payload.Queries)) usersMap[userAddr] += uint64(len(req.Payload.Queries)) } - node = node.parent + iter = iter.parent } ub = types.NewUpdateBilling(&types.UpdateBillingHeader{ Users: make([]*types.UserCost, len(usersMap)), }) + ub.Version = int32(ub.HSPDefaultVersion()) i = 0 j = 0 for userAddr, cost := range usersMap { - log.WithField("db", c.databaseID).Debugf("user %s, cost %d", userAddr.String(), cost) + le.Debugf("user %s, cost %d", userAddr.String(), cost) ub.Users[i] = &types.UserCost{ User: userAddr, Cost: cost, @@ -1272,5 +1083,33 @@ func (c *Chain) billing(node *blockNode) (ub *types.UpdateBilling, err error) { i++ } ub.Receiver, err = c.databaseID.AccountAddress() + ub.Range.From = uint32(minHeight) + ub.Range.To = uint32(h) return } + +// SetLastBillingHeight sets the last billing height of this chain instance. +func (c *Chain) SetLastBillingHeight(h int32) { + c.logEntryWithHeadState().WithFields( + log.Fields{"new_height": h}).Debug("set last billing height") + c.rt.setLastBillingHeight(h) +} + +func (c *Chain) logEntry() *log.Entry { + return log.WithFields(log.Fields{ + "db": c.databaseID, + "peer": c.rt.getPeerInfoString(), + "offset": c.rt.getChainTimeString(), + }) +} + +func (c *Chain) logEntryWithHeadState() *log.Entry { + return log.WithFields(log.Fields{ + "db": c.databaseID, + "peer": c.rt.getPeerInfoString(), + "offset": c.rt.getChainTimeString(), + "curr_turn": c.rt.getNextTurn(), + "head_height": c.rt.getHead().Height, + "head_block": c.rt.getHead().Head.String(), + }) +} diff --git a/sqlchain/chain_test.go b/sqlchain/chain_test.go index b8c2e6b68..8c15d4066 100644 --- a/sqlchain/chain_test.go +++ b/sqlchain/chain_test.go @@ -33,7 +33,7 @@ import ( "github.com/CovenantSQL/CovenantSQL/proto" "github.com/CovenantSQL/CovenantSQL/route" "github.com/CovenantSQL/CovenantSQL/rpc" - "github.com/CovenantSQL/CovenantSQL/utils/log" + "github.com/CovenantSQL/CovenantSQL/types" ) var ( @@ -88,7 +88,7 @@ func TestIndexKey(t *testing.T) { } func TestMultiChain(t *testing.T) { - log.SetLevel(log.InfoLevel) + //log.SetLevel(log.InfoLevel) // Create genesis block genesis, err := createRandomBlock(genesisHash, true) @@ -104,7 +104,7 @@ func TestMultiChain(t *testing.T) { } for i, p := range peers.Servers { - t.Logf("Peer #%d: %s", i, p) + t.Logf("peer #%d: %s", i, p) } // Create config info from created nodes @@ -241,7 +241,9 @@ func TestMultiChain(t *testing.T) { for _, n := range conf.GConf.KnownNodes { rawNodeID := n.ID.ToRawNodeID() - route.SetNodeAddrCache(rawNodeID, n.Addr) + if err = route.SetNodeAddrCache(rawNodeID, n.Addr); err != nil { + t.Fatalf("error occurred: %v", err) + } node := &proto.Node{ ID: n.ID, Addr: n.Addr, @@ -265,7 +267,7 @@ func TestMultiChain(t *testing.T) { if chain, err := NewChain(p.config); err != nil { t.Errorf("error occurred: %v", err) } else { - t.Logf("Load chain from file %s: head = %s height = %d", + t.Logf("load chain from file %s: head = %s height = %d", p.dbfile, chain.rt.getHead().Head, chain.rt.getHead().Height) } }(v) @@ -279,7 +281,7 @@ func TestMultiChain(t *testing.T) { defer func(c *Chain) { // Stop chain main process before exit - c.Stop() + _ = c.Stop() }(v.chain) } @@ -290,30 +292,50 @@ func TestMultiChain(t *testing.T) { for i := int32(0); i <= ch; i++ { var node *blockNode if node = c.rt.getHead().node.ancestor(i); node == nil { - t.Logf("Block at height %d not found in peer %s, continue", + t.Logf("block at height %d not found in peer %s, continue", i, c.rt.getPeerInfoString()) continue } if node.block != nil { - t.Logf("Checking block %v at height %d in peer %s", + t.Logf("checking block %v at height %d in peer %s", node.block.BlockHash(), i, c.rt.getPeerInfoString()) } } }(v.chain) } + // Create table + cli, err := newRandomNode(chains[0].chain, true) + if err != nil { + t.Fatalf("error occurred: %v", err) + } + req, err := cli.buildQuery(types.WriteQuery, []types.Query{ + buildQuery(`CREATE TABLE t1 (k INT, v TEXT, PRIMARY KEY(k))`), + buildQuery(`INSERT INTO t1 (k, v) VALUES (?, ?), (?, ?), (?, ?), (?, ?), (?, ?)`, + 1, "v1", 2, "v2", 3, "v3", 4, "v4", 5, "v5", + ), + }) + if err != nil { + t.Fatalf("error occurred: %v", err) + } + for i, v := range chains { + cli, err := newRandomNode(v.chain, i == 0) + if err != nil { + t.Fatalf("error occurred: %v", err) + } + err = cli.sendQuery(req) + if err != nil { + t.Fatalf("error occurred: %v", err) + } + } + // Create some random clients to push new queries for i, v := range chains { sC := make(chan struct{}) wg := &sync.WaitGroup{} - wk := &nodeProfile{ - NodeID: peers.Servers[i], - PrivateKey: testPrivKey, - PublicKey: testPubKey, - } for j := 0; j < testClientNumberPerChain; j++ { - cli, err := newRandomNode() + cli, err := newRandomNode(v.chain, i == 0) if err != nil { t.Fatalf("error occurred: %v", err) @@ -328,22 +350,13 @@ func TestMultiChain(t *testing.T) { case <-sC: break foreverLoop default: + var err error // Send a random query - resp, err := createRandomQueryResponse(p, wk) - + err = cli.query(types.ReadQuery, []types.Query{ + buildQuery(`SELECT v FROM t1 WHERE k=?`, rand.Intn(5)), + }) if err != nil { t.Errorf("error occurred: %v", err) - } else if err = c.AddResponse(resp); err != nil { - t.Errorf("error occurred: %v", err) - } - - time.Sleep(time.Duration(rand.Int63n(500)+1) * time.Millisecond) - ack, err := createRandomQueryAckWithResponse(resp, p) - - if err != nil { - t.Errorf("error occurred: %v", err) - } else if err = c.VerifyAndPushAckedQuery(ack); err != nil { - t.Errorf("error occurred: %v", err) } } } diff --git a/sqlchain/config.go b/sqlchain/config.go index 3822870e7..769ceb8e4 100644 --- a/sqlchain/config.go +++ b/sqlchain/config.go @@ -37,20 +37,14 @@ type Config struct { Peers *proto.Peers Server proto.NodeID - // Price sets query price in gases. - Price map[types.QueryType]uint64 - ProducingReward uint64 - BillingPeriods int32 - // QueryTTL sets the unacknowledged query TTL in block periods. - QueryTTL int32 - + QueryTTL int32 BlockCacheTTL int32 // DBAccount info - TokenType types.TokenType - GasPrice uint64 - UpdatePeriod uint64 - - IsolationLevel int + TokenType types.TokenType + GasPrice uint64 + UpdatePeriod uint64 + LastBillingHeight int32 + IsolationLevel int } diff --git a/sqlchain/mirror/mirror_test.go b/sqlchain/mirror/mirror_test.go index 7980ebc77..0cc8759d0 100644 --- a/sqlchain/mirror/mirror_test.go +++ b/sqlchain/mirror/mirror_test.go @@ -193,7 +193,13 @@ func waitForMirrorComplete(ctx context.Context, dbID string, tick time.Duration, case <-time.After(tick): progressData, _ := ioutil.ReadFile(progressFile) progressCount, _ := strconv.Atoi(string(progressData)) + log.WithFields(log.Fields{ + "lastUpdate": lastUpdate.String(), + "stableDuration": stableDuration, + "progressCount": progressCount, + }).Infof("current mirror count progress") if progressCount > lastProgress { + lastProgress = progressCount lastUpdate = time.Now() } if progressCount > 5 || (progressCount > 0 && time.Now().Sub(lastUpdate) > stableDuration) { diff --git a/sqlchain/runtime.go b/sqlchain/runtime.go index 88c680d7e..9cb7dc441 100644 --- a/sqlchain/runtime.go +++ b/sqlchain/runtime.go @@ -28,6 +28,13 @@ import ( "github.com/CovenantSQL/CovenantSQL/utils/log" ) +// state represents a snapshot of current best chain. +type state struct { + node *blockNode + Head hash.Hash + Height int32 +} + // runtime represents a chain runtime state. type runtime struct { wg *sync.WaitGroup @@ -69,8 +76,8 @@ type runtime struct { nextTurn int32 // head is the current head of the best chain. head *state - // forks is the alternative head of the sql-chain. - forks []*state + // lastBillingHeight is the last success billing height of the current database. + lastBillingHeight int32 // timeMutex protects following time-relative fields. timeMutex sync.Mutex @@ -81,7 +88,7 @@ type runtime struct { } func blockCacheTTLRequired(c *Config) (ttl int32) { - var billingRequiredTTL = 2 * c.BillingPeriods + var billingRequiredTTL = int32(2 * c.UpdatePeriod) ttl = c.BlockCacheTTL if ttl < minBlockCacheTTL { ttl = minBlockCacheTTL @@ -119,10 +126,11 @@ func newRunTime(ctx context.Context, c *Config) (r *runtime) { return -1 }(), - total: int32(len(c.Peers.Servers)), - nextTurn: 1, - head: &state{}, - offset: time.Duration(0), + total: int32(len(c.Peers.Servers)), + nextTurn: 1, + head: &state{}, + lastBillingHeight: c.LastBillingHeight, + offset: time.Duration(0), } if c.Genesis != nil { @@ -138,7 +146,7 @@ func (r *runtime) setGenesis(b *types.Block) { r.head = &state{ node: nil, Head: *b.GenesisHash(), - Height: -1, + Height: 0, } } @@ -311,6 +319,18 @@ func (r *runtime) getPeers() *proto.Peers { return &peers } +func (r *runtime) getLastBillingHeight() int32 { + r.stateMutex.Lock() + defer r.stateMutex.Unlock() + return r.lastBillingHeight +} + +func (r *runtime) setLastBillingHeight(h int32) { + r.stateMutex.Lock() + defer r.stateMutex.Unlock() + r.lastBillingHeight = h +} + func (r *runtime) getHead() *state { r.stateMutex.Lock() defer r.stateMutex.Unlock() diff --git a/sqlchain/runtime_test.go b/sqlchain/runtime_test.go index 040dc6785..bd59a0d25 100644 --- a/sqlchain/runtime_test.go +++ b/sqlchain/runtime_test.go @@ -30,22 +30,22 @@ func TestBlockCacheTTL(t *testing.T) { }{ { config: &Config{ - BlockCacheTTL: 0, - BillingPeriods: 0, + BlockCacheTTL: 0, + UpdatePeriod: 0, }, expect: minBlockCacheTTL, }, { config: &Config{ - BlockCacheTTL: minBlockCacheTTL + 1, - BillingPeriods: 0, + BlockCacheTTL: minBlockCacheTTL + 1, + UpdatePeriod: 0, }, expect: minBlockCacheTTL + 1, }, { config: &Config{ - BlockCacheTTL: 0, - BillingPeriods: minBlockCacheTTL + 1, + BlockCacheTTL: 0, + UpdatePeriod: uint64(minBlockCacheTTL + 1), }, expect: 2 * (minBlockCacheTTL + 1), }, diff --git a/sqlchain/state.go b/sqlchain/state.go deleted file mode 100644 index 9f7c00593..000000000 --- a/sqlchain/state.go +++ /dev/null @@ -1,42 +0,0 @@ -/* - * Copyright 2018 The CovenantSQL Authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package sqlchain - -import ( - "github.com/CovenantSQL/CovenantSQL/crypto/hash" -) - -// state represents a snapshot of current best chain. -type state struct { - node *blockNode - Head hash.Hash - Height int32 -} - -//// MarshalHash marshals for hash -//func (s *state) MarshalHash() ([]byte, error) { -// buffer := bytes.NewBuffer(nil) -// -// if err := utils.WriteElements(buffer, binary.BigEndian, -// s.Head, -// s.Height, -// ); err != nil { -// return nil, err -// } -// -// return buffer.Bytes(), nil -//} diff --git a/sqlchain/state_test.go b/sqlchain/state_test.go deleted file mode 100644 index 622e8bce1..000000000 --- a/sqlchain/state_test.go +++ /dev/null @@ -1,60 +0,0 @@ -/* - * Copyright 2018 The CovenantSQL Authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package sqlchain - -import ( - "math/rand" - "reflect" - "testing" - - "github.com/CovenantSQL/CovenantSQL/crypto/hash" - "github.com/CovenantSQL/CovenantSQL/utils" -) - -func TestState(t *testing.T) { - st := &state{ - node: nil, - Head: hash.Hash{}, - Height: 0, - } - - rand.Read(st.Head[:]) - buffer, err := utils.EncodeMsgPack(st) - - if err != nil { - t.Fatalf("error occurred: %v", err) - } - - rState := &state{} - err = utils.DecodeMsgPack(buffer.Bytes(), rState) - - if err != nil { - t.Fatalf("error occurred: %v", err) - } - - //err = rState.UnmarshalBinary(nil) - // - //if err != nil { - // t.Logf("Error occurred as expected: %v", err) - //} else { - // t.Fatal("unexpected result: returned nil while expecting an error") - //} - - if !reflect.DeepEqual(st, rState) { - t.Fatalf("values don't match: v1 = %v, v2 = %v", st, rState) - } -} diff --git a/sqlchain/xxx_test.go b/sqlchain/xxx_test.go index 5f2fcb45e..994555fad 100644 --- a/sqlchain/xxx_test.go +++ b/sqlchain/xxx_test.go @@ -22,6 +22,7 @@ import ( "os" "path" "sync" + "sync/atomic" "testing" "time" @@ -39,6 +40,7 @@ var ( genesisHash = hash.Hash{} testDifficulty = 4 testMasterKey = []byte(".9K.sgch!3;C>w0v") + testConnIDSeed = rand.Uint64() testDataDir string testPrivKeyFile string testPubKeysFile string @@ -48,12 +50,16 @@ var ( ) type nodeProfile struct { - NodeID proto.NodeID - PrivateKey *asymmetric.PrivateKey - PublicKey *asymmetric.PublicKey + NodeID proto.NodeID + PrivateKey *asymmetric.PrivateKey + PublicKey *asymmetric.PublicKey + ConnectionID uint64 + SeqNo uint64 + Chain *Chain + IsLeader bool } -func newRandomNode() (node *nodeProfile, err error) { +func newRandomNode(chain *Chain, isLeader bool) (node *nodeProfile, err error) { priv, pub, err := asymmetric.GenSecp256k1KeyPair() if err != nil { @@ -64,9 +70,13 @@ func newRandomNode() (node *nodeProfile, err error) { rand.Read(h[:]) node = &nodeProfile{ - NodeID: proto.NodeID(h.String()), - PrivateKey: priv, - PublicKey: pub, + NodeID: proto.NodeID(h.String()), + PrivateKey: priv, + PublicKey: pub, + ConnectionID: atomic.AddUint64(&testConnIDSeed, 1), + SeqNo: rand.Uint64(), + Chain: chain, + IsLeader: isLeader, } return @@ -216,7 +226,7 @@ func registerNodesWithPublicKey(pub *asymmetric.PublicKey, diff int, num int) ( wg.Add(1) go func() { defer wg.Done() - miner.ComputeBlockNonce(block, next, diff) + _ = miner.ComputeBlockNonce(block, next, diff) }() n := <-nCh nis[i] = n @@ -369,3 +379,73 @@ func TestMain(m *testing.M) { return m.Run() }()) } + +func buildQuery(query string, args ...interface{}) types.Query { + var nargs = make([]types.NamedArg, len(args)) + for i := range args { + nargs[i] = types.NamedArg{ + Name: "", + Value: args[i], + } + } + return types.Query{ + Pattern: query, + Args: nargs, + } +} + +func (p *nodeProfile) buildQuery( + qt types.QueryType, qs []types.Query) (req *types.Request, err error, +) { + req = &types.Request{ + Header: types.SignedRequestHeader{ + RequestHeader: types.RequestHeader{ + QueryType: qt, + NodeID: p.NodeID, + DatabaseID: p.Chain.databaseID, + ConnectionID: p.ConnectionID, + SeqNo: atomic.AddUint64(&p.SeqNo, 1), + Timestamp: time.Now().UTC(), + // BatchCount and QueriesHash will be set by req.Sign() + }, + }, + Payload: types.RequestPayload{Queries: qs}, + } + if err = req.Sign(p.PrivateKey); err != nil { + return + } + return +} + +func (p *nodeProfile) sendQuery(req *types.Request) (err error) { + tracker, resp, err := p.Chain.Query(req, p.IsLeader) + if err != nil { + return + } + if err = resp.BuildHash(); err != nil { + return + } + if err = p.Chain.AddResponse(&resp.Header); err != nil { + return + } + tracker.UpdateResp(resp) + + ack, err := createRandomQueryAckWithResponse(&resp.Header, p) + if err != nil { + return + } + if err = p.Chain.VerifyAndPushAckedQuery(ack); err != nil { + return + } + return +} + +func (p *nodeProfile) query( + qt types.QueryType, qs []types.Query) (err error, +) { + req, err := p.buildQuery(qt, qs) + if err != nil { + return + } + return p.sendQuery(req) +} diff --git a/types/billing.go b/types/billing.go deleted file mode 100644 index a57ae9184..000000000 --- a/types/billing.go +++ /dev/null @@ -1,98 +0,0 @@ -/* - * Copyright 2018 The CovenantSQL Authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package types - -import ( - pi "github.com/CovenantSQL/CovenantSQL/blockproducer/interfaces" - "github.com/CovenantSQL/CovenantSQL/crypto/asymmetric" - "github.com/CovenantSQL/CovenantSQL/crypto/verifier" - "github.com/CovenantSQL/CovenantSQL/proto" -) - -//go:generate hsp - -// BillingHeader defines the customer's billing and block rewards in transaction. -type BillingHeader struct { - // Transaction nonce - Nonce pi.AccountNonce - BillingRequest BillingRequest - // Bill producer - Producer proto.AccountAddress - // Bill receivers - Receivers []*proto.AccountAddress - // Fee paid by stable coin - Fees []uint64 - // Reward is share coin - Rewards []uint64 -} - -// NewBillingHeader generates new BillingHeader. -func NewBillingHeader(nonce pi.AccountNonce, bReq *BillingRequest, producer proto.AccountAddress, receivers []*proto.AccountAddress, - fees []uint64, rewards []uint64) *BillingHeader { - return &BillingHeader{ - Nonce: nonce, - BillingRequest: *bReq, - Producer: producer, - Receivers: receivers, - Fees: fees, - Rewards: rewards, - } -} - -// Billing is a type of tx, that is used to record sql chain billing and block rewards. -type Billing struct { - BillingHeader - pi.TransactionTypeMixin - verifier.DefaultHashSignVerifierImpl -} - -// NewBilling generates a new Billing. -func NewBilling(header *BillingHeader) *Billing { - return &Billing{ - BillingHeader: *header, - TransactionTypeMixin: *pi.NewTransactionTypeMixin(pi.TransactionTypeBilling), - } -} - -// Sign implements interfaces/Transaction.Sign. -func (tb *Billing) Sign(signer *asymmetric.PrivateKey) (err error) { - return tb.DefaultHashSignVerifierImpl.Sign(&tb.BillingHeader, signer) -} - -// Verify implements interfaces/Transaction.Verify. -func (tb *Billing) Verify() error { - return tb.DefaultHashSignVerifierImpl.Verify(&tb.BillingHeader) -} - -// GetAccountAddress implements interfaces/Transaction.GetAccountAddress. -func (tb *Billing) GetAccountAddress() proto.AccountAddress { - return tb.Producer -} - -// GetAccountNonce implements interfaces/Transaction.GetAccountNonce. -func (tb *Billing) GetAccountNonce() pi.AccountNonce { - return tb.Nonce -} - -// GetDatabaseID gets the database ID. -func (tb *Billing) GetDatabaseID() proto.DatabaseID { - return tb.BillingRequest.Header.DatabaseID -} - -func init() { - pi.RegisterTransaction(pi.TransactionTypeBilling, (*Billing)(nil)) -} diff --git a/types/billing_gen.go b/types/billing_gen.go deleted file mode 100644 index 6d5d57f6b..000000000 --- a/types/billing_gen.go +++ /dev/null @@ -1,95 +0,0 @@ -package types - -// Code generated by github.com/CovenantSQL/HashStablePack DO NOT EDIT. - -import ( - hsp "github.com/CovenantSQL/HashStablePack/marshalhash" -) - -// MarshalHash marshals for hash -func (z *Billing) MarshalHash() (o []byte, err error) { - var b []byte - o = hsp.Require(b, z.Msgsize()) - // map header, size 3 - o = append(o, 0x83) - if oTemp, err := z.BillingHeader.MarshalHash(); err != nil { - return nil, err - } else { - o = hsp.AppendBytes(o, oTemp) - } - if oTemp, err := z.DefaultHashSignVerifierImpl.MarshalHash(); err != nil { - return nil, err - } else { - o = hsp.AppendBytes(o, oTemp) - } - if oTemp, err := z.TransactionTypeMixin.MarshalHash(); err != nil { - return nil, err - } else { - o = hsp.AppendBytes(o, oTemp) - } - return -} - -// Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message -func (z *Billing) Msgsize() (s int) { - s = 1 + 14 + z.BillingHeader.Msgsize() + 28 + z.DefaultHashSignVerifierImpl.Msgsize() + 21 + z.TransactionTypeMixin.Msgsize() - return -} - -// MarshalHash marshals for hash -func (z *BillingHeader) MarshalHash() (o []byte, err error) { - var b []byte - o = hsp.Require(b, z.Msgsize()) - // map header, size 6 - o = append(o, 0x86) - if oTemp, err := z.BillingRequest.MarshalHash(); err != nil { - return nil, err - } else { - o = hsp.AppendBytes(o, oTemp) - } - o = hsp.AppendArrayHeader(o, uint32(len(z.Fees))) - for za0002 := range z.Fees { - o = hsp.AppendUint64(o, z.Fees[za0002]) - } - if oTemp, err := z.Nonce.MarshalHash(); err != nil { - return nil, err - } else { - o = hsp.AppendBytes(o, oTemp) - } - if oTemp, err := z.Producer.MarshalHash(); err != nil { - return nil, err - } else { - o = hsp.AppendBytes(o, oTemp) - } - o = hsp.AppendArrayHeader(o, uint32(len(z.Receivers))) - for za0001 := range z.Receivers { - if z.Receivers[za0001] == nil { - o = hsp.AppendNil(o) - } else { - if oTemp, err := z.Receivers[za0001].MarshalHash(); err != nil { - return nil, err - } else { - o = hsp.AppendBytes(o, oTemp) - } - } - } - o = hsp.AppendArrayHeader(o, uint32(len(z.Rewards))) - for za0003 := range z.Rewards { - o = hsp.AppendUint64(o, z.Rewards[za0003]) - } - return -} - -// Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message -func (z *BillingHeader) Msgsize() (s int) { - s = 1 + 15 + z.BillingRequest.Msgsize() + 5 + hsp.ArrayHeaderSize + (len(z.Fees) * (hsp.Uint64Size)) + 6 + z.Nonce.Msgsize() + 9 + z.Producer.Msgsize() + 10 + hsp.ArrayHeaderSize - for za0001 := range z.Receivers { - if z.Receivers[za0001] == nil { - s += hsp.NilSize - } else { - s += z.Receivers[za0001].Msgsize() - } - } - s += 8 + hsp.ArrayHeaderSize + (len(z.Rewards) * (hsp.Uint64Size)) - return -} diff --git a/types/billing_gen_test.go b/types/billing_gen_test.go deleted file mode 100644 index 845a15213..000000000 --- a/types/billing_gen_test.go +++ /dev/null @@ -1,84 +0,0 @@ -package types - -// Code generated by github.com/CovenantSQL/HashStablePack DO NOT EDIT. - -import ( - "bytes" - "crypto/rand" - "encoding/binary" - "testing" -) - -func TestMarshalHashBilling(t *testing.T) { - v := Billing{} - binary.Read(rand.Reader, binary.BigEndian, &v) - bts1, err := v.MarshalHash() - if err != nil { - t.Fatal(err) - } - bts2, err := v.MarshalHash() - if err != nil { - t.Fatal(err) - } - if !bytes.Equal(bts1, bts2) { - t.Fatal("hash not stable") - } -} - -func BenchmarkMarshalHashBilling(b *testing.B) { - v := Billing{} - b.ReportAllocs() - b.ResetTimer() - for i := 0; i < b.N; i++ { - v.MarshalHash() - } -} - -func BenchmarkAppendMsgBilling(b *testing.B) { - v := Billing{} - bts := make([]byte, 0, v.Msgsize()) - bts, _ = v.MarshalHash() - b.SetBytes(int64(len(bts))) - b.ReportAllocs() - b.ResetTimer() - for i := 0; i < b.N; i++ { - bts, _ = v.MarshalHash() - } -} - -func TestMarshalHashBillingHeader(t *testing.T) { - v := BillingHeader{} - binary.Read(rand.Reader, binary.BigEndian, &v) - bts1, err := v.MarshalHash() - if err != nil { - t.Fatal(err) - } - bts2, err := v.MarshalHash() - if err != nil { - t.Fatal(err) - } - if !bytes.Equal(bts1, bts2) { - t.Fatal("hash not stable") - } -} - -func BenchmarkMarshalHashBillingHeader(b *testing.B) { - v := BillingHeader{} - b.ReportAllocs() - b.ResetTimer() - for i := 0; i < b.N; i++ { - v.MarshalHash() - } -} - -func BenchmarkAppendMsgBillingHeader(b *testing.B) { - v := BillingHeader{} - bts := make([]byte, 0, v.Msgsize()) - bts, _ = v.MarshalHash() - b.SetBytes(int64(len(bts))) - b.ReportAllocs() - b.ResetTimer() - for i := 0; i < b.N; i++ { - bts, _ = v.MarshalHash() - } -} diff --git a/types/billing_req.go b/types/billing_req.go deleted file mode 100644 index d3a50a210..000000000 --- a/types/billing_req.go +++ /dev/null @@ -1,33 +0,0 @@ -/* - * Copyright 2018 The CovenantSQL Authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package types - -import ( - "github.com/CovenantSQL/CovenantSQL/proto" -) - -// AdviseBillingReq defines a request of the AdviseBillingRequest RPC method. -type AdviseBillingReq struct { - proto.Envelope - Req *BillingRequest -} - -// AdviseBillingResp defines a request of the AdviseBillingRequest RPC method. -type AdviseBillingResp struct { - proto.Envelope - Resp *BillingRequest -} diff --git a/types/billing_request.go b/types/billing_request.go deleted file mode 100644 index 14d96c1b9..000000000 --- a/types/billing_request.go +++ /dev/null @@ -1,155 +0,0 @@ -/* - * Copyright 2018 The CovenantSQL Authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package types - -import ( - "reflect" - - "github.com/CovenantSQL/CovenantSQL/crypto/asymmetric" - "github.com/CovenantSQL/CovenantSQL/crypto/hash" - "github.com/CovenantSQL/CovenantSQL/proto" -) - -//go:generate hsp - -// BillingRequestHeader includes contents that need to be signed. Billing blocks should be within -// height range [low, high] (inclusive). -type BillingRequestHeader struct { - DatabaseID proto.DatabaseID - // sqlchain block hash and its height - LowBlock hash.Hash - LowHeight int32 - HighBlock hash.Hash - HighHeight int32 - GasAmounts []*proto.AddrAndGas -} - -// BillingRequest defines periodically Billing sync. -type BillingRequest struct { - Header BillingRequestHeader - RequestHash hash.Hash - Signees []*asymmetric.PublicKey - Signatures []*asymmetric.Signature -} - -// PackRequestHeader computes the hash of header. -func (br *BillingRequest) PackRequestHeader() (h *hash.Hash, err error) { - var enc []byte - if enc, err = br.Header.MarshalHash(); err != nil { - return - } - - br.RequestHash = hash.THashH(enc) - h = &br.RequestHash - return -} - -// SignRequestHeader first computes the hash of BillingRequestHeader, then signs the request. -func (br *BillingRequest) SignRequestHeader(signer *asymmetric.PrivateKey, calcHash bool) ( - signee *asymmetric.PublicKey, signature *asymmetric.Signature, err error) { - if calcHash { - if _, err = br.PackRequestHeader(); err != nil { - return - } - } - - if signature, err = signer.Sign(br.RequestHash[:]); err == nil { - // append to current signatures - signee = signer.PubKey() - br.Signees = append(br.Signees, signee) - br.Signatures = append(br.Signatures, signature) - } - - return -} - -// AddSignature add existing signature to BillingRequest, requires the structure to be packed first. -func (br *BillingRequest) AddSignature( - signee *asymmetric.PublicKey, signature *asymmetric.Signature, calcHash bool) (err error) { - if calcHash { - if _, err = br.PackRequestHeader(); err != nil { - return - } - } - - if !signature.Verify(br.RequestHash[:], signee) { - err = ErrSignVerification - return - } - - // append - br.Signees = append(br.Signees, signee) - br.Signatures = append(br.Signatures, signature) - - return -} - -// VerifySignatures verify existing signatures. -func (br *BillingRequest) VerifySignatures() (err error) { - if len(br.Signees) != len(br.Signatures) { - return ErrSignVerification - } - - var enc []byte - if enc, err = br.Header.MarshalHash(); err != nil { - return - } - - h := hash.THashH(enc) - if !br.RequestHash.IsEqual(&h) { - return ErrSignVerification - } - - if len(br.Signees) == 0 { - return - } - - for idx, signee := range br.Signees { - if !br.Signatures[idx].Verify(br.RequestHash[:], signee) { - return ErrSignVerification - } - } - - return -} - -// Compare returns if two billing records are identical. -func (br *BillingRequest) Compare(r *BillingRequest) (err error) { - if !br.Header.LowBlock.IsEqual(&r.Header.LowBlock) || - !br.Header.HighBlock.IsEqual(&br.Header.HighBlock) { - err = ErrBillingNotMatch - return - } - - reqMap := make(map[proto.AccountAddress]*proto.AddrAndGas) - locMap := make(map[proto.AccountAddress]*proto.AddrAndGas) - - for _, v := range br.Header.GasAmounts { - reqMap[v.AccountAddress] = v - } - - for _, v := range r.Header.GasAmounts { - locMap[v.AccountAddress] = v - } - - if !reflect.DeepEqual(reqMap, locMap) { - err = ErrBillingNotMatch - return - } - - return -} diff --git a/types/billing_request_gen.go b/types/billing_request_gen.go deleted file mode 100644 index 9a631c25f..000000000 --- a/types/billing_request_gen.go +++ /dev/null @@ -1,123 +0,0 @@ -package types - -// Code generated by github.com/CovenantSQL/HashStablePack DO NOT EDIT. - -import ( - hsp "github.com/CovenantSQL/HashStablePack/marshalhash" -) - -// MarshalHash marshals for hash -func (z *BillingRequest) MarshalHash() (o []byte, err error) { - var b []byte - o = hsp.Require(b, z.Msgsize()) - // map header, size 4 - o = append(o, 0x84) - if oTemp, err := z.Header.MarshalHash(); err != nil { - return nil, err - } else { - o = hsp.AppendBytes(o, oTemp) - } - if oTemp, err := z.RequestHash.MarshalHash(); err != nil { - return nil, err - } else { - o = hsp.AppendBytes(o, oTemp) - } - o = hsp.AppendArrayHeader(o, uint32(len(z.Signatures))) - for za0002 := range z.Signatures { - if z.Signatures[za0002] == nil { - o = hsp.AppendNil(o) - } else { - if oTemp, err := z.Signatures[za0002].MarshalHash(); err != nil { - return nil, err - } else { - o = hsp.AppendBytes(o, oTemp) - } - } - } - o = hsp.AppendArrayHeader(o, uint32(len(z.Signees))) - for za0001 := range z.Signees { - if z.Signees[za0001] == nil { - o = hsp.AppendNil(o) - } else { - if oTemp, err := z.Signees[za0001].MarshalHash(); err != nil { - return nil, err - } else { - o = hsp.AppendBytes(o, oTemp) - } - } - } - return -} - -// Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message -func (z *BillingRequest) Msgsize() (s int) { - s = 1 + 7 + z.Header.Msgsize() + 12 + z.RequestHash.Msgsize() + 11 + hsp.ArrayHeaderSize - for za0002 := range z.Signatures { - if z.Signatures[za0002] == nil { - s += hsp.NilSize - } else { - s += z.Signatures[za0002].Msgsize() - } - } - s += 8 + hsp.ArrayHeaderSize - for za0001 := range z.Signees { - if z.Signees[za0001] == nil { - s += hsp.NilSize - } else { - s += z.Signees[za0001].Msgsize() - } - } - return -} - -// MarshalHash marshals for hash -func (z *BillingRequestHeader) MarshalHash() (o []byte, err error) { - var b []byte - o = hsp.Require(b, z.Msgsize()) - // map header, size 6 - o = append(o, 0x86) - if oTemp, err := z.DatabaseID.MarshalHash(); err != nil { - return nil, err - } else { - o = hsp.AppendBytes(o, oTemp) - } - o = hsp.AppendArrayHeader(o, uint32(len(z.GasAmounts))) - for za0001 := range z.GasAmounts { - if z.GasAmounts[za0001] == nil { - o = hsp.AppendNil(o) - } else { - if oTemp, err := z.GasAmounts[za0001].MarshalHash(); err != nil { - return nil, err - } else { - o = hsp.AppendBytes(o, oTemp) - } - } - } - if oTemp, err := z.HighBlock.MarshalHash(); err != nil { - return nil, err - } else { - o = hsp.AppendBytes(o, oTemp) - } - o = hsp.AppendInt32(o, z.HighHeight) - if oTemp, err := z.LowBlock.MarshalHash(); err != nil { - return nil, err - } else { - o = hsp.AppendBytes(o, oTemp) - } - o = hsp.AppendInt32(o, z.LowHeight) - return -} - -// Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message -func (z *BillingRequestHeader) Msgsize() (s int) { - s = 1 + 11 + z.DatabaseID.Msgsize() + 11 + hsp.ArrayHeaderSize - for za0001 := range z.GasAmounts { - if z.GasAmounts[za0001] == nil { - s += hsp.NilSize - } else { - s += z.GasAmounts[za0001].Msgsize() - } - } - s += 10 + z.HighBlock.Msgsize() + 11 + hsp.Int32Size + 9 + z.LowBlock.Msgsize() + 10 + hsp.Int32Size - return -} diff --git a/types/billing_request_gen_test.go b/types/billing_request_gen_test.go deleted file mode 100644 index d46613c46..000000000 --- a/types/billing_request_gen_test.go +++ /dev/null @@ -1,84 +0,0 @@ -package types - -// Code generated by github.com/CovenantSQL/HashStablePack DO NOT EDIT. - -import ( - "bytes" - "crypto/rand" - "encoding/binary" - "testing" -) - -func TestMarshalHashBillingRequest(t *testing.T) { - v := BillingRequest{} - binary.Read(rand.Reader, binary.BigEndian, &v) - bts1, err := v.MarshalHash() - if err != nil { - t.Fatal(err) - } - bts2, err := v.MarshalHash() - if err != nil { - t.Fatal(err) - } - if !bytes.Equal(bts1, bts2) { - t.Fatal("hash not stable") - } -} - -func BenchmarkMarshalHashBillingRequest(b *testing.B) { - v := BillingRequest{} - b.ReportAllocs() - b.ResetTimer() - for i := 0; i < b.N; i++ { - v.MarshalHash() - } -} - -func BenchmarkAppendMsgBillingRequest(b *testing.B) { - v := BillingRequest{} - bts := make([]byte, 0, v.Msgsize()) - bts, _ = v.MarshalHash() - b.SetBytes(int64(len(bts))) - b.ReportAllocs() - b.ResetTimer() - for i := 0; i < b.N; i++ { - bts, _ = v.MarshalHash() - } -} - -func TestMarshalHashBillingRequestHeader(t *testing.T) { - v := BillingRequestHeader{} - binary.Read(rand.Reader, binary.BigEndian, &v) - bts1, err := v.MarshalHash() - if err != nil { - t.Fatal(err) - } - bts2, err := v.MarshalHash() - if err != nil { - t.Fatal(err) - } - if !bytes.Equal(bts1, bts2) { - t.Fatal("hash not stable") - } -} - -func BenchmarkMarshalHashBillingRequestHeader(b *testing.B) { - v := BillingRequestHeader{} - b.ReportAllocs() - b.ResetTimer() - for i := 0; i < b.N; i++ { - v.MarshalHash() - } -} - -func BenchmarkAppendMsgBillingRequestHeader(b *testing.B) { - v := BillingRequestHeader{} - bts := make([]byte, 0, v.Msgsize()) - bts, _ = v.MarshalHash() - b.SetBytes(int64(len(bts))) - b.ReportAllocs() - b.ResetTimer() - for i := 0; i < b.N; i++ { - bts, _ = v.MarshalHash() - } -} diff --git a/types/billing_request_test.go b/types/billing_request_test.go deleted file mode 100644 index c6d7ee998..000000000 --- a/types/billing_request_test.go +++ /dev/null @@ -1,339 +0,0 @@ -/* - * Copyright 2018 The CovenantSQL Authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package types - -import ( - "reflect" - "testing" - - "github.com/CovenantSQL/CovenantSQL/crypto/asymmetric" - "github.com/CovenantSQL/CovenantSQL/crypto/hash" - "github.com/CovenantSQL/CovenantSQL/proto" - "github.com/CovenantSQL/CovenantSQL/utils" - "github.com/CovenantSQL/CovenantSQL/utils/log" -) - -var ( - peerNum = 32 -) - -func TestBillingRequestHeader_MarshalUnmarshalBinary(t *testing.T) { - reqHeader := generateRandomBillingRequestHeader() - b, err := utils.EncodeMsgPack(reqHeader) - if err != nil { - t.Fatalf("unexpect error when marshal request header: %v", err) - } - - newReqHeader := &BillingRequestHeader{} - err = utils.DecodeMsgPack(b.Bytes(), newReqHeader) - if err != nil { - t.Fatalf("unexpect error when unmashll request header: %v", err) - } - - if !reflect.DeepEqual(reqHeader, newReqHeader) { - t.Fatalf("values not match:\n\tv0=%+v\n\tv1=%+v", reqHeader, newReqHeader) - } -} - -func TestBillingRequest_MarshalUnmarshalBinary(t *testing.T) { - req, err := generateRandomBillingRequest() - if err != nil { - t.Fatalf("unexpected error: %v", err) - } - - enc, err := utils.EncodeMsgPack(req) - if err != nil { - t.Fatalf("unexpected error: %v", err) - } - - dec := &BillingRequest{} - err = utils.DecodeMsgPack(enc.Bytes(), dec) - if err != nil { - t.Fatalf("unexpected error: %v", err) - } - - if !reflect.DeepEqual(req, dec) { - log.Debug(req) - log.Debug(dec) - t.Fatal("values not match") - } -} - -func TestBillingRequest_PackRequestHeader(t *testing.T) { - req, err := generateRandomBillingRequest() - if err != nil { - t.Fatalf("unexpected error: %v", err) - } - - enc, err := req.Header.MarshalHash() - if err != nil { - t.Fatalf("unexpected error: %v", err) - } - - h := hash.THashH(enc) - if !h.IsEqual(&req.RequestHash) { - t.Fatalf("hash not matched: \n\tv1=%v\n\tv2=%v", req.RequestHash, h) - } -} - -func TestBillingRequest_SignRequestHeader(t *testing.T) { - req, err := generateRandomBillingRequest() - if err != nil { - t.Fatalf("unexpected error: %v", err) - } - - enc, err := req.Header.MarshalHash() - if err != nil { - t.Fatalf("unexpected error: %v", err) - } - - h := hash.THashH(enc) - if !h.IsEqual(&req.RequestHash) { - t.Fatalf("hash not matched: \n\tv1=%v\n\tv2=%v", req.RequestHash, h) - } - - for i, sign := range req.Signatures { - if !sign.Verify(req.RequestHash[:], req.Signees[i]) { - - t.Fatalf("signature cannot match the hash and public key: %v", req) - } - } - - priv, pub, err := asymmetric.GenSecp256k1KeyPair() - _, sign, err := req.SignRequestHeader(priv, false) - if err != nil || !sign.Verify(req.RequestHash[:], pub) { - t.Fatalf("signature cannot match the hash and public key: %v", req) - } -} - -func TestBillingRequest_SignRequestHeader2(t *testing.T) { - header := generateRandomBillingRequestHeader() - req := &BillingRequest{ - Header: *header, - } - - priv, _, err := asymmetric.GenSecp256k1KeyPair() - signee, sign, err := req.SignRequestHeader(priv, true) - if err != nil || !sign.Verify(req.RequestHash[:], signee) { - t.Fatalf("signature cannot match the hash and public key: %v", req) - } -} - -func TestBillingRequest_AddSignature(t *testing.T) { - header := generateRandomBillingRequestHeader() - req := &BillingRequest{ - Header: *header, - } - - priv, _, err := asymmetric.GenSecp256k1KeyPair() - signee, sign, err := req.SignRequestHeader(priv, true) - if err != nil || !sign.Verify(req.RequestHash[:], signee) { - t.Fatalf("signature cannot match the hash and public key, req: %v, err: %v", req, err) - } - - // clear previous signees and signatures - req.Signees = req.Signees[:0] - req.Signatures = req.Signatures[:0] - - if err := req.AddSignature(signee, sign, false); err != nil { - t.Fatalf("add signature failed, req: %v, err: %v", req, err) - } -} - -func TestBillingRequest_AddSignature2(t *testing.T) { - header := generateRandomBillingRequestHeader() - req := &BillingRequest{ - Header: *header, - } - - priv, _, err := asymmetric.GenSecp256k1KeyPair() - signee, sign, err := req.SignRequestHeader(priv, true) - if err != nil || !sign.Verify(req.RequestHash[:], signee) { - t.Fatalf("signature cannot match the hash and public key, req: %v, err: %v", req, err) - } - - // clear previous signees and signatures - req.RequestHash = hash.Hash{} - req.Signees = req.Signees[:0] - req.Signatures = req.Signatures[:0] - - if err := req.AddSignature(signee, sign, true); err != nil { - t.Fatalf("add signature failed, req: %v, err: %v", req, err) - } -} - -func TestBillingRequest_AddSignature3(t *testing.T) { - header := generateRandomBillingRequestHeader() - req := &BillingRequest{ - Header: *header, - } - - priv, _, err := asymmetric.GenSecp256k1KeyPair() - signee, sign, err := req.SignRequestHeader(priv, true) - if err != nil || !sign.Verify(req.RequestHash[:], signee) { - t.Fatalf("signature cannot match the hash and public key, req: %v, err: %v", req, err) - } - - // clear previous signees and signatures - req.RequestHash = hash.Hash{} - req.Signees = req.Signees[:0] - req.Signatures = req.Signatures[:0] - - _, signee, _ = asymmetric.GenSecp256k1KeyPair() - if err := req.AddSignature(signee, sign, true); err != ErrSignVerification { - t.Fatalf("add signature should failed, req: %v, err: %v", req, err) - } -} - -func TestBillingRequest_VerifySignatures(t *testing.T) { - header := generateRandomBillingRequestHeader() - req := &BillingRequest{ - Header: *header, - } - - addSignature := func(calcHash bool) { - priv, _, err := asymmetric.GenSecp256k1KeyPair() - _, _, err = req.SignRequestHeader(priv, calcHash) - if err != nil { - t.Fatalf("sign request failed, req: %v, err: %v", req, err) - } - } - - // add 3 signatures - addSignature(true) - addSignature(false) - addSignature(false) - - if err := req.VerifySignatures(); err != nil { - t.Fatalf("verify signature failed, req: %v, err: %v", req, err) - } -} - -func TestBillingRequest_VerifySignatures2(t *testing.T) { - header := generateRandomBillingRequestHeader() - req := &BillingRequest{ - Header: *header, - } - - addSignature := func(calcHash bool) { - priv, _, err := asymmetric.GenSecp256k1KeyPair() - _, _, err = req.SignRequestHeader(priv, calcHash) - if err != nil { - t.Fatalf("sign request failed, req: %v, err: %v", req, err) - } - } - - // add 3 signatures - addSignature(true) - addSignature(false) - addSignature(false) - - // length invalidation - req.Signees = req.Signees[:0] - - if err := req.VerifySignatures(); err != ErrSignVerification { - t.Fatalf("verify should be failed, req: %v, err: %v", req, err) - } -} - -func TestBillingRequest_VerifySignatures3(t *testing.T) { - header := generateRandomBillingRequestHeader() - req := &BillingRequest{ - Header: *header, - } - - addSignature := func(calcHash bool) { - priv, _, err := asymmetric.GenSecp256k1KeyPair() - _, _, err = req.SignRequestHeader(priv, calcHash) - if err != nil { - t.Fatalf("sign request failed, req: %v, err: %v", req, err) - } - } - - // add 3 signatures - addSignature(true) - addSignature(false) - addSignature(false) - - // length invalidation - req.RequestHash = hash.Hash{} - - if err := req.VerifySignatures(); err != ErrSignVerification { - t.Fatalf("verify should be failed, req: %v, err: %v", req, err) - } -} - -func TestBillingRequest_VerifySignatures4(t *testing.T) { - header := generateRandomBillingRequestHeader() - req := &BillingRequest{ - Header: *header, - } - - addSignature := func(calcHash bool) { - priv, _, err := asymmetric.GenSecp256k1KeyPair() - _, _, err = req.SignRequestHeader(priv, calcHash) - if err != nil { - t.Fatalf("sign request failed, req: %v, err: %v", req, err) - } - } - - // add 3 signatures - addSignature(true) - addSignature(false) - addSignature(false) - - // length invalidation - _, req.Signees[0], _ = asymmetric.GenSecp256k1KeyPair() - - if err := req.VerifySignatures(); err == nil || err != ErrSignVerification { - t.Fatalf("verify should be failed, req: %v, err: %v", req, err) - } -} - -func TestBillingRequest_Compare(t *testing.T) { - req, _ := generateRandomBillingRequest() - - if err := req.Compare(req); err != nil { - t.Fatalf("compare failed, req: %v, err: %v", req, err) - } - - var req2 BillingRequest - req2 = *req - - req2.Header.LowBlock = hash.Hash{} - - if err := req.Compare(&req2); err != ErrBillingNotMatch { - t.Fatalf("compare should be failed, req: %v, req2: %v, err: %v", req, req2, err) - } -} - -func TestBillingRequest_Compare2(t *testing.T) { - req, _ := generateRandomBillingRequest() - var req2 BillingRequest - req2 = *req - - var gasAmount proto.AddrAndGas - gasAmount = *req.Header.GasAmounts[0] - gasAmount.GasAmount += 10 - req2.Header.GasAmounts = nil - req2.Header.GasAmounts = append(req2.Header.GasAmounts, &gasAmount) - req2.Header.GasAmounts = append(req2.Header.GasAmounts, req.Header.GasAmounts[1:]...) - - if err := req.Compare(&req2); err != ErrBillingNotMatch { - t.Fatalf("compare should be failed, req: %v, req2: %v, err: %v", req, req2, err) - } -} diff --git a/types/billing_test.go b/types/billing_test.go deleted file mode 100644 index 71ba70ec4..000000000 --- a/types/billing_test.go +++ /dev/null @@ -1,141 +0,0 @@ -/* - * Copyright 2018 The CovenantSQL Authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package types - -import ( - "reflect" - "testing" - - "github.com/CovenantSQL/CovenantSQL/crypto/asymmetric" - "github.com/CovenantSQL/CovenantSQL/crypto/hash" - "github.com/CovenantSQL/CovenantSQL/utils" -) - -func TestBillingHeader_MarshalUnmarshalBinary(t *testing.T) { - tc, err := generateRandomBillingHeader() - if err != nil { - t.Fatalf("unexpeted error: %v", err) - } - - enc, err := utils.EncodeMsgPack(tc) - if err != nil { - t.Fatalf("unexpeted error: %v", err) - } - - dec := &BillingHeader{} - err = utils.DecodeMsgPack(enc.Bytes(), dec) - if err != nil { - t.Fatalf("unexpeted error: %v", err) - } - - if tc.Nonce != dec.Nonce { - t.Fatalf("value not match: \n\tv1=%v\n\tv2=%v", tc.Nonce, tc.Nonce) - } - if tc.BillingRequest.RequestHash != dec.BillingRequest.RequestHash { - t.Fatalf("value not match: \n\tv1=%v\n\tv2=%v", tc.BillingRequest.RequestHash, tc.BillingRequest.RequestHash) - } - if !tc.BillingRequest.Signatures[0].IsEqual(dec.BillingRequest.Signatures[0]) { - t.Fatalf("value not match: \n\tv1=%v\n\tv2=%v", tc.BillingRequest.Signatures[0], dec.BillingRequest.Signatures[0]) - } - for i := range tc.Receivers { - if !reflect.DeepEqual(tc.Receivers[i], dec.Receivers[i]) { - t.Fatalf("value not match: \n\ttc.Receivers[%d]=%v\n\tReceive[%d]=%v", i, i, tc.Receivers[i], tc.Receivers[0]) - } - if tc.Rewards[i] != dec.Rewards[i] { - t.Fatalf("value not match: \n\ttc.Rewards[%d]=%v\n\tRewards[%d]=%v", i, i, tc.Rewards[i], tc.Rewards[0]) - } - if tc.Fees[i] != dec.Fees[i] { - t.Fatalf("value not match: \n\ttc.Fees[%d]=%v\n\tFees[%d]=%v", i, i, tc.Fees[i], tc.Fees[0]) - } - } -} - -func TestBilling_SerializeDeserialize(t *testing.T) { - tb, err := generateRandomBilling() - if err != nil { - t.Fatalf("unexpeted error: %v", err) - } - - enc, err := utils.EncodeMsgPack(tb) - if err != nil { - t.Fatalf("unexpeted error: %v", err) - } - - dec := Billing{} - err = utils.DecodeMsgPack(enc.Bytes(), &dec) - if err != nil { - t.Fatalf("unexpeted error: %v", err) - } - - if !tb.Signature.IsEqual(dec.Signature) { - t.Fatalf("value not match: \n\tv1=%v\n\tv2=%v", tb.Signature, dec.Signature) - } - if !tb.Signee.IsEqual(dec.Signee) { - t.Fatalf("value not match: \n\tv1=%v\n\tv2=%v", tb.Signee, dec.Signee) - } - if tb.Hash() != dec.Hash() { - t.Fatalf("value not match: \n\tv1=%v\n\tv2=%v", tb.Hash(), dec.Hash()) - } -} - -func TestBilling_PackAndSignTx(t *testing.T) { - tb, err := generateRandomBilling() - if err != nil { - t.Fatalf("unexpeted error: %v", err) - } - - priv, _, err := asymmetric.GenSecp256k1KeyPair() - if err != nil { - t.Fatalf("unexpeted error: %v", err) - } - tb.Sign(priv) - enc, err := tb.BillingHeader.MarshalHash() - if err != nil { - t.Fatalf("unexpeted error: %v", err) - } - h := hash.THashH(enc[:]) - sign, err := priv.Sign(h[:]) - if err != nil { - t.Fatalf("unexpeted error: %v", err) - } - if !sign.IsEqual(tb.Signature) { - t.Fatalf("value not match: \n\tv1=%v\n\tv2=%v", sign, tb.Signature) - } - - err = tb.Verify() - if err != nil { - t.Fatalf("verify signature failed: %v", err) - } - - // get - addr := hash.Hash(tb.GetAccountAddress()) - if addr.IsEqual(&hash.Hash{}) { - t.Fatal("get hash failed") - } - - tb.GetAccountNonce() - - if len(tb.GetDatabaseID()) == 0 { - t.Fatal("get empty DatabaseID") - } - - tb.Signature = nil - err = tb.Verify() - if err == nil { - t.Fatal("verify signature should failed") - } -} diff --git a/types/bp_block_test.go b/types/bp_block_test.go index 14d6d6d96..24025381d 100644 --- a/types/bp_block_test.go +++ b/types/bp_block_test.go @@ -164,7 +164,7 @@ func TestBlock_PackAndSignBlock(t *testing.T) { t.Fatalf("unexpected error: %v", err) } - tb, err := generateRandomBilling() + tb, err := generateRandomTransfer() if err != nil { t.Fatalf("unexpected error: %v", err) } diff --git a/types/bprpc.go b/types/bprpc.go index a68025267..73a74593c 100644 --- a/types/bprpc.go +++ b/types/bprpc.go @@ -34,17 +34,6 @@ type AdviseNewBlockResp struct { proto.Envelope } -// AdviseTxBillingReq defines a request of the AdviseTxBilling RPC method. -type AdviseTxBillingReq struct { - proto.Envelope - TxBilling *Billing -} - -// AdviseTxBillingResp defines a response of the AdviseTxBilling RPC method. -type AdviseTxBillingResp struct { - proto.Envelope -} - // FetchBlockReq defines a request of the FetchBlock RPC method. type FetchBlockReq struct { proto.Envelope diff --git a/types/msgpack_test.go b/types/msgpack_test.go index 248a889b9..0698b9bc9 100644 --- a/types/msgpack_test.go +++ b/types/msgpack_test.go @@ -50,7 +50,6 @@ func TestEncodeDecodeTransactions(t *testing.T) { var t []pi.Transaction t = append(t, NewBaseAccount(&Account{})) t = append(t, NewTransfer(&TransferHeader{})) - t = append(t, NewBilling(&BillingHeader{})) t = append(t, NewCreateDatabase(&CreateDatabaseHeader{})) buf, err := utils.EncodeMsgPack(t) @@ -73,12 +72,10 @@ func TestEncodeDecodeTransactions(t *testing.T) { t.Tx = NewBaseAccount(&Account{}) t.Txs = append(t.Txs, NewBaseAccount(&Account{})) t.Txs = append(t.Txs, NewTransfer(&TransferHeader{})) - t.Txs = append(t.Txs, NewBilling(&BillingHeader{})) t.Txs = append(t.Txs, NewCreateDatabase(&CreateDatabaseHeader{})) t.Maps = make(map[string]pi.Transaction) t.Maps["BaseAccount"] = NewBaseAccount(&Account{}) t.Maps["Transfer"] = NewTransfer(&TransferHeader{}) - t.Maps["Billing"] = NewBilling(&BillingHeader{}) t.Maps["CreateDatabase"] = NewCreateDatabase(&CreateDatabaseHeader{}) buf, err := utils.EncodeMsgPack(t) So(err, ShouldBeNil) diff --git a/types/updatebilling.go b/types/updatebilling.go index 366b59378..144bfc91a 100644 --- a/types/updatebilling.go +++ b/types/updatebilling.go @@ -26,6 +26,11 @@ import ( //go:generate hsp +// Range defines a height range (from, to]. +type Range struct { + From, To uint32 +} + // MinerIncome defines the income of miner. type MinerIncome struct { Miner proto.AccountAddress @@ -44,6 +49,8 @@ type UpdateBillingHeader struct { Receiver proto.AccountAddress Nonce pi.AccountNonce Users []*UserCost + Range Range + Version int32 `hsp:"v,version"` } // UpdateBilling defines the UpdateBilling transaction. diff --git a/types/updatebilling_gen.go b/types/updatebilling_gen.go index 79a626ffb..b14103749 100644 --- a/types/updatebilling_gen.go +++ b/types/updatebilling_gen.go @@ -3,6 +3,8 @@ package types // Code generated by github.com/CovenantSQL/HashStablePack DO NOT EDIT. import ( + herr "errors" + hsp "github.com/CovenantSQL/HashStablePack/marshalhash" ) @@ -27,6 +29,23 @@ func (z *MinerIncome) Msgsize() (s int) { return } +// MarshalHash marshals for hash +func (z Range) MarshalHash() (o []byte, err error) { + var b []byte + o = hsp.Require(b, z.Msgsize()) + // map header, size 2 + o = append(o, 0x82) + o = hsp.AppendUint32(o, z.From) + o = hsp.AppendUint32(o, z.To) + return +} + +// Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message +func (z Range) Msgsize() (s int) { + s = 1 + 5 + hsp.Uint32Size + 3 + hsp.Uint32Size + return +} + // MarshalHash marshals for hash func (z *UpdateBilling) MarshalHash() (o []byte, err error) { var b []byte @@ -57,46 +76,49 @@ func (z *UpdateBilling) Msgsize() (s int) { return } +var hspVersionsUpdateBillingHeader = []string{ + "oldver", + "9ef447", +} + +// HSPCurrentVersion returns current struct version +func (z *UpdateBillingHeader) HSPCurrentVersion() int { + return int(z.Version) +} + +// HSPMaxVersion returns max struct version +func (z *UpdateBillingHeader) HSPMaxVersion() int { + return 1 +} + +// HSPDefaultVersion returns default struct version +func (z *UpdateBillingHeader) HSPDefaultVersion() int { + return 1 +} + // MarshalHash marshals for hash func (z *UpdateBillingHeader) MarshalHash() (o []byte, err error) { - var b []byte - o = hsp.Require(b, z.Msgsize()) - // map header, size 3 - o = append(o, 0x83) - if oTemp, err := z.Nonce.MarshalHash(); err != nil { - return nil, err - } else { - o = hsp.AppendBytes(o, oTemp) - } - if oTemp, err := z.Receiver.MarshalHash(); err != nil { - return nil, err - } else { - o = hsp.AppendBytes(o, oTemp) - } - o = hsp.AppendArrayHeader(o, uint32(len(z.Users))) - for za0001 := range z.Users { - if z.Users[za0001] == nil { - o = hsp.AppendNil(o) - } else { - if oTemp, err := z.Users[za0001].MarshalHash(); err != nil { - return nil, err - } else { - o = hsp.AppendBytes(o, oTemp) - } - } + switch z.HSPCurrentVersion() { + case 0: + return z.MarshalHasholdver() + case 1: + return z.MarshalHash9ef447() + default: + err = herr.New("invalid struct version") + return } return } // Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message func (z *UpdateBillingHeader) Msgsize() (s int) { - s = 1 + 6 + z.Nonce.Msgsize() + 9 + z.Receiver.Msgsize() + 6 + hsp.ArrayHeaderSize - for za0001 := range z.Users { - if z.Users[za0001] == nil { - s += hsp.NilSize - } else { - s += z.Users[za0001].Msgsize() - } + switch z.HSPCurrentVersion() { + case 0: + return z.Msgsizeoldver() + case 1: + return z.Msgsize9ef447() + default: + return 0 } return } diff --git a/types/updatebilling_gen_test.go b/types/updatebilling_gen_test.go index d6ab30c03..8881dc0d3 100644 --- a/types/updatebilling_gen_test.go +++ b/types/updatebilling_gen_test.go @@ -46,6 +46,43 @@ func BenchmarkAppendMsgMinerIncome(b *testing.B) { } } +func TestMarshalHashRange(t *testing.T) { + v := Range{} + binary.Read(rand.Reader, binary.BigEndian, &v) + bts1, err := v.MarshalHash() + if err != nil { + t.Fatal(err) + } + bts2, err := v.MarshalHash() + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(bts1, bts2) { + t.Fatal("hash not stable") + } +} + +func BenchmarkMarshalHashRange(b *testing.B) { + v := Range{} + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + v.MarshalHash() + } +} + +func BenchmarkAppendMsgRange(b *testing.B) { + v := Range{} + bts := make([]byte, 0, v.Msgsize()) + bts, _ = v.MarshalHash() + b.SetBytes(int64(len(bts))) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + bts, _ = v.MarshalHash() + } +} + func TestMarshalHashUpdateBilling(t *testing.T) { v := UpdateBilling{} binary.Read(rand.Reader, binary.BigEndian, &v) diff --git a/types/updatebilling_updatebillingheader_9ef447_gen.go b/types/updatebilling_updatebillingheader_9ef447_gen.go new file mode 100644 index 000000000..d90f54b01 --- /dev/null +++ b/types/updatebilling_updatebillingheader_9ef447_gen.go @@ -0,0 +1,57 @@ +package types + +// Code generated by github.com/CovenantSQL/HashStablePack DO NOT EDIT. + +import ( + hsp "github.com/CovenantSQL/HashStablePack/marshalhash" +) + +// MarshalHash9ef447 marshals for hash +func (z *UpdateBillingHeader) MarshalHash9ef447() (o []byte, err error) { + var b []byte + o = hsp.Require(b, z.Msgsize9ef447()) + // map header, size 5 + o = append(o, 0x85) + if oTemp, err := z.Nonce.MarshalHash(); err != nil { + return nil, err + } else { + o = hsp.AppendBytes(o, oTemp) + } + // map header, size 2 + o = append(o, 0x82) + o = hsp.AppendUint32(o, z.Range.From) + o = hsp.AppendUint32(o, z.Range.To) + if oTemp, err := z.Receiver.MarshalHash(); err != nil { + return nil, err + } else { + o = hsp.AppendBytes(o, oTemp) + } + o = hsp.AppendArrayHeader(o, uint32(len(z.Users))) + for za0001 := range z.Users { + if z.Users[za0001] == nil { + o = hsp.AppendNil(o) + } else { + if oTemp, err := z.Users[za0001].MarshalHash(); err != nil { + return nil, err + } else { + o = hsp.AppendBytes(o, oTemp) + } + } + } + o = hsp.AppendInt32(o, z.Version) + return +} + +// Msgsize9ef447 returns an upper bound estimate of the number of bytes occupied by the serialized message +func (z *UpdateBillingHeader) Msgsize9ef447() (s int) { + s = 1 + 6 + z.Nonce.Msgsize() + 6 + 1 + 5 + hsp.Uint32Size + 3 + hsp.Uint32Size + 9 + z.Receiver.Msgsize() + 6 + hsp.ArrayHeaderSize + for za0001 := range z.Users { + if z.Users[za0001] == nil { + s += hsp.NilSize + } else { + s += z.Users[za0001].Msgsize() + } + } + s += 2 + hsp.Int32Size + return +} diff --git a/types/updatebilling_updatebillingheader_9ef447_gen_test.go b/types/updatebilling_updatebillingheader_9ef447_gen_test.go new file mode 100644 index 000000000..3f48bd2e1 --- /dev/null +++ b/types/updatebilling_updatebillingheader_9ef447_gen_test.go @@ -0,0 +1,47 @@ +package types + +// Code generated by github.com/CovenantSQL/HashStablePack DO NOT EDIT. + +import ( + "bytes" + "crypto/rand" + "encoding/binary" + "testing" +) + +func TestMarshalHash9ef447UpdateBillingHeader(t *testing.T) { + v := UpdateBillingHeader{} + binary.Read(rand.Reader, binary.BigEndian, &v) + bts1, err := v.MarshalHash9ef447() + if err != nil { + t.Fatal(err) + } + bts2, err := v.MarshalHash9ef447() + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(bts1, bts2) { + t.Fatal("hash not stable") + } +} + +func BenchmarkMarshalHash9ef447UpdateBillingHeader(b *testing.B) { + v := UpdateBillingHeader{} + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + v.MarshalHash9ef447() + } +} + +func BenchmarkAppendMsg9ef447UpdateBillingHeader(b *testing.B) { + v := UpdateBillingHeader{} + bts := make([]byte, 0, v.Msgsize9ef447()) + bts, _ = v.MarshalHash9ef447() + b.SetBytes(int64(len(bts))) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + bts, _ = v.MarshalHash9ef447() + } +} diff --git a/types/updatebilling_updatebillingheader_oldver_gen.go b/types/updatebilling_updatebillingheader_oldver_gen.go new file mode 100644 index 000000000..681658afd --- /dev/null +++ b/types/updatebilling_updatebillingheader_oldver_gen.go @@ -0,0 +1,51 @@ +package types + +// Code generated by github.com/CovenantSQL/HashStablePack DO NOT EDIT. + +import ( + hsp "github.com/CovenantSQL/HashStablePack/marshalhash" +) + +// MarshalHasholdver marshals for hash +func (z *UpdateBillingHeader) MarshalHasholdver() (o []byte, err error) { + var b []byte + o = hsp.Require(b, z.Msgsize()) + + o = append(o, 0x83) + if oTemp, err := z.Nonce.MarshalHash(); err != nil { + return nil, err + } else { + o = hsp.AppendBytes(o, oTemp) + } + if oTemp, err := z.Receiver.MarshalHash(); err != nil { + return nil, err + } else { + o = hsp.AppendBytes(o, oTemp) + } + o = hsp.AppendArrayHeader(o, uint32(len(z.Users))) + for za0001 := range z.Users { + if z.Users[za0001] == nil { + o = hsp.AppendNil(o) + } else { + if oTemp, err := z.Users[za0001].MarshalHash(); err != nil { + return nil, err + } else { + o = hsp.AppendBytes(o, oTemp) + } + } + } + return +} + +// Msgsizeoldver returns an upper bound estimate of the number of bytes occupied by the serialized message +func (z *UpdateBillingHeader) Msgsizeoldver() (s int) { + s = 1 + 6 + z.Nonce.Msgsize() + 9 + z.Receiver.Msgsize() + 6 + hsp.ArrayHeaderSize + for za0001 := range z.Users { + if z.Users[za0001] == nil { + s += hsp.NilSize + } else { + s += z.Users[za0001].Msgsize() + } + } + return +} diff --git a/types/updatebilling_updatebillingheader_oldver_gen_test.go b/types/updatebilling_updatebillingheader_oldver_gen_test.go new file mode 100644 index 000000000..818fb978d --- /dev/null +++ b/types/updatebilling_updatebillingheader_oldver_gen_test.go @@ -0,0 +1,47 @@ +package types + +// Code generated by github.com/CovenantSQL/HashStablePack DO NOT EDIT. + +import ( + "bytes" + "crypto/rand" + "encoding/binary" + "testing" +) + +func TestMarshalHasholdverUpdateBillingHeader(t *testing.T) { + v := UpdateBillingHeader{} + binary.Read(rand.Reader, binary.BigEndian, &v) + bts1, err := v.MarshalHasholdver() + if err != nil { + t.Fatal(err) + } + bts2, err := v.MarshalHasholdver() + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(bts1, bts2) { + t.Fatal("hash not stable") + } +} + +func BenchmarkMarshalHasholdverUpdateBillingHeader(b *testing.B) { + v := UpdateBillingHeader{} + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + v.MarshalHasholdver() + } +} + +func BenchmarkAppendMsgoldverUpdateBillingHeader(b *testing.B) { + v := UpdateBillingHeader{} + bts := make([]byte, 0, v.Msgsizeoldver()) + bts, _ = v.MarshalHasholdver() + b.SetBytes(int64(len(bts))) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + bts, _ = v.MarshalHasholdver() + } +} diff --git a/types/xxx_test.go b/types/xxx_test.go index a15dd8666..42d719e19 100644 --- a/types/xxx_test.go +++ b/types/xxx_test.go @@ -60,6 +60,32 @@ func randStringBytes(n int) string { return string(b) } +func generateRandomTransferHeader() (header *TransferHeader, err error) { + header = &TransferHeader{ + Nonce: pi.AccountNonce(rand.Uint64()), + Amount: rand.Uint64(), + TokenType: TokenType(rand.Intn(int(SupportTokenNumber))), + } + return +} + +func generateRandomTransfer() (tx *Transfer, err error) { + header, err := generateRandomTransferHeader() + if err != nil { + return + + } + priv, _, err := asymmetric.GenSecp256k1KeyPair() + if err != nil { + return + } + tx = NewTransfer(header) + if err = tx.Sign(priv); err != nil { + return + } + return +} + func generateRandomBlock(parent hash.Hash, isGenesis bool) (b *BPBlock, err error) { // Generate key pair priv, _, err := asymmetric.GenSecp256k1KeyPair() @@ -83,7 +109,7 @@ func generateRandomBlock(parent hash.Hash, isGenesis bool) (b *BPBlock, err erro } for i, n := 0, rand.Intn(10)+10; i < n; i++ { - tb, err := generateRandomBilling() + tb, err := generateRandomTransfer() if err != nil { return nil, err @@ -97,103 +123,6 @@ func generateRandomBlock(parent hash.Hash, isGenesis bool) (b *BPBlock, err erro return } -func generateRandomBillingRequestHeader() *BillingRequestHeader { - return &BillingRequestHeader{ - DatabaseID: generateRandomDatabaseID(), - LowBlock: generateRandomHash(), - LowHeight: rand.Int31(), - HighBlock: generateRandomHash(), - HighHeight: rand.Int31(), - GasAmounts: generateRandomGasAmount(peerNum), - } -} - -func generateRandomBillingRequest() (req *BillingRequest, err error) { - reqHeader := generateRandomBillingRequestHeader() - req = &BillingRequest{ - Header: *reqHeader, - } - if _, err = req.PackRequestHeader(); err != nil { - return nil, err - } - - for i := 0; i < peerNum; i++ { - // Generate key pair - var priv *asymmetric.PrivateKey - - if priv, _, err = asymmetric.GenSecp256k1KeyPair(); err != nil { - return - } - - if _, _, err = req.SignRequestHeader(priv, false); err != nil { - return - } - } - - return -} - -func generateRandomBillingHeader() (tc *BillingHeader, err error) { - var req *BillingRequest - if req, err = generateRandomBillingRequest(); err != nil { - return - } - - var priv *asymmetric.PrivateKey - if priv, _, err = asymmetric.GenSecp256k1KeyPair(); err != nil { - return - } - - if _, _, err = req.SignRequestHeader(priv, false); err != nil { - return - } - - receivers := make([]*proto.AccountAddress, peerNum) - fees := make([]uint64, peerNum) - rewards := make([]uint64, peerNum) - for i := range fees { - h := generateRandomHash() - accountAddress := proto.AccountAddress(h) - receivers[i] = &accountAddress - fees[i] = rand.Uint64() - rewards[i] = rand.Uint64() - } - - producer := proto.AccountAddress(generateRandomHash()) - tc = NewBillingHeader(pi.AccountNonce(rand.Uint32()), req, producer, receivers, fees, rewards) - return tc, nil -} - -func generateRandomBilling() (*Billing, error) { - header, err := generateRandomBillingHeader() - if err != nil { - return nil, err - } - priv, _, err := asymmetric.GenSecp256k1KeyPair() - if err != nil { - return nil, err - } - txBilling := NewBilling(header) - if err := txBilling.Sign(priv); err != nil { - return nil, err - } - return txBilling, nil -} - -func generateRandomGasAmount(n int) []*proto.AddrAndGas { - gasAmount := make([]*proto.AddrAndGas, n) - - for i := range gasAmount { - gasAmount[i] = &proto.AddrAndGas{ - AccountAddress: proto.AccountAddress(generateRandomHash()), - RawNodeID: proto.RawNodeID{Hash: generateRandomHash()}, - GasAmount: rand.Uint64(), - } - } - - return gasAmount -} - func randBytes(n int) (b []byte) { b = make([]byte, n) rand.Read(b) diff --git a/worker/db.go b/worker/db.go index 4e27c2148..6a04297bd 100644 --- a/worker/db.go +++ b/worker/db.go @@ -150,7 +150,7 @@ func NewDatabase(cfg *DBConfig, peers *proto.Peers, } // init chain - chainFile := filepath.Join(cfg.DataDir, SQLChainFileName) + chainFile := filepath.Join(cfg.RootDir, SQLChainFileName) if db.nodeID, err = kms.GetLocalNodeID(); err != nil { return } @@ -166,13 +166,12 @@ func NewDatabase(cfg *DBConfig, peers *proto.Peers, MuxService: cfg.ChainMux, Server: db.nodeID, - Period: conf.GConf.SQLChainPeriod, - Tick: conf.GConf.SQLChainTick, - QueryTTL: conf.GConf.SQLChainTTL, - - UpdatePeriod: cfg.UpdateBlockCount, - - IsolationLevel: cfg.IsolationLevel, + Period: conf.GConf.SQLChainPeriod, + Tick: conf.GConf.SQLChainTick, + QueryTTL: conf.GConf.SQLChainTTL, + LastBillingHeight: cfg.LastBillingHeight, + UpdatePeriod: cfg.UpdateBlockCount, + IsolationLevel: cfg.IsolationLevel, } if db.chain, err = sqlchain.NewChain(chainCfg); err != nil { return diff --git a/worker/db_config.go b/worker/db_config.go index 97270a627..ec6a2792a 100644 --- a/worker/db_config.go +++ b/worker/db_config.go @@ -26,6 +26,7 @@ import ( // DBConfig defines the database config. type DBConfig struct { DatabaseID proto.DatabaseID + RootDir string DataDir string KayakMux *DBKayakMuxService ChainMux *sqlchain.MuxService @@ -33,6 +34,7 @@ type DBConfig struct { EncryptionKey string SpaceLimit uint64 UpdateBlockCount uint64 + LastBillingHeight int32 UseEventualConsistency bool ConsistencyLevel float64 IsolationLevel int diff --git a/worker/db_test.go b/worker/db_test.go index 3debf8569..34d5ea7a3 100644 --- a/worker/db_test.go +++ b/worker/db_test.go @@ -70,7 +70,7 @@ func TestSingleDatabase(t *testing.T) { // create file cfg := &DBConfig{ - DatabaseID: "TEST", + DatabaseID: "00000bef611d346c0cbe1beaa76e7f0ed705a194fdf9ac3a248ec70e9c198bf9", DataDir: rootDir, KayakMux: kayakMuxService, ChainMux: chainMuxService, @@ -409,7 +409,7 @@ func TestInitFailed(t *testing.T) { // create file cfg := &DBConfig{ - DatabaseID: "TEST", + DatabaseID: "00000bef611d346c0cbe1beaa76e7f0ed705a194fdf9ac3a248ec70e9c198bf9", DataDir: rootDir, KayakMux: kayakMuxService, ChainMux: chainMuxService, @@ -464,7 +464,7 @@ func TestDatabaseRecycle(t *testing.T) { // create file cfg := &DBConfig{ - DatabaseID: "TEST", + DatabaseID: "00000bef611d346c0cbe1beaa76e7f0ed705a194fdf9ac3a248ec70e9c198bf9", DataDir: rootDir, KayakMux: kayakMuxService, ChainMux: chainMuxService, diff --git a/worker/dbms.go b/worker/dbms.go index fa65144d2..4093cdb0c 100644 --- a/worker/dbms.go +++ b/worker/dbms.go @@ -182,11 +182,44 @@ func (dbms *DBMS) Init() (err error) { err = errors.Wrap(err, "init chain bus failed") return } + if err = dbms.busService.Subscribe("/UpdateBilling/", dbms.updateBilling); err != nil { + err = errors.Wrap(err, "init chain bus failed") + return + } dbms.busService.Start() return } +func (dbms *DBMS) updateBilling(itx interfaces.Transaction, count uint32) { + var ( + tx *types.UpdateBilling + ok bool + ) + if tx, ok = itx.(*types.UpdateBilling); !ok { + log.WithFields(log.Fields{ + "type": itx.GetTransactionType(), + }).WithError(ErrInvalidTransactionType).Warn("invalid tx type in update billing") + return + } + // Get profile and database instance + var ( + id = tx.Receiver.DatabaseID() + profile *types.SQLChainProfile + database *Database + ) + le := log.WithFields(log.Fields{ + "id": id, + }) + if database, ok = dbms.getMeta(id); !ok { + le.Warn("cannot find database") + } + if profile, ok = dbms.busService.RequestSQLProfile(id); !ok { + le.Warn("cannot find profile") + } + database.chain.SetLastBillingHeight(int32(profile.LastUpdatedHeight)) +} + func (dbms *DBMS) createDatabase(tx interfaces.Transaction, count uint32) { cd, ok := tx.(*types.CreateDatabase) if !ok { @@ -388,6 +421,7 @@ func (dbms *DBMS) Create(instance *types.ServiceInstance, cleanup bool) (err err // new db dbCfg := &DBConfig{ DatabaseID: instance.DatabaseID, + RootDir: dbms.cfg.RootDir, DataDir: rootDir, KayakMux: dbms.kayakMux, ChainMux: dbms.chainMux, @@ -401,6 +435,11 @@ func (dbms *DBMS) Create(instance *types.ServiceInstance, cleanup bool) (err err SlowQueryTime: DefaultSlowQueryTime, } + // set last billing height + if profile, ok := dbms.busService.RequestSQLProfile(dbCfg.DatabaseID); ok { + dbCfg.LastBillingHeight = int32(profile.LastUpdatedHeight) + } + if db, err = NewDatabase(dbCfg, instance.Peers, instance.GenesisBlock); err != nil { return } diff --git a/worker/helper_test.go b/worker/helper_test.go index 9810f098e..195e35d5f 100644 --- a/worker/helper_test.go +++ b/worker/helper_test.go @@ -89,9 +89,9 @@ var ( }, }, Transactions: []interfaces.Transaction{ - &types.Transfer{}, - &types.Transfer{}, - &types.Transfer{}, + types.NewTransfer(&types.TransferHeader{}), + types.NewTransfer(&types.TransferHeader{}), + types.NewTransfer(&types.TransferHeader{}), }, } testOddBlocks = types.BPBlock{ @@ -101,7 +101,7 @@ var ( }, }, Transactions: []interfaces.Transaction{ - &types.Transfer{}, + types.NewTransfer(&types.TransferHeader{}), }, } testID = proto.DatabaseID("111") diff --git a/xenomint/mux_test.go b/xenomint/mux_test.go index 358ebf1a9..a5fb174cc 100644 --- a/xenomint/mux_test.go +++ b/xenomint/mux_test.go @@ -101,8 +101,12 @@ func setupMuxParallel(priv *ca.PrivateKey) ( } kms.SetLocalNodeIDNonce(nis[2].ID.ToRawNodeID().CloneBytes(), &nis[2].Nonce) for i := range nis { - route.SetNodeAddrCache(nis[i].ID.ToRawNodeID(), nis[i].Addr) - kms.SetNode(&nis[i]) + if err = route.SetNodeAddrCache(nis[i].ID.ToRawNodeID(), nis[i].Addr); err != nil { + return + } + if err = kms.SetNode(&nis[i]); err != nil { + return + } } // Register mux service if ms, err = NewMuxService(benchmarkRPCName, mnSv); err != nil { @@ -277,7 +281,7 @@ func TestMuxService(t *testing.T) { ms.register(benchmarkDatabaseID, c) defer func() { ms.unregister(benchmarkDatabaseID) - teardownChain(t.Name(), c) + _ = teardownChain(t.Name(), c) }() // Setup query requests diff --git a/xenomint/pool.go b/xenomint/pool.go index 38a1f6511..33494e183 100644 --- a/xenomint/pool.go +++ b/xenomint/pool.go @@ -50,6 +50,7 @@ type pool struct { // Failed queries: hash => Request failed map[hash.Hash]*types.Request // Succeeded queries and their index + reads map[hash.Hash]*QueryTracker queries []*QueryTracker index map[uint64]int // Atomic counters for stats @@ -60,6 +61,7 @@ type pool struct { func newPool() *pool { return &pool{ failed: make(map[hash.Hash]*types.Request), + reads: make(map[hash.Hash]*QueryTracker), queries: make([]*QueryTracker, 0), index: make(map[uint64]int), } @@ -73,6 +75,11 @@ func (p *pool) enqueue(sp uint64, q *QueryTracker) { return } +func (p *pool) enqueueRead(q *QueryTracker) { + // NOTE(leventeliu): this overwrites any request with a same hash + p.reads[q.Req.Header.Hash()] = q +} + func (p *pool) setFailed(req *types.Request) { p.failed[req.Header.Hash()] = req atomic.StoreInt32(&p.failedRequestCount, int32(len(p.failed))) diff --git a/xenomint/state.go b/xenomint/state.go index 7bcbf5557..0de4731e1 100644 --- a/xenomint/state.go +++ b/xenomint/state.go @@ -37,23 +37,18 @@ type sqlQuerier interface { } type sqlExecuter interface { - sqlQuerier Exec(query string, args ...interface{}) (sql.Result, error) ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error) - Commit() error - Rollback() error -} - -type sqlDB struct { - *sql.DB } -func (db *sqlDB) Commit() error { - return nil +type sqlTransaction interface { + Commit() error + Rollback() error } -func (db *sqlDB) Rollback() error { - return nil +type sqlHandler interface { + sqlQuerier + sqlExecuter } // State defines a xenomint state which is bound to a underlying storage. @@ -66,7 +61,7 @@ type State struct { closed bool nodeID proto.NodeID - executer sqlExecuter + handler sqlHandler maxTx uint64 lastCommitPoint uint64 current uint64 // current is the current lastSeq of the current transaction @@ -82,18 +77,18 @@ func NewState(level sql.IsolationLevel, nodeID proto.NodeID, strg xi.Storage) (s pool: newPool(), maxTx: 100, } - s.openSQLExecuter() + s.openHandler() return } -func (s *State) openSQLExecuter() { +func (s *State) openHandler() { if s.level == sql.LevelReadUncommitted { var err error - if s.executer, err = s.strg.Writer().Begin(); err != nil { + if s.handler, err = s.strg.Writer().Begin(); err != nil { log.WithError(err).Fatal("failed to open transaction") } } else { - s.executer = &sqlDB{DB: s.strg.Writer()} + s.handler = s.strg.Writer() } } @@ -128,11 +123,11 @@ func (s *State) Close(commit bool) (err error) { if s.closed { return } - if s.executer != nil { + if s.handler != nil { if commit { - s.commitSQLExecuter() + s.commitHandler() } else { - s.rollbackSQLExecuter() + s.rollbackHandler() } } if err = s.strg.Close(); err != nil { @@ -168,7 +163,9 @@ func readSingle( if rows, err = qer.QueryContext(ctx, pattern, args...); err != nil { return } - defer rows.Close() + defer func() { + _ = rows.Close() + }() // Fetch column names and types if names, err = rows.Columns(); err != nil { return @@ -226,6 +223,9 @@ func (s *State) readWithContext( } // Build query response ref = &QueryTracker{Req: req} + s.Lock() + s.pool.enqueueRead(ref) + s.Unlock() resp = &types.Response{ Header: types.SignedResponseHeader{ ResponseHeader: types.ResponseHeader{ @@ -260,7 +260,7 @@ func (s *State) readTx( // lock transaction s.Lock() defer s.Unlock() - querier = s.executer + querier = s.handler } else { var tx *sql.Tx if tx, ierr = s.reader().Begin(); ierr != nil { @@ -268,7 +268,9 @@ func (s *State) readTx( return } querier = tx - defer tx.Rollback() + defer func() { + _ = tx.Rollback() + }() } defer func() { @@ -291,6 +293,9 @@ func (s *State) readTx( } // Build query response ref = &QueryTracker{Req: req} + s.Lock() + s.pool.enqueueRead(ref) + s.Unlock() resp = &types.Response{ Header: types.SignedResponseHeader{ ResponseHeader: types.ResponseHeader{ @@ -338,7 +343,7 @@ func (s *State) writeSingle( return } //parsed = time.Since(start) - if res, err = s.executer.Exec(pattern, args...); err == nil { + if res, err = s.handler.Exec(pattern, args...); err == nil { if containsDDL { atomic.StoreUint32(&s.hasSchemaChange, 1) } @@ -398,16 +403,20 @@ func (s *State) write( lastSeq = s.getSeq() if qcnt > 1 && s.level == sql.LevelReadUncommitted { // Set savepoint - if _, ierr = s.executer.Exec(`SAVEPOINT "?"`, lastSeq); ierr != nil { + if _, ierr = s.handler.Exec(`SAVEPOINT "?"`, lastSeq); ierr != nil { err = errors.Wrapf(ierr, "failed to create savepoint %d", lastSeq) return } - defer s.executer.Exec(`ROLLBACK TO "?"`, lastSeq) + defer func() { + _, _ = s.handler.Exec(`ROLLBACK TO "?"`, lastSeq) + }() } if s.level != sql.LevelReadUncommitted { // NOTE(leventeliu): this will cancel any uncommitted transaction, and do not harm to // committed ones. - defer s.executer.Exec(`ROLLBACK`) + defer func() { + _, _ = s.handler.Exec(`ROLLBACK`) + }() } for i, v := range req.Payload.Queries { var res sql.Result @@ -426,7 +435,7 @@ func (s *State) write( if s.level == sql.LevelReadUncommitted { if qcnt > 1 { // Release savepoint - if _, ierr = s.executer.Exec(`RELEASE SAVEPOINT "?"`, lastSeq); ierr != nil { + if _, ierr = s.handler.Exec(`RELEASE SAVEPOINT "?"`, lastSeq); ierr != nil { err = errors.Wrapf(ierr, "failed to release savepoint %d", lastSeq) return } @@ -435,7 +444,7 @@ func (s *State) write( // Try to commit if the ongoing tx is too large or schema is changed if s.getSeq()-s.getLastCommitPoint() > s.maxTx || atomic.LoadUint32(&s.hasSchemaChange) != 0 { - s.flushSQLExecuter() + s.flushHandler() } writeDone = time.Since(start) if isLeader { @@ -491,7 +500,7 @@ func (s *State) replay(ctx context.Context, req *types.Request, resp *types.Resp // Try to commit if the ongoing tx is too large or schema is changed if s.getSeq()-s.getLastCommitPoint() > s.maxTx || atomic.LoadUint32(&s.hasSchemaChange) != 0 { - s.flushSQLExecuter() + s.flushHandler() } s.pool.enqueue(lastSeq, query) return @@ -541,7 +550,7 @@ func (s *State) ReplayBlockWithContext(ctx context.Context, block *types.Block) s.pool.enqueue(lastsp, query) } // Always try to commit after a block is successfully replayed - s.flushSQLExecuter() + s.flushHandler() // Remove duplicate failed queries from local pool for _, r := range block.FailedReqs { s.pool.removeFailed(r) @@ -579,7 +588,7 @@ func (s *State) commit() (err error) { lockReleased = time.Since(start) }() lockAcquired = time.Since(start) - s.flushSQLExecuter() + s.flushHandler() committed = time.Since(start) _ = s.pool.queries s.pool = newPool() @@ -625,33 +634,40 @@ func (s *State) CommitExWithContext( lockReleased = time.Since(start) }() // Always try to commit before the block is produced - s.flushSQLExecuter() + s.flushHandler() committed = time.Since(start) // Return pooled items and reset failed = s.pool.failedList() queries = s.pool.queries + for _, v := range s.pool.reads { + queries = append(queries, v) + } s.pool = newPool() poolCleaned = time.Since(start) return } -func (s *State) flushSQLExecuter() { - s.commitSQLExecuter() - s.openSQLExecuter() +func (s *State) flushHandler() { + s.commitHandler() + s.openHandler() } -func (s *State) commitSQLExecuter() { - if err := s.executer.Commit(); err != nil { - log.WithError(err).Fatal("failed to commit") +func (s *State) commitHandler() { + if tx, ok := s.handler.(sqlTransaction); ok { + if err := tx.Commit(); err != nil { + log.WithError(err).Fatal("failed to commit") + } } // reset schema change flag atomic.StoreUint32(&s.hasSchemaChange, 0) atomic.StoreUint64(&s.lastCommitPoint, s.getSeq()) } -func (s *State) rollbackSQLExecuter() { - if err := s.executer.Rollback(); err != nil { - log.WithError(err).Fatal("failed to rollback") +func (s *State) rollbackHandler() { + if tx, ok := s.handler.(sqlTransaction); ok { + if err := tx.Rollback(); err != nil { + log.WithError(err).Fatal("failed to rollback") + } } // reset schema change flag atomic.StoreUint32(&s.hasSchemaChange, 0) diff --git a/xenomint/state_test.go b/xenomint/state_test.go index 9811f38be..de962e931 100644 --- a/xenomint/state_test.go +++ b/xenomint/state_test.go @@ -741,10 +741,10 @@ func TestSerializableState(t *testing.T) { for { _, resp, err = state.Query(iReq, true) c.So(err, ShouldBeNil) - c.Printf("insert affected rows: %d\n", resp.Header.AffectedRows) + _, _ = c.Printf("insert affected rows: %d\n", resp.Header.AffectedRows) _, resp, err = state.Query(dReq, true) c.So(err, ShouldBeNil) - c.Printf("delete affected rows: %d\n", resp.Header.AffectedRows) + _, _ = c.Printf("delete affected rows: %d\n", resp.Header.AffectedRows) select { case <-ctx.Done(): return @@ -766,7 +766,7 @@ func TestSerializableState(t *testing.T) { DeclTypes: []string{""}, Rows: []types.ResponseRow{{Values: []interface{}{int64(count)}}}, }), ShouldBeTrue) - Printf("index = %d, count = %v\n", i, resp) + _, _ = Printf("index = %d, count = %v\n", i, resp) } }) Convey("The state should not see uncommitted changes", func(c C) { diff --git a/xenomint/xxx_test.go b/xenomint/xxx_test.go index b0a7686d0..07d053f2f 100644 --- a/xenomint/xxx_test.go +++ b/xenomint/xxx_test.go @@ -136,7 +136,7 @@ func createNodesWithPublicKey( wg.Add(1) go func() { defer wg.Done() - miner.ComputeBlockNonce(block, next, diff) + _ = miner.ComputeBlockNonce(block, next, diff) }() ni = <-nic nis[i] = proto.Node{