package iavl import ( "fmt" "io" "sync" "github.com/pkg/errors" "github.com/tendermint/iavl" abci "github.com/tendermint/tendermint/abci/types" "github.com/tendermint/tendermint/crypto/merkle" tmkv "github.com/tendermint/tendermint/libs/kv" dbm "github.com/tendermint/tm-db" "github.com/cosmos/cosmos-sdk/store/cachekv" "github.com/cosmos/cosmos-sdk/store/tracekv" "github.com/cosmos/cosmos-sdk/store/types" sdk "github.com/cosmos/cosmos-sdk/types" sdkerrors "github.com/cosmos/cosmos-sdk/types/errors" ) const ( defaultIAVLCacheSize = 10000 ) var ( _ types.KVStore = (*Store)(nil) _ types.CommitStore = (*Store)(nil) _ types.CommitKVStore = (*Store)(nil) _ types.Queryable = (*Store)(nil) ) // Store Implements types.KVStore and CommitKVStore. type Store struct { tree Tree pruning types.PruningOptions } // LoadStore returns an IAVL Store as a CommitKVStore. Internally, it will load the // store's version (id) from the provided DB. An error is returned if the version // fails to load. func LoadStore(db dbm.DB, id types.CommitID, pruning types.PruningOptions, lazyLoading bool) (types.CommitKVStore, error) { if !pruning.IsValid() { return nil, fmt.Errorf("pruning options are invalid: %v", pruning) } var keepRecent int64 // Determine the value of keepRecent based on the following: // // If KeepEvery = 1, keepRecent should be 0 since there is no need to keep // latest version in a in-memory cache. // // If KeepEvery > 1, keepRecent should be 1 so that state changes in between // flushed states can be saved in the in-memory latest tree. if pruning.KeepEvery == 1 { keepRecent = 0 } else { keepRecent = 1 } tree, err := iavl.NewMutableTreeWithOpts( db, dbm.NewMemDB(), defaultIAVLCacheSize, iavl.PruningOptions(pruning.KeepEvery, keepRecent), ) if err != nil { return nil, err } if lazyLoading { _, err = tree.LazyLoadVersion(id.Version) } else { _, err = tree.LoadVersion(id.Version) } if err != nil { return nil, err } return &Store{ tree: tree, pruning: pruning, }, nil } // UnsafeNewStore returns a reference to a new IAVL Store with a given mutable // IAVL tree reference. It should only be used for testing purposes. // // CONTRACT: The IAVL tree should be fully loaded. // CONTRACT: PruningOptions passed in as argument must be the same as pruning options // passed into iavl.MutableTree func UnsafeNewStore(tree *iavl.MutableTree, po types.PruningOptions) *Store { return &Store{ tree: tree, pruning: po, } } // GetImmutable returns a reference to a new store backed by an immutable IAVL // tree at a specific version (height) without any pruning options. This should // be used for querying and iteration only. If the version does not exist or has // been pruned, an error will be returned. Any mutable operations executed will // result in a panic. func (st *Store) GetImmutable(version int64) (*Store, error) { if !st.VersionExists(version) { return nil, iavl.ErrVersionDoesNotExist } iTree, err := st.tree.GetImmutable(version) if err != nil { return nil, err } return &Store{ tree: &immutableTree{iTree}, pruning: st.pruning, }, nil } // Commit commits the current store state and returns a CommitID with the new // version and hash. func (st *Store) Commit() types.CommitID { hash, version, err := st.tree.SaveVersion() if err != nil { // TODO: Do we want to extend Commit to allow returning errors? panic(err) } // If the version we saved got flushed to disk, check if previous flushed // version should be deleted. if st.pruning.FlushVersion(version) { previous := version - st.pruning.KeepEvery // Previous flushed version should only be pruned if the previous version is // not a snapshot version OR if snapshotting is disabled (SnapshotEvery == 0). if previous != 0 && !st.pruning.SnapshotVersion(previous) { err := st.tree.DeleteVersion(previous) if errCause := errors.Cause(err); errCause != nil && errCause != iavl.ErrVersionDoesNotExist { panic(err) } } } return types.CommitID{ Version: version, Hash: hash, } } // Implements Committer. func (st *Store) LastCommitID() types.CommitID { return types.CommitID{ Version: st.tree.Version(), Hash: st.tree.Hash(), } } // SetPruning panics as pruning options should be provided at initialization // since IAVl accepts pruning options directly. func (st *Store) SetPruning(_ types.PruningOptions) { panic("cannot set pruning options on an initialized IAVL store") } // VersionExists returns whether or not a given version is stored. func (st *Store) VersionExists(version int64) bool { return st.tree.VersionExists(version) } // Implements Store. func (st *Store) GetStoreType() types.StoreType { return types.StoreTypeIAVL } // Implements Store. func (st *Store) CacheWrap() types.CacheWrap { return cachekv.NewStore(st) } // CacheWrapWithTrace implements the Store interface. func (st *Store) CacheWrapWithTrace(w io.Writer, tc types.TraceContext) types.CacheWrap { return cachekv.NewStore(tracekv.NewStore(st, w, tc)) } // Implements types.KVStore. func (st *Store) Set(key, value []byte) { types.AssertValidValue(value) st.tree.Set(key, value) } // Implements types.KVStore. func (st *Store) Get(key []byte) []byte { _, value := st.tree.Get(key) return value } // Implements types.KVStore. func (st *Store) Has(key []byte) (exists bool) { return st.tree.Has(key) } // Implements types.KVStore. func (st *Store) Delete(key []byte) { st.tree.Remove(key) } // Implements types.KVStore. func (st *Store) Iterator(start, end []byte) types.Iterator { var iTree *iavl.ImmutableTree switch tree := st.tree.(type) { case *immutableTree: iTree = tree.ImmutableTree case *iavl.MutableTree: iTree = tree.ImmutableTree } return newIAVLIterator(iTree, start, end, true) } // Implements types.KVStore. func (st *Store) ReverseIterator(start, end []byte) types.Iterator { var iTree *iavl.ImmutableTree switch tree := st.tree.(type) { case *immutableTree: iTree = tree.ImmutableTree case *iavl.MutableTree: iTree = tree.ImmutableTree } return newIAVLIterator(iTree, start, end, false) } // Handle gatest the latest height, if height is 0 func getHeight(tree Tree, req abci.RequestQuery) int64 { height := req.Height if height == 0 { latest := tree.Version() if tree.VersionExists(latest - 1) { height = latest - 1 } else { height = latest } } return height } // Query implements ABCI interface, allows queries // // by default we will return from (latest height -1), // as we will have merkle proofs immediately (header height = data height + 1) // If latest-1 is not present, use latest (which must be present) // if you care to have the latest data to see a tx results, you must // explicitly set the height you want to see func (st *Store) Query(req abci.RequestQuery) (res abci.ResponseQuery) { if len(req.Data) == 0 { return sdkerrors.QueryResult(sdkerrors.Wrap(sdkerrors.ErrTxDecode, "query cannot be zero length")) } tree := st.tree // store the height we chose in the response, with 0 being changed to the // latest height res.Height = getHeight(tree, req) switch req.Path { case "/key": // get by key key := req.Data // data holds the key bytes res.Key = key if !st.VersionExists(res.Height) { res.Log = iavl.ErrVersionDoesNotExist.Error() break } if req.Prove { value, proof, err := tree.GetVersionedWithProof(key, res.Height) if err != nil { res.Log = err.Error() break } if proof == nil { // Proof == nil implies that the store is empty. if value != nil { panic("unexpected value for an empty proof") } } if value != nil { // value was found res.Value = value res.Proof = &merkle.Proof{Ops: []merkle.ProofOp{iavl.NewValueOp(key, proof).ProofOp()}} } else { // value wasn't found res.Value = nil res.Proof = &merkle.Proof{Ops: []merkle.ProofOp{iavl.NewAbsenceOp(key, proof).ProofOp()}} } } else { _, res.Value = tree.GetVersioned(key, res.Height) } case "/subspace": var KVs []types.KVPair subspace := req.Data res.Key = subspace iterator := types.KVStorePrefixIterator(st, subspace) for ; iterator.Valid(); iterator.Next() { KVs = append(KVs, types.KVPair{Key: iterator.Key(), Value: iterator.Value()}) } iterator.Close() res.Value = cdc.MustMarshalBinaryBare(KVs) default: return sdkerrors.QueryResult(sdkerrors.Wrapf(sdkerrors.ErrUnknownRequest, "unexpected query path: %v", req.Path)) } return res } //---------------------------------------- // Implements types.Iterator. type iavlIterator struct { // Domain start, end []byte key []byte // The current key (mutable) value []byte // The current value (mutable) // Underlying store tree *iavl.ImmutableTree // Channel to push iteration values. iterCh chan tmkv.Pair // Close this to release goroutine. quitCh chan struct{} // Close this to signal that state is initialized. initCh chan struct{} mtx sync.Mutex ascending bool // Iteration order invalid bool // True once, true forever (mutable) } var _ types.Iterator = (*iavlIterator)(nil) // newIAVLIterator will create a new iavlIterator. // CONTRACT: Caller must release the iavlIterator, as each one creates a new // goroutine. func newIAVLIterator(tree *iavl.ImmutableTree, start, end []byte, ascending bool) *iavlIterator { iter := &iavlIterator{ tree: tree, start: sdk.CopyBytes(start), end: sdk.CopyBytes(end), ascending: ascending, iterCh: make(chan tmkv.Pair), // Set capacity > 0? quitCh: make(chan struct{}), initCh: make(chan struct{}), } go iter.iterateRoutine() go iter.initRoutine() return iter } // Run this to funnel items from the tree to iterCh. func (iter *iavlIterator) iterateRoutine() { iter.tree.IterateRange( iter.start, iter.end, iter.ascending, func(key, value []byte) bool { select { case <-iter.quitCh: return true // done with iteration. case iter.iterCh <- tmkv.Pair{Key: key, Value: value}: return false // yay. } }, ) close(iter.iterCh) // done. } // Run this to fetch the first item. func (iter *iavlIterator) initRoutine() { iter.receiveNext() close(iter.initCh) } // Implements types.Iterator. func (iter *iavlIterator) Domain() (start, end []byte) { return iter.start, iter.end } // Implements types.Iterator. func (iter *iavlIterator) Valid() bool { iter.waitInit() iter.mtx.Lock() validity := !iter.invalid iter.mtx.Unlock() return validity } // Implements types.Iterator. func (iter *iavlIterator) Next() { iter.waitInit() iter.mtx.Lock() iter.assertIsValid(true) iter.receiveNext() iter.mtx.Unlock() } // Implements types.Iterator. func (iter *iavlIterator) Key() []byte { iter.waitInit() iter.mtx.Lock() iter.assertIsValid(true) key := iter.key iter.mtx.Unlock() return key } // Implements types.Iterator. func (iter *iavlIterator) Value() []byte { iter.waitInit() iter.mtx.Lock() iter.assertIsValid(true) val := iter.value iter.mtx.Unlock() return val } // Close closes the IAVL iterator by closing the quit channel and waiting for // the iterCh to finish/close. func (iter *iavlIterator) Close() { close(iter.quitCh) // wait iterCh to close for range iter.iterCh { } } // Error performs a no-op. func (iter *iavlIterator) Error() error { return nil } //---------------------------------------- func (iter *iavlIterator) setNext(key, value []byte) { iter.assertIsValid(false) iter.key = key iter.value = value } func (iter *iavlIterator) setInvalid() { iter.assertIsValid(false) iter.invalid = true } func (iter *iavlIterator) waitInit() { <-iter.initCh } func (iter *iavlIterator) receiveNext() { kvPair, ok := <-iter.iterCh if ok { iter.setNext(kvPair.Key, kvPair.Value) } else { iter.setInvalid() } } // assertIsValid panics if the iterator is invalid. If unlockMutex is true, // it also unlocks the mutex before panicing, to prevent deadlocks in code that // recovers from panics func (iter *iavlIterator) assertIsValid(unlockMutex bool) { if iter.invalid { if unlockMutex { iter.mtx.Unlock() } panic("invalid iterator") } }