Skip to content
Closed
Show file tree
Hide file tree
Changes from 8 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -221,6 +221,11 @@ test_sortition:
test_persistence:
go test ${VERBOSE_TEST} -p=1 ./persistence/...

.PHONY: test_persistence_state_hash
## Run all go unit tests in the Persistence module
test_persistence_state_hash:
go test -run StateHash ${VERBOSE_TEST} -p=1 ./persistence/...

.PHONY: benchmark_sortition
## Benchmark the Sortition library
benchmark_sortition:
Expand Down
2 changes: 1 addition & 1 deletion consensus/module.go
Original file line number Diff line number Diff line change
Expand Up @@ -243,7 +243,7 @@ func (m *consensusModule) handleHotstuffMessage(msg *typesCons.HotstuffMessage)
}

func (m *consensusModule) AppHash() string {
return m.appHash
return m.appHash // TODO: This is a problem

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This isn't very helpful

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed. I should have made the comment: "reminder in this commit: do integration with consensus module."

I was hoping for more preliminary feedback on the business logic in updateStateHash.

Going to leave a comment in the main PR with more pointers.

}

func (m *consensusModule) CurrentHeight() uint64 {
Expand Down
3 changes: 2 additions & 1 deletion go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ require (
)

require (
github.com/celestiaorg/smt v0.2.1-0.20220414134126-dba215ccb884
github.com/dgraph-io/badger/v3 v3.2103.2
github.com/iancoleman/strcase v0.2.0
)
Expand All @@ -41,7 +42,7 @@ require (
github.com/gogo/protobuf v1.3.2 // indirect
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b // indirect
github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6 // indirect
github.com/golang/protobuf v1.5.2 // indirect
github.com/golang/protobuf v1.5.2
github.com/golang/snappy v0.0.3 // indirect
github.com/google/flatbuffers v1.12.1 // indirect
github.com/gotestyourself/gotestyourself v2.2.0+incompatible // indirect
Expand Down
2 changes: 2 additions & 0 deletions go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@ github.com/ProtonMail/go-ecvrf v0.0.1/go.mod h1:fhZbiRYn62/JGnBG2NGwCx0oT+gr/+I5
github.com/ajstarks/svgo v0.0.0-20180226025133-644b8db467af/go.mod h1:K08gAheRH3/J6wwsYMMT4xOr94bZjxIelGM0+d/wbFw=
github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
github.com/boombuler/barcode v1.0.0/go.mod h1:paBWMcWSl3LHKBqUq+rly7CNSldXjb2rDl3JlRe0mD8=
github.com/celestiaorg/smt v0.2.1-0.20220414134126-dba215ccb884 h1:iRNKw2WmAbVgGMNYzDH5Y2yY3+jyxwEK9Hc5pwIjZAE=
github.com/celestiaorg/smt v0.2.1-0.20220414134126-dba215ccb884/go.mod h1:/sdYDakowo/XaxS2Fl7CBqtuf/O2uTqF2zmAUFAtAiw=
github.com/cenkalti/backoff v2.2.1+incompatible h1:tNowT99t7UNflLxfYYSlKYsBpXdEet03Pg2g16Swow4=
github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
Expand Down
51 changes: 51 additions & 0 deletions persistence/application.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,16 +4,67 @@ import (
"encoding/hex"
"log"

"github.com/golang/protobuf/proto"
"github.com/pokt-network/pocket/persistence/schema"
"github.com/pokt-network/pocket/shared/types"

typesGenesis "github.com/pokt-network/pocket/shared/types/genesis"
)

func (p PostgresContext) GetAppExists(address []byte, height int64) (exists bool, err error) {
return p.GetExists(schema.ApplicationActor, address, height)
}

func (p PostgresContext) UpdateAppTree(apps [][]byte) error {
for _, app := range apps {
appProto := typesGenesis.App{}
if err := proto.Unmarshal(app, &appProto); err != nil {
return err
}
if _, err := p.MerkleTrees[AppMerkleTree].Update(appProto.Address, app); err != nil {
return err
}
}
return nil
}

func (p PostgresContext) getAppsUpdated(height int64) (apps []*typesGenesis.App, err error) {
actors, err := p.GetActorsUpdated(schema.ApplicationActor, height)
if err != nil {
return nil, err
}

for _, actor := range actors {
// DISCUSS_IN_THIS_COMMIT: This breaks the pattern of protos in persistence.
// - Is it okay?
// - Do we embed this logic in `UpdateAppTree`
app := &typesGenesis.App{
Address: []byte(actor.Address),
PublicKey: []byte(actor.PublicKey),
// Paused: actor.Paused, // DISCUSS_IN_THIS_COMMIT: Is this just a check for pause height = -1?
// Status: actor.Status, // TODO_IN_THIS_COMMIT: Use logic from `GetActorStatus` without an extra query
Chains: actor.Chains,
MaxRelays: actor.ActorSpecificParam,
StakedTokens: actor.StakedTokens,
PausedHeight: actor.PausedHeight,
UnstakingHeight: actor.UnstakingHeight,
Output: []byte(actor.OutputAddress),
}
// appBytes, err := proto.Marshal(&app)
// if err != nil {
// return nil, err
// }
// apps = append(apps, appBytes)
apps = append(apps, app)
}
return
}

func (p PostgresContext) GetApp(address []byte, height int64) (operator, publicKey, stakedTokens, maxRelays, outputAddress string, pauseHeight, unstakingHeight int64, chains []string, err error) {
actor, err := p.GetActor(schema.ApplicationActor, address, height)
if err != nil {
return
}
operator = actor.Address
publicKey = actor.PublicKey
stakedTokens = actor.StakedTokens
Expand Down
11 changes: 10 additions & 1 deletion persistence/block.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,19 @@ import (
"encoding/binary"
"encoding/hex"
"log"
"os"

"github.com/pokt-network/pocket/persistence/schema"
)

// TODO_IN_THIS_COMMIT: Out of scope so we might just need to do this as part of state sync
func (p *persistenceModule) shouldLoadBlockStore() bool {
if _, err := os.Stat(p.GetBus().GetConfig().Persistence.BlockStorePath); err == nil {
return true
}
return false
}

// OPTIMIZE(team): get from blockstore or keep in cache/memory
func (p PostgresContext) GetLatestBlockHeight() (latestHeight int64, err error) {
ctx, conn, err := p.GetCtxAndConnection()
Expand Down Expand Up @@ -53,7 +62,7 @@ func (p PostgresContext) StoreBlock(blockProtoBytes []byte) error {
// INVESTIGATE: Note that we are writing this directly to the blockStore. Depending on how
// the use of the PostgresContext evolves, we may need to write this to `ContextStore` and copy
// over to `BlockStore` when the block is committed.
return p.BlockStore.Put(heightToBytes(p.Height), blockProtoBytes)
return p.BlockStore.Set(heightToBytes(p.Height), blockProtoBytes)
}

func (p PostgresContext) InsertBlock(height uint64, hash string, proposerAddr []byte, quorumCert []byte) error {
Expand Down
17 changes: 14 additions & 3 deletions persistence/context.go
Original file line number Diff line number Diff line change
Expand Up @@ -14,18 +14,29 @@ func (p PostgresContext) RollbackToSavePoint(bytes []byte) error {
return nil
}

func (p PostgresContext) UpdateAppHash() ([]byte, error) {
if _, err := p.updateStateHash(); err != nil {
return nil, err
}
return p.StateHash, nil
}

func (p PostgresContext) AppHash() ([]byte, error) {
log.Println("TODO: AppHash not implemented")
return []byte("A real app hash, I am not"), nil
// log.Println("TODO: AppHash not implemented")
// return []byte("A real app hash, I am not"), n
return p.StateHash, nil
}

func (p PostgresContext) Reset() error {
panic("TODO: PostgresContext Reset not implemented")
}

func (p PostgresContext) Commit() error {
// HACK: The data has already been written to the postgres DB, so what should we do here? The idea I have is:
log.Println("TODO: Postgres context commit is currently a NOOP")
// HACK: The data has already been written to the postgres DB, so what should we do here? The idea I have is:
// if _, err := p.updateStateHash(); err != nil {
// return err
// }
return nil
}

Expand Down
7 changes: 5 additions & 2 deletions persistence/db.go
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import (
"math/rand"
"time"

"github.com/celestiaorg/smt"
"github.com/jackc/pgx/v4"
"github.com/pokt-network/pocket/persistence/kvstore"
"github.com/pokt-network/pocket/persistence/schema"
Expand All @@ -32,11 +33,13 @@ var _ modules.PersistenceContext = &PostgresContext{}
type PostgresContext struct {
Height int64
ContextStore kvstore.KVStore
StateHash []byte

// IMPROVE: Depending on how the use of `PostgresContext` evolves, we may be able to get
// access to these directly via the postgres module.
PostgresDB *pgx.Conn
BlockStore kvstore.KVStore
PostgresDB *pgx.Conn
BlockStore kvstore.KVStore // REARCHITECT_IN_THIS_COMMIT: This is a passthrough from the persistence module (i.e. not context)
MerkleTrees map[MerkleTree]*smt.SparseMerkleTree // REARCHITECT_IN_THIS_COMMIT: This is a passthrough from the persistence module (i.e. not context)
}

func (pg *PostgresContext) GetCtxAndConnection() (context.Context, *pgx.Conn, error) {
Expand Down
29 changes: 24 additions & 5 deletions persistence/kvstore/kvstore.go
Original file line number Diff line number Diff line change
@@ -1,29 +1,43 @@
package kvstore

import (
"errors"
"log"

"github.com/celestiaorg/smt"
badger "github.com/dgraph-io/badger/v3"
)

// TODO_IN_THIS_COMMIT: We might be able to remove the `KVStore` interface altogether if we end up using smt.MapStore
type KVStore interface {
// Lifecycle methods
Stop() error

// Accessors
Put(key []byte, value []byte) error
Get(key []byte) ([]byte, error)
Exists(key []byte) (bool, error)
ClearAll() error

// Same interface as in `smt.MapStore``
Set(key []byte, value []byte) error
Get(key []byte) ([]byte, error)
Delete(key []byte) error
}

var _ KVStore = &badgerKVStore{}
var _ smt.MapStore = &badgerKVStore{}

var (
ErrKVStoreExists = errors.New("kvstore already exists")
ErrKVStoreNotExists = errors.New("kvstore does not exist")
)

type badgerKVStore struct {
db *badger.DB
}

func NewKVStore(path string) (KVStore, error) {
// REFACTOR: Loads or creates a badgerDb at `path`. This may potentially need to be refactored
// into `NewKVStore` and `LoadKVStore` depending on how state sync evolves by leveraging `os.Stat`
// on the file path.
func OpenKVStore(path string) (KVStore, error) {
db, err := badger.Open(badger.DefaultOptions(path))
if err != nil {
return nil, err
Expand All @@ -39,7 +53,7 @@ func NewMemKVStore() KVStore {
return badgerKVStore{db: db}
}

func (store badgerKVStore) Put(key []byte, value []byte) error {
func (store badgerKVStore) Set(key []byte, value []byte) error {
txn := store.db.NewTransaction(true)
defer txn.Discard()

Expand Down Expand Up @@ -76,6 +90,11 @@ func (store badgerKVStore) Get(key []byte) ([]byte, error) {
return value, nil
}

func (store badgerKVStore) Delete(key []byte) error {
log.Fatalf("badgerKVStore.Delete not implemented yet")
return nil
}

func (store badgerKVStore) Exists(key []byte) (bool, error) {
val, err := store.Get(key)
if err != nil {
Expand Down
30 changes: 20 additions & 10 deletions persistence/module.go
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package persistence
import (
"log"

"github.com/celestiaorg/smt"
"github.com/jackc/pgx/v4"
"github.com/pokt-network/pocket/persistence/kvstore"
"github.com/pokt-network/pocket/shared/config"
Expand Down Expand Up @@ -56,6 +57,8 @@ type persistenceModule struct {
blockStore kvstore.KVStore
// A mapping of context IDs to persistence contexts
contexts map[contextId]modules.PersistenceContext
// Merkle trees
trees map[MerkleTree]*smt.SparseMerkleTree
}

type contextId uint64
Expand All @@ -66,7 +69,7 @@ func Create(c *config.Config) (modules.PersistenceModule, error) {
return nil, err
}

blockStore, err := kvstore.NewKVStore(c.Persistence.BlockStorePath)
blockStore, err := kvstore.OpenKVStore(c.Persistence.BlockStorePath)
if err != nil {
return nil, err
}
Expand All @@ -77,26 +80,33 @@ func Create(c *config.Config) (modules.PersistenceModule, error) {
postgresConn: postgresDb,
blockStore: blockStore,
contexts: make(map[contextId]modules.PersistenceContext),
trees: make(map[MerkleTree]*smt.SparseMerkleTree),
}, nil
}

func (p *persistenceModule) Start() error {
log.Println("Starting persistence module...")

shouldHydrateGenesis := false
shouldHydrateGenesis, err := p.shouldHydrateGenesisDb()
if err != nil {
return err
}

if shouldHydrateGenesis {
if shouldHydrateGenesis, err := p.shouldHydrateGenesisDb(); err != nil {
return nil
} else if shouldHydrateGenesis {
log.Println("Hydrating genesis state...")
if err := p.hydrateGenesisDbState(); err != nil {
return err
}
log.Println("Hydrating genesis state...")
} else {
log.Println("Loading state from previous state...")
log.Println("TODO: Finish loading previous state...")

}

// DISCUSS_IN_THIS_COMMIT: We've been using the module function pattern, but this making `initializeTrees`
// be able to create and/or load trees outside the scope of the persistence module makes it easier to test.
trees, err := newMerkleTrees()
if err != nil {
return err
}
// TODO_IN_THIS_COMMIT: load trees from state
p.trees = trees

return nil
}
Expand Down
10 changes: 10 additions & 0 deletions persistence/pre_persistence/app.go
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,16 @@ import (
"google.golang.org/protobuf/proto"
)

// func (m *PrePersistenceContext) GetAppsUpdated(height int64) ([][]byte, error) {
// // Not implemented
// return nil, nil
// }

func (m *PrePersistenceContext) UpdateAppTree([][]byte) error {
// Not implemented
return nil
}

func (m *PrePersistenceContext) GetAppExists(address []byte, height int64) (exists bool, err error) {
db := m.Store()
key := append(AppPrefixKey, address...)
Expand Down
5 changes: 5 additions & 0 deletions persistence/pre_persistence/persistence.go
Original file line number Diff line number Diff line change
Expand Up @@ -184,6 +184,11 @@ func (m *PrePersistenceContext) RollbackToSavePoint(bytes []byte) error {
return nil
}

func (m *PrePersistenceContext) UpdateAppHash() ([]byte, error) {
// log.Fatalf("PrePersistenceContext not implementing UpdateAppHash")
return nil, nil
}

// AppHash creates a unique hash based on the global state object
// NOTE: AppHash is an inefficient, arbitrary, mock implementation that enables the functionality
// TODO written for replacement, taking any and all better implementation suggestions - even if a temporary measure
Expand Down
4 changes: 4 additions & 0 deletions persistence/schema/base_actor.go
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,10 @@ func (actor *BaseProtocolActorSchema) GetChainsTableSchema() string {
return ProtocolActorChainsTableSchema(actor.chainsHeightConstraintName)
}

func (actor *BaseProtocolActorSchema) GetUpdatedAtHeightQuery(height int64) string {
return SelectAtHeight(AllColsSelector, height, actor.tableName)
}

func (actor *BaseProtocolActorSchema) GetQuery(address string, height int64) string {
return Select(AllColsSelector, address, height, actor.tableName)
}
Expand Down
2 changes: 2 additions & 0 deletions persistence/schema/protocol_actor.go
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,8 @@ type ProtocolActorSchema interface {

/*** Read/Get Queries ***/

// Returns a query to retrieve all of a Actors updated at that specific height.
GetUpdatedAtHeightQuery(height int64) string
// Returns a query to retrieve all of a single Actor's attributes.
GetQuery(address string, height int64) string
// Returns a query for the existence of an Actor given its address.
Expand Down
5 changes: 5 additions & 0 deletions persistence/schema/shared_sql.go
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,11 @@ func ProtocolActorChainsTableSchema(constraintName string) string {
)`, AddressCol, ChainIDCol, HeightCol, DefaultBigInt, constraintName, AddressCol, ChainIDCol, HeightCol)
}

func SelectAtHeight(selector string, height int64, tableName string) string {
return fmt.Sprintf(`SELECT %s FROM %s WHERE height=%d`,
selector, tableName, height)
}

func Select(selector, address string, height int64, tableName string) string {
return fmt.Sprintf(`SELECT %s FROM %s WHERE address='%s' AND height<=%d ORDER BY height DESC LIMIT 1`,
selector, tableName, address, height)
Expand Down
Loading