Implement blockdb levelDB backend (#6)

5 files changed, 305 insertions, 12 deletions

diff --git a/blockdb/interfaces.go b/blockdb/interfaces.go
index 9822dfc..c2cf8a6 100644
--- a/blockdb/interfaces.go
+++ b/blockdb/interfaces.go
@@ -37,7 +37,6 @@ var (
 type BlockDatabase interface {
 	Reader
 	Writer
-	Deleter
 }
 
 // Reader defines the interface for reading blocks into DB.
@@ -52,8 +51,3 @@ type Writer interface {
 	Update(block types.Block) error
 	Put(block types.Block) error
 }
-
-// Deleter defines the interface for deleting blocks in the DB.
-type Deleter interface {
-	Delete(hash common.Hash)
-}
diff --git a/blockdb/level-db.go b/blockdb/level-db.go
new file mode 100644
index 0000000..c6d0a7b
--- /dev/null
+++ b/blockdb/level-db.go
@@ -0,0 +1,213 @@
+// Copyright 2018 The dexon-consensus-core Authors
+// This file is part of the dexon-consensus-core library.
+//
+// The dexon-consensus-core library is free software: you can redistribute it
+// and/or modify it under the terms of the GNU Lesser General Public License as
+// published by the Free Software Foundation, either version 3 of the License,
+// or (at your option) any later version.
+//
+// The dexon-consensus-core library is distributed in the hope that it will be
+// useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+// General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the dexon-consensus-core library. If not, see
+// <http://www.gnu.org/licenses/>.
+
+package blockdb
+
+import (
+	"encoding/json"
+	"sync"
+
+	"github.com/syndtr/goleveldb/leveldb"
+	"github.com/syndtr/goleveldb/leveldb/opt"
+
+	"github.com/dexon-foundation/dexon-consensus-core/common"
+	"github.com/dexon-foundation/dexon-consensus-core/core/types"
+)
+
+// LevelDBBackendBlockDB is a leveldb backend BlockDB implementation.
+type LevelDBBackendBlockDB struct {
+	db        *leveldb.DB
+	index     map[types.ValidatorID]map[uint64]common.Hash
+	indexLock sync.RWMutex
+}
+
+// NewLevelDBBackendBlockDB initialize a leveldb-backed block database.
+func NewLevelDBBackendBlockDB(
+	path string) (lvl *LevelDBBackendBlockDB, err error) {
+
+	db, err := leveldb.OpenFile(path, nil)
+	if err != nil {
+		return
+	}
+	lvl = &LevelDBBackendBlockDB{db: db}
+	err = lvl.syncIndex()
+	if err != nil {
+		return
+	}
+	return
+}
+
+// Close would release allocated resource.
+func (lvl *LevelDBBackendBlockDB) Close() error {
+	return lvl.db.Close()
+}
+
+// Has implements the Reader.Has method.
+func (lvl *LevelDBBackendBlockDB) Has(hash common.Hash) bool {
+	exists, err := lvl.db.Has([]byte(hash[:]), nil)
+	if err != nil {
+		// TODO(missionliao): Modify the interface to return error.
+		panic(err)
+	}
+	return exists
+}
+
+// Get implements the Reader.Get method.
+func (lvl *LevelDBBackendBlockDB) Get(
+	hash common.Hash) (block types.Block, err error) {
+
+	queried, err := lvl.db.Get([]byte(hash[:]), nil)
+	if err != nil {
+		if err == leveldb.ErrNotFound {
+			err = ErrBlockDoesNotExist
+		}
+		return
+	}
+	err = json.Unmarshal(queried, &block)
+	if err != nil {
+		return
+	}
+	return
+}
+
+// GetByValidatorAndHeight implements
+// the Reader.GetByValidatorAndHeight method.
+func (lvl *LevelDBBackendBlockDB) GetByValidatorAndHeight(
+	vID types.ValidatorID, height uint64) (block types.Block, err error) {
+
+	lvl.indexLock.RLock()
+	defer lvl.indexLock.RUnlock()
+
+	// Get block's hash from in-memory index.
+	vMap, exists := lvl.index[vID]
+	if !exists {
+		err = ErrBlockDoesNotExist
+		return
+	}
+	hash, exists := vMap[height]
+	if !exists {
+		err = ErrBlockDoesNotExist
+		return
+	}
+
+	// Get block from hash.
+	queried, err := lvl.db.Get([]byte(hash[:]), nil)
+	if err != nil {
+		if err == leveldb.ErrNotFound {
+			err = ErrBlockDoesNotExist
+		}
+		return
+	}
+
+	err = json.Unmarshal(queried, &block)
+	if err != nil {
+		return
+	}
+	return
+}
+
+// Update implements the Writer.Update method.
+func (lvl *LevelDBBackendBlockDB) Update(block types.Block) (err error) {
+	// NOTE: we didn't handle changes of block hash (and it
+	//       should not happen).
+	marshaled, err := json.Marshal(&block)
+	if err != nil {
+		return
+	}
+
+	if !lvl.Has(block.Hash) {
+		err = ErrBlockDoesNotExist
+		return
+	}
+	err = lvl.db.Put(
+		[]byte(block.Hash[:]),
+		marshaled,
+		&opt.WriteOptions{
+			Sync: true,
+		})
+	if err != nil {
+		return
+	}
+	return
+}
+
+// Put implements the Writer.Put method.
+func (lvl *LevelDBBackendBlockDB) Put(block types.Block) (err error) {
+	marshaled, err := json.Marshal(&block)
+	if err != nil {
+		return
+	}
+	if lvl.Has(block.Hash) {
+		err = ErrBlockExists
+		return
+	}
+	syncedOpt := &opt.WriteOptions{
+		// We should force to sync for each write, it's safer
+		// from crash.
+		Sync: true,
+	}
+	err = lvl.db.Put(
+		[]byte(block.Hash[:]),
+		marshaled,
+		syncedOpt)
+	if err != nil {
+		return
+	}
+
+	// Build in-memory index.
+	lvl.addIndex(&block)
+	return
+}
+
+func (lvl *LevelDBBackendBlockDB) syncIndex() (err error) {
+	// Reset index.
+	lvl.index = make(map[types.ValidatorID]map[uint64]common.Hash)
+
+	// Construct index from DB.
+	iter := lvl.db.NewIterator(nil, nil)
+	defer func() {
+		iter.Release()
+		if err == nil {
+			// Only return iterator's error when no error
+			// is presented so far.
+			err = iter.Error()
+		}
+	}()
+
+	// Build index from blocks in DB, it may take time.
+	var block types.Block
+	for iter.Next() {
+		err = json.Unmarshal(iter.Value(), &block)
+		if err != nil {
+			return
+		}
+		lvl.addIndex(&block)
+	}
+	return
+}
+
+func (lvl *LevelDBBackendBlockDB) addIndex(block *types.Block) {
+	lvl.indexLock.Lock()
+	defer lvl.indexLock.Unlock()
+
+	heightMap, exists := lvl.index[block.ProposerID]
+	if !exists {
+		heightMap = make(map[uint64]common.Hash)
+		lvl.index[block.ProposerID] = heightMap
+	}
+	heightMap[block.Height] = block.Hash
+}
diff --git a/blockdb/level-db_test.go b/blockdb/level-db_test.go
new file mode 100644
index 0000000..5937d80
--- /dev/null
+++ b/blockdb/level-db_test.go
@@ -0,0 +1,90 @@
+// Copyright 2018 The dexon-consensus-core Authors
+// This file is part of the dexon-consensus-core library.
+//
+// The dexon-consensus-core library is free software: you can redistribute it
+// and/or modify it under the terms of the GNU Lesser General Public License as
+// published by the Free Software Foundation, either version 3 of the License,
+// or (at your option) any later version.
+//
+// The dexon-consensus-core library is distributed in the hope that it will be
+// useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+// General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the dexon-consensus-core library. If not, see
+// <http://www.gnu.org/licenses/>.
+
+package blockdb
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/suite"
+
+	"github.com/dexon-foundation/dexon-consensus-core/common"
+	"github.com/dexon-foundation/dexon-consensus-core/core/types"
+)
+
+type LevelDBTestSuite struct {
+	suite.Suite
+}
+
+func (s *LevelDBTestSuite) TestBasicUsage() {
+	dbName := fmt.Sprintf("test-db-%v.db", time.Now().UTC())
+	db, err := NewLevelDBBackendBlockDB(dbName)
+	s.Require().Nil(err)
+	defer func() {
+		err = db.Close()
+		s.Nil(err)
+
+		// TODO(missionliao): remove test db.
+	}()
+
+	// Queried something from an empty database.
+	hash1 := common.NewRandomHash()
+	_, err = db.Get(hash1)
+	s.Equal(ErrBlockDoesNotExist, err)
+
+	// Update on an empty database should not success.
+	validator1 := types.ValidatorID{Hash: common.NewRandomHash()}
+	block1 := types.Block{
+		ProposerID: validator1,
+		Hash:       hash1,
+		Height:     1,
+		State:      types.BlockStatusInit,
+	}
+	err = db.Update(block1)
+	s.Equal(ErrBlockDoesNotExist, err)
+
+	// Put to create a new record should just work fine.
+	err = db.Put(block1)
+	s.Nil(err)
+
+	// Get it back should work fine.
+	queried, err := db.Get(block1.Hash)
+	s.Nil(err)
+	s.Equal(queried.ProposerID, block1.ProposerID)
+	s.Equal(queried.State, block1.State)
+
+	// Test Update.
+	now := time.Now().UTC()
+	queried.Timestamps = map[types.ValidatorID]time.Time{
+		queried.ProposerID: now,
+	}
+
+	err = db.Update(queried)
+	s.Nil(err)
+
+	// Try to get it back via ValidatorID and height.
+	queried, err = db.GetByValidatorAndHeight(block1.ProposerID, block1.Height)
+
+	s.Nil(err)
+	s.Equal(now, queried.Timestamps[queried.ProposerID])
+}
+
+func TestLevelDB(t *testing.T) {
+	suite.Run(t, new(LevelDBTestSuite))
+}
diff --git a/blockdb/memory.go b/blockdb/memory.go
index fd62b71..5f95d1a 100644
--- a/blockdb/memory.go
+++ b/blockdb/memory.go
@@ -22,7 +22,7 @@ import (
 	"github.com/dexon-foundation/dexon-consensus-core/core/types"
 )
 
-// MemBackendBlockDB is a memory bakcend BlockDB implementation.
+// MemBackendBlockDB is a memory backend BlockDB implementation.
 type MemBackendBlockDB struct {
 	blocksByHash      map[common.Hash]*types.Block
 	blocksByValidator map[types.ValidatorID]map[uint64]*types.Block
@@ -78,8 +78,3 @@ func (m *MemBackendBlockDB) Update(block types.Block) error {
 	m.blocksByHash[block.Hash] = &block
 	return nil
 }
-
-// Delete deletes a block in the database.
-func (m *MemBackendBlockDB) Delete(hash common.Hash) {
-	delete(m.blocksByHash, hash)
-}
diff --git a/core/blocklattice.go b/core/blocklattice.go
index f746c20..9b43b67 100644
--- a/core/blocklattice.go
+++ b/core/blocklattice.go
@@ -97,6 +97,7 @@ func (l *BlockLattice) AddValidator(
 	l.fmax = (len(l.ValidatorSet) - 1) / 3
 	l.phi = 2*l.fmax + 1
 
+	// TODO: We should not make genesis blocks 'final' directly.
 	genesis.State = types.BlockStatusFinal
 	l.blockDB.Put(*genesis)
 }