diff --git a/go.mod b/go.mod
index f6a1da972..ccf135fc8 100644
--- a/go.mod
+++ b/go.mod
@@ -6,6 +6,7 @@ require (
 	github.com/confio/ics23/go v0.6.6
 	github.com/gogo/gateway v1.1.0
 	github.com/gogo/protobuf v1.3.2
+	github.com/golang/mock v1.6.0 // indirect
 	github.com/golang/protobuf v1.5.2
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0
 	github.com/grpc-ecosystem/grpc-gateway v1.16.0
diff --git a/go.sum b/go.sum
index f79727677..75a8c6c87 100644
--- a/go.sum
+++ b/go.sum
@@ -310,6 +310,7 @@ github.com/golang/mock v1.4.1/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt
 github.com/golang/mock v1.4.3/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt3cw=
 github.com/golang/mock v1.4.4/go.mod h1:l3mdAwkq5BuhzHwde/uurv3sEJeZMXNpwsxVWU71h+4=
 github.com/golang/mock v1.5.0/go.mod h1:CWnOUgYIOo4TcNZ0wHX3YZCqsaM1I1Jvs6v3mP3KVu8=
+github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc=
 github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs=
 github.com/golang/protobuf v1.1.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
diff --git a/immutable_tree.go b/immutable_tree.go
index bfbca019d..dfeee2b05 100644
--- a/immutable_tree.go
+++ b/immutable_tree.go
@@ -109,8 +109,8 @@ func (t *ImmutableTree) Version() int64 {
 	return t.version
 }
 
-// IsLatestVersion returns true if curren tree is of the latest version, false otherwise.
-func (t *ImmutableTree) IsLatestVersion() bool {
+// IsLatestTreeVersion returns true if curren tree is of the latest version, false otherwise.
+func (t *ImmutableTree) IsLatestTreeVersion() bool {
 	return t.version == t.ndb.getLatestVersion()
 }
 
@@ -236,8 +236,7 @@ func (t *ImmutableTree) Iterate(fn func(key []byte, value []byte) bool) bool {
 
 // Iterator returns an iterator over the immutable tree.
 func (t *ImmutableTree) Iterator(start, end []byte, ascending bool) dbm.Iterator {
-	isFastTraversal := t.IsLatestVersion()
-	if isFastTraversal {
+	if t.IsFastCacheEnabled() {
 		return NewFastIterator(start, end, ascending, t.ndb)
 	} else {
 		return NewIterator(start, end, ascending, t)
@@ -259,6 +258,16 @@ func (t *ImmutableTree) IterateRange(start, end []byte, ascending bool, fn func(
 	})
 }
 
+// GetStorageVersion returns the version of the underlying storage.
+func (t *ImmutableTree) GetStorageVersion() (string) {
+	return t.ndb.getStorageVersion()
+}
+
+// IsFastCacheEnabled returns true if fast storage is enabled, false otherwise.
+func (t *ImmutableTree) IsFastCacheEnabled() bool {
+	return t.IsLatestTreeVersion() && t.ndb.isFastStorageEnabled()
+}
+
 // IterateRangeInclusive makes a callback for all nodes with key between start and end inclusive.
 // If either are nil, then it is open on that side (nil, nil is the same as Iterate). The keys and
 // values must not be modified, since they may point to data stored within IAVL.
diff --git a/mock/db_mock.go b/mock/db_mock.go
new file mode 100644
index 000000000..3212eca20
--- /dev/null
+++ b/mock/db_mock.go
@@ -0,0 +1,420 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: /root/go/pkg/mod/github.com/tendermint/tm-db@v0.6.4/types.go
+
+// Package mocks is a generated GoMock package.
+package mock
+
+import (
+	reflect "reflect"
+
+	gomock "github.com/golang/mock/gomock"
+	db "github.com/tendermint/tm-db"
+)
+
+// MockDB is a mock of DB interface.
+type MockDB struct {
+	ctrl     *gomock.Controller
+	recorder *MockDBMockRecorder
+}
+
+// MockDBMockRecorder is the mock recorder for MockDB.
+type MockDBMockRecorder struct {
+	mock *MockDB
+}
+
+// NewMockDB creates a new mock instance.
+func NewMockDB(ctrl *gomock.Controller) *MockDB {
+	mock := &MockDB{ctrl: ctrl}
+	mock.recorder = &MockDBMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockDB) EXPECT() *MockDBMockRecorder {
+	return m.recorder
+}
+
+// Close mocks base method.
+func (m *MockDB) Close() error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Close")
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Close indicates an expected call of Close.
+func (mr *MockDBMockRecorder) Close() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockDB)(nil).Close))
+}
+
+// Delete mocks base method.
+func (m *MockDB) Delete(arg0 []byte) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Delete", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Delete indicates an expected call of Delete.
+func (mr *MockDBMockRecorder) Delete(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Delete", reflect.TypeOf((*MockDB)(nil).Delete), arg0)
+}
+
+// DeleteSync mocks base method.
+func (m *MockDB) DeleteSync(arg0 []byte) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "DeleteSync", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// DeleteSync indicates an expected call of DeleteSync.
+func (mr *MockDBMockRecorder) DeleteSync(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteSync", reflect.TypeOf((*MockDB)(nil).DeleteSync), arg0)
+}
+
+// Get mocks base method.
+func (m *MockDB) Get(arg0 []byte) ([]byte, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Get", arg0)
+	ret0, _ := ret[0].([]byte)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// Get indicates an expected call of Get.
+func (mr *MockDBMockRecorder) Get(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockDB)(nil).Get), arg0)
+}
+
+// Has mocks base method.
+func (m *MockDB) Has(key []byte) (bool, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Has", key)
+	ret0, _ := ret[0].(bool)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// Has indicates an expected call of Has.
+func (mr *MockDBMockRecorder) Has(key interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Has", reflect.TypeOf((*MockDB)(nil).Has), key)
+}
+
+// Iterator mocks base method.
+func (m *MockDB) Iterator(start, end []byte) (db.Iterator, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Iterator", start, end)
+	ret0, _ := ret[0].(db.Iterator)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// Iterator indicates an expected call of Iterator.
+func (mr *MockDBMockRecorder) Iterator(start, end interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Iterator", reflect.TypeOf((*MockDB)(nil).Iterator), start, end)
+}
+
+// NewBatch mocks base method.
+func (m *MockDB) NewBatch() db.Batch {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "NewBatch")
+	ret0, _ := ret[0].(db.Batch)
+	return ret0
+}
+
+// NewBatch indicates an expected call of NewBatch.
+func (mr *MockDBMockRecorder) NewBatch() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "NewBatch", reflect.TypeOf((*MockDB)(nil).NewBatch))
+}
+
+// Print mocks base method.
+func (m *MockDB) Print() error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Print")
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Print indicates an expected call of Print.
+func (mr *MockDBMockRecorder) Print() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Print", reflect.TypeOf((*MockDB)(nil).Print))
+}
+
+// ReverseIterator mocks base method.
+func (m *MockDB) ReverseIterator(start, end []byte) (db.Iterator, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "ReverseIterator", start, end)
+	ret0, _ := ret[0].(db.Iterator)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// ReverseIterator indicates an expected call of ReverseIterator.
+func (mr *MockDBMockRecorder) ReverseIterator(start, end interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReverseIterator", reflect.TypeOf((*MockDB)(nil).ReverseIterator), start, end)
+}
+
+// Set mocks base method.
+func (m *MockDB) Set(arg0, arg1 []byte) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Set", arg0, arg1)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Set indicates an expected call of Set.
+func (mr *MockDBMockRecorder) Set(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Set", reflect.TypeOf((*MockDB)(nil).Set), arg0, arg1)
+}
+
+// SetSync mocks base method.
+func (m *MockDB) SetSync(arg0, arg1 []byte) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "SetSync", arg0, arg1)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// SetSync indicates an expected call of SetSync.
+func (mr *MockDBMockRecorder) SetSync(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetSync", reflect.TypeOf((*MockDB)(nil).SetSync), arg0, arg1)
+}
+
+// Stats mocks base method.
+func (m *MockDB) Stats() map[string]string {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Stats")
+	ret0, _ := ret[0].(map[string]string)
+	return ret0
+}
+
+// Stats indicates an expected call of Stats.
+func (mr *MockDBMockRecorder) Stats() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Stats", reflect.TypeOf((*MockDB)(nil).Stats))
+}
+
+// MockBatch is a mock of Batch interface.
+type MockBatch struct {
+	ctrl     *gomock.Controller
+	recorder *MockBatchMockRecorder
+}
+
+// MockBatchMockRecorder is the mock recorder for MockBatch.
+type MockBatchMockRecorder struct {
+	mock *MockBatch
+}
+
+// NewMockBatch creates a new mock instance.
+func NewMockBatch(ctrl *gomock.Controller) *MockBatch {
+	mock := &MockBatch{ctrl: ctrl}
+	mock.recorder = &MockBatchMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockBatch) EXPECT() *MockBatchMockRecorder {
+	return m.recorder
+}
+
+// Close mocks base method.
+func (m *MockBatch) Close() error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Close")
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Close indicates an expected call of Close.
+func (mr *MockBatchMockRecorder) Close() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockBatch)(nil).Close))
+}
+
+// Delete mocks base method.
+func (m *MockBatch) Delete(key []byte) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Delete", key)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Delete indicates an expected call of Delete.
+func (mr *MockBatchMockRecorder) Delete(key interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Delete", reflect.TypeOf((*MockBatch)(nil).Delete), key)
+}
+
+// Set mocks base method.
+func (m *MockBatch) Set(key, value []byte) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Set", key, value)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Set indicates an expected call of Set.
+func (mr *MockBatchMockRecorder) Set(key, value interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Set", reflect.TypeOf((*MockBatch)(nil).Set), key, value)
+}
+
+// Write mocks base method.
+func (m *MockBatch) Write() error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Write")
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Write indicates an expected call of Write.
+func (mr *MockBatchMockRecorder) Write() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Write", reflect.TypeOf((*MockBatch)(nil).Write))
+}
+
+// WriteSync mocks base method.
+func (m *MockBatch) WriteSync() error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "WriteSync")
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// WriteSync indicates an expected call of WriteSync.
+func (mr *MockBatchMockRecorder) WriteSync() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "WriteSync", reflect.TypeOf((*MockBatch)(nil).WriteSync))
+}
+
+// MockIterator is a mock of Iterator interface.
+type MockIterator struct {
+	ctrl     *gomock.Controller
+	recorder *MockIteratorMockRecorder
+}
+
+// MockIteratorMockRecorder is the mock recorder for MockIterator.
+type MockIteratorMockRecorder struct {
+	mock *MockIterator
+}
+
+// NewMockIterator creates a new mock instance.
+func NewMockIterator(ctrl *gomock.Controller) *MockIterator {
+	mock := &MockIterator{ctrl: ctrl}
+	mock.recorder = &MockIteratorMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockIterator) EXPECT() *MockIteratorMockRecorder {
+	return m.recorder
+}
+
+// Close mocks base method.
+func (m *MockIterator) Close() error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Close")
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Close indicates an expected call of Close.
+func (mr *MockIteratorMockRecorder) Close() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockIterator)(nil).Close))
+}
+
+// Domain mocks base method.
+func (m *MockIterator) Domain() ([]byte, []byte) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Domain")
+	ret0, _ := ret[0].([]byte)
+	ret1, _ := ret[1].([]byte)
+	return ret0, ret1
+}
+
+// Domain indicates an expected call of Domain.
+func (mr *MockIteratorMockRecorder) Domain() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Domain", reflect.TypeOf((*MockIterator)(nil).Domain))
+}
+
+// Error mocks base method.
+func (m *MockIterator) Error() error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Error")
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Error indicates an expected call of Error.
+func (mr *MockIteratorMockRecorder) Error() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Error", reflect.TypeOf((*MockIterator)(nil).Error))
+}
+
+// Key mocks base method.
+func (m *MockIterator) Key() []byte {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Key")
+	ret0, _ := ret[0].([]byte)
+	return ret0
+}
+
+// Key indicates an expected call of Key.
+func (mr *MockIteratorMockRecorder) Key() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Key", reflect.TypeOf((*MockIterator)(nil).Key))
+}
+
+// Next mocks base method.
+func (m *MockIterator) Next() {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "Next")
+}
+
+// Next indicates an expected call of Next.
+func (mr *MockIteratorMockRecorder) Next() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Next", reflect.TypeOf((*MockIterator)(nil).Next))
+}
+
+// Valid mocks base method.
+func (m *MockIterator) Valid() bool {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Valid")
+	ret0, _ := ret[0].(bool)
+	return ret0
+}
+
+// Valid indicates an expected call of Valid.
+func (mr *MockIteratorMockRecorder) Valid() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Valid", reflect.TypeOf((*MockIterator)(nil).Valid))
+}
+
+// Value mocks base method.
+func (m *MockIterator) Value() []byte {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Value")
+	ret0, _ := ret[0].([]byte)
+	return ret0
+}
+
+// Value indicates an expected call of Value.
+func (mr *MockIteratorMockRecorder) Value() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Value", reflect.TypeOf((*MockIterator)(nil).Value))
+}
diff --git a/mutable_tree.go b/mutable_tree.go
index 585bddf03..ff7cf1cec 100644
--- a/mutable_tree.go
+++ b/mutable_tree.go
@@ -43,19 +43,30 @@ func NewMutableTree(db dbm.DB, cacheSize int) (*MutableTree, error) {
 
 // NewMutableTreeWithOpts returns a new tree with the specified options.
 func NewMutableTreeWithOpts(db dbm.DB, cacheSize int, opts *Options) (*MutableTree, error) {
+	tree := newMutableTreeWithOpts(db, cacheSize, opts)
+
+	if err := tree.ndb.upgradeToFastCacheFromLeaves(); err != nil {
+		return nil, err
+	}
+	return tree, nil
+}
+
+// newMutableTreeWithOpts returns a mutable tree on the default storage version. Used for testing
+func newMutableTreeWithOpts(db dbm.DB, cacheSize int, opts *Options) (*MutableTree) {
 	ndb := newNodeDB(db, cacheSize, opts)
 	head := &ImmutableTree{ndb: ndb}
 
-	return &MutableTree{
-		ImmutableTree: head,
-		lastSaved:     head.clone(),
-		orphans:       map[string]int64{},
-		versions:      map[int64]bool{},
-		allRootLoaded: false,
+	tree := &MutableTree{
+		ImmutableTree:            head,
+		lastSaved:                head.clone(),
+		orphans:                  map[string]int64{},
+		versions:                 map[int64]bool{},
+		allRootLoaded:            false,
 		unsavedFastNodeAdditions: make(map[string]*FastNode),
-		unsavedFastNodeRemovals: make(map[string]interface{}),
-		ndb:           ndb,
-	}, nil
+		unsavedFastNodeRemovals:  make(map[string]interface{}),
+		ndb:                      ndb,
+	}
+	return tree
 }
 
 // IsEmpty returns whether or not the tree has any keys. Only trees that are
@@ -163,7 +174,7 @@ func (t *MutableTree) Iterate(fn func(key []byte, value []byte) bool) (stopped b
 		return false
 	}
 
-	if t.version == t.ndb.getLatestVersion() {
+	if t.IsLatestTreeVersion() && t.IsFastCacheEnabled() {
 		// We need to ensure that we iterate over saved and unsaved state in order.
 		// The strategy is to sort unsaved nodes, the fast node on disk are already sorted.
 		// Then, we keep a pointer to both the unsaved and saved nodes, and iterate over them in sorted order efficiently.
@@ -439,6 +450,12 @@ func (tree *MutableTree) LazyLoadVersion(targetVersion int64) (int64, error) {
 
 	tree.mtx.Lock()
 	defer tree.mtx.Unlock()
+
+	// Attempt to upgrade
+	if err := tree.ndb.upgradeToFastCacheFromLeaves(); err != nil {
+		return 0, err
+	}
+
 	tree.versions[targetVersion] = true
 
 	iTree := &ImmutableTree{
@@ -478,6 +495,11 @@ func (tree *MutableTree) LoadVersion(targetVersion int64) (int64, error) {
 	tree.mtx.Lock()
 	defer tree.mtx.Unlock()
 
+	// Attempt to upgrade
+	if err := tree.ndb.upgradeToFastCacheFromLeaves(); err != nil {
+		return 0, err
+	}
+
 	var latestRoot []byte
 	for version, r := range roots {
 		tree.versions[version] = true
@@ -613,13 +635,16 @@ func (tree *MutableTree) Rollback() {
 // modified, since it may point to data stored within IAVL.
 func (tree *MutableTree) GetVersioned(key []byte, version int64) []byte {
 	if tree.VersionExists(version) {
-		fastNode, _ := tree.ndb.GetFastNode(key)
-		if fastNode == nil && version == tree.ndb.latestVersion {
-			return nil
-		}
-	
-		if  fastNode != nil && fastNode.versionLastUpdatedAt <= version {
-			return fastNode.value
+
+		if tree.IsFastCacheEnabled() {
+			fastNode, _ := tree.ndb.GetFastNode(key)
+			if fastNode == nil && version == tree.ndb.latestVersion {
+				return nil
+			}
+		
+			if fastNode != nil && fastNode.versionLastUpdatedAt <= version {
+				return fastNode.value
+			}
 		}
 
 		t, err := tree.GetImmutable(version)
diff --git a/mutable_tree_test.go b/mutable_tree_test.go
index e22291e2e..306e6f921 100644
--- a/mutable_tree_test.go
+++ b/mutable_tree_test.go
@@ -2,11 +2,14 @@ package iavl
 
 import (
 	"bytes"
+	"errors"
 	"fmt"
 	"runtime"
 	"strconv"
 	"testing"
 
+	"github.com/cosmos/iavl/mock"
+	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -663,3 +666,151 @@ func TestIterator_MutableTree_Invalid(t *testing.T) {
 	require.NotNil(t, itr)
 	require.False(t, itr.Valid())
 }
+
+func TestUpgradeStorageToFastCache_LatestVersion_Success(t *testing.T) {
+	// Setup
+	db := db.NewMemDB()
+	oldTree := newMutableTreeWithOpts(db, 1000, nil)
+	mirror := make(map[string]string)
+	// Fill with some data
+	randomizeTreeAndMirror(t, oldTree, mirror)
+
+	require.True(t, oldTree.IsLatestTreeVersion())
+	require.Equal(t, defaultStorageVersionValue, oldTree.GetStorageVersion())
+
+	// Test new tree from not upgraded db, should upgrade
+	sut, err := NewMutableTree(db, 0)
+	require.NoError(t, err)
+	require.Equal(t, fastStorageVersionValue, sut.GetStorageVersion())
+}
+
+func TestUpgrade_AlreadyUpgraded_Success(t *testing.T) {
+	// Setup
+	db := db.NewMemDB()
+	oldTree := newMutableTreeWithOpts(db, 1000, nil)
+	mirror := make(map[string]string)
+	// Fill with some data
+	randomizeTreeAndMirror(t, oldTree, mirror)
+	// Upgrade
+	require.NoError(t, oldTree.ndb.upgradeToFastCacheFromLeaves())
+	require.Equal(t, fastStorageVersionValue, oldTree.GetStorageVersion())
+
+	// Test new tree from upgraded db
+	sut, err := NewMutableTree(db, 0)
+	require.NoError(t, err)
+	require.Equal(t, fastStorageVersionValue, sut.GetStorageVersion())
+}
+
+func TestUpgradeStorageToFastCache_DbError_Failure(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	dbMock := mock.NewMockDB(ctrl)
+
+	dbMock.EXPECT().Get(gomock.Any()).Return([]byte(defaultStorageVersionValue), nil).Times(1)
+	dbMock.EXPECT().NewBatch().Return(nil).Times(1)
+
+	expectedError := errors.New("some db error")
+
+	dbMock.EXPECT().Iterator(gomock.Any(), gomock.Any()).Return(nil, expectedError).Times(1)
+
+	tree, err := NewMutableTree(dbMock, 0)
+	require.Equal(t, expectedError, err)
+	require.Nil(t, tree)
+}
+
+func TestUpgradeStorageToFastCache_Integration_Upgraded_FastIterator_Success(t *testing.T) {	
+	oldTree, mirror := setupTreeAndMirrorForUpgrade(t)
+	require.Equal(t, defaultStorageVersionValue, oldTree.GetStorageVersion())
+
+	sut, err := NewMutableTreeWithOpts(oldTree.ndb.db, 100, nil)
+	require.NoError(t, err)
+	require.NotNil(t, sut)
+	require.Equal(t, fastStorageVersionValue, sut.GetStorageVersion())
+
+	// Load version
+	version, err := sut.Load()
+	require.NoError(t, err)
+	require.Equal(t, int64(1), version)
+	
+	// Test that upgraded mutable tree iterates as expected
+	t.Run("Mutable tree", func (t *testing.T)  {
+		i := 0
+		oldTree.Iterate(func (k, v []byte) bool {
+			require.Equal(t, []byte(mirror[i][0]), k)
+			require.Equal(t, []byte(mirror[i][1]), v)
+			i++
+			return false	
+		})
+	})
+
+	// Test that upgraded immutable tree iterates as expected
+	t.Run("Immutable tree", func (t *testing.T)  {
+		immutableTree, err := oldTree.GetImmutable(oldTree.version)
+		require.NoError(t, err)
+
+		i := 0
+		immutableTree.Iterate(func (k, v []byte) bool {
+			require.Equal(t, []byte(mirror[i][0]), k)
+			require.Equal(t, []byte(mirror[i][1]), v)
+			i++
+			return false	
+		})
+	})
+}
+
+func TestUpgradeStorageToFastCache_Integration_Upgraded_GetFast_Success(t *testing.T) {
+	oldTree, mirror := setupTreeAndMirrorForUpgrade(t)
+	require.Equal(t, defaultStorageVersionValue, oldTree.GetStorageVersion())
+
+	sut, err := NewMutableTreeWithOpts(oldTree.ndb.db, 100, nil)
+	require.NoError(t, err)
+	require.NotNil(t, sut)
+	require.Equal(t, fastStorageVersionValue, sut.GetStorageVersion())
+
+	// Lazy Load version
+	version, err := sut.LazyLoadVersion(1)
+	require.NoError(t, err)
+	require.Equal(t, int64(1), version)
+
+	t.Run("Mutable tree", func (t *testing.T)  {
+		for _, kv := range mirror {
+			v := sut.GetFast([]byte(kv[0]))
+			require.Equal(t, []byte(kv[1]), v)
+		}
+	})
+
+	t.Run("Immutable tree", func (t *testing.T)  {
+		immutableTree, err := sut.GetImmutable(sut.version)
+		require.NoError(t, err)
+
+		for _, kv := range mirror {
+			v := immutableTree.GetFast([]byte(kv[0]))
+			require.Equal(t, []byte(kv[1]), v)
+		}
+	})
+}
+
+func setupTreeAndMirrorForUpgrade(t *testing.T) (*MutableTree, [][]string) {
+	db := db.NewMemDB()
+
+	tree := newMutableTreeWithOpts(db, 0, nil)
+
+	var keyPrefix, valPrefix string = "key", "val"
+
+	mirror := make([][]string, 0, 10)
+	for i := 0; i < 10; i++ {
+		key := fmt.Sprintf("%s_%d", keyPrefix, i)
+		val := fmt.Sprintf("%s_%d", valPrefix, i)
+		mirror = append(mirror, []string{key, val})
+		require.False(t, tree.Set([]byte(key), []byte(val)))
+	}
+
+	_, _, err := tree.SaveVersion()
+	require.NoError(t, err)
+
+	// Delete fast nodes from database to mimic a version with no upgrade
+	for i := 0; i < 10; i++ {
+		key := fmt.Sprintf("%s_%d", keyPrefix, i)
+		require.NoError(t, db.Delete(fastKeyFormat.Key([]byte(key))))
+	}
+	return tree, mirror
+}
diff --git a/nodedb.go b/nodedb.go
index 75aa04f73..9e379614b 100644
--- a/nodedb.go
+++ b/nodedb.go
@@ -17,6 +17,10 @@ const (
 	int64Size      = 8
 	hashSize       = sha256.Size
 	genesisVersion = 1
+	storageVersionKey = "chain_version"
+	// Using semantic versioning: https://semver.org/
+	defaultStorageVersionValue = "1.0.0"
+	fastStorageVersionValue = "1.1.0"
 )
 
 var (
@@ -41,6 +45,11 @@ var (
 	// return result_version. Else, go through old (slow) IAVL get method that walks through tree.
 	fastKeyFormat = NewKeyFormat('f', 0) // f<keystring>
 
+	// Key Format for storing metadata about the chain such as the vesion number.
+	// The value at an entry will be in a variable format and up to the caller to
+	// decide how to parse.
+	metadataKeyFormat = NewKeyFormat('m', 0) // v<keystring>
+
 	// Root nodes are indexed separately by their version
 	rootKeyFormat = NewKeyFormat('r', int64Size) // r<version>
 )
@@ -51,6 +60,7 @@ type nodeDB struct {
 	batch          dbm.Batch        // Batched writing buffer.
 	opts           Options          // Options to customize for pruning/writing
 	versionReaders map[int64]uint32 // Number of active version readers
+	storageVersion string              // Chain version
 
 	latestVersion  int64
 	nodeCache      map[string]*list.Element // Node cache.
@@ -67,6 +77,13 @@ func newNodeDB(db dbm.DB, cacheSize int, opts *Options) *nodeDB {
 		o := DefaultOptions()
 		opts = &o
 	}
+
+	storeVersion, err := db.Get(metadataKeyFormat.Key([]byte(storageVersionKey)))
+
+	if err != nil || storeVersion == nil {
+		storeVersion = []byte(defaultStorageVersionValue)
+	}
+
 	return &nodeDB{
 		db:                 db,
 		batch:              db.NewBatch(),
@@ -79,6 +96,7 @@ func newNodeDB(db dbm.DB, cacheSize int, opts *Options) *nodeDB {
 		fastNodeCacheSize:  cacheSize,
 		fastNodeCacheQueue: list.New(),
 		versionReaders:     make(map[int64]uint32, 8),
+		storageVersion: 	 string(storeVersion),
 	}
 }
 
@@ -94,7 +112,7 @@ func (ndb *nodeDB) GetNode(hash []byte) *Node {
 
 	// Check the cache.
 	if elem, ok := ndb.nodeCache[string(hash)]; ok {
-		// Already exists. Move to back of nodeCacheQueue.
+			// Already exists. Move to back of nodeCacheQueue.
 		ndb.nodeCacheQueue.MoveToBack(elem)
 		return elem.Value.(*Node)
 	}
@@ -121,6 +139,10 @@ func (ndb *nodeDB) GetNode(hash []byte) *Node {
 }
 
 func (ndb *nodeDB) GetFastNode(key []byte) (*FastNode, error) {
+	if !ndb.isFastStorageEnabled() {
+		return nil, errors.New("storage version is not fast")
+	}
+
 	ndb.mtx.Lock()
 	defer ndb.mtx.Unlock()
 
@@ -128,7 +150,6 @@ func (ndb *nodeDB) GetFastNode(key []byte) (*FastNode, error) {
 		return nil, fmt.Errorf("nodeDB.GetFastNode() requires key, len(key) equals 0")
 	}
 
-	// TODO make a second write lock just for fastNodeCacheQueue later
 	// Check the cache.
 	if elem, ok := ndb.fastNodeCache[string(key)]; ok {
 		// Already exists. Move to back of fastNodeCacheQueue.
@@ -189,6 +210,53 @@ func (ndb *nodeDB) SaveFastNode(node *FastNode) error {
 	return ndb.saveFastNodeUnlocked(node)
 }
 
+func (ndb *nodeDB) setStorageVersion(newVersion string) error {
+	if err := ndb.db.Set(metadataKeyFormat.Key([]byte(storageVersionKey)), []byte(newVersion)); err != nil {
+		return err
+	}
+	ndb.storageVersion = string(newVersion)
+	return nil
+}
+
+func (ndb *nodeDB) upgradeToFastCacheFromLeaves() error {
+	if ndb.isFastStorageEnabled() {
+		return nil
+	}
+
+	err := ndb.traverseNodes(func(hash []byte, node *Node) error {
+		if node.isLeaf() && node.version == ndb.getLatestVersion() {
+			fastNode := NewFastNode(node.key, node.value, node.version)
+			if err := ndb.saveFastNodeUnlocked(fastNode); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+
+	if err != nil {
+		return err
+	}
+
+	if err := ndb.batch.Set(metadataKeyFormat.Key([]byte(storageVersionKey)), []byte(fastStorageVersionValue)); err != nil {
+		return err
+	}
+
+	if err = ndb.resetBatch(); err != nil {
+		return err
+	}
+
+	ndb.storageVersion = fastStorageVersionValue
+	return err
+}
+
+func (ndb *nodeDB) getStorageVersion() string {
+	return ndb.storageVersion
+}
+
+func (ndb *nodeDB) isFastStorageEnabled() bool {
+	return ndb.getStorageVersion() >= fastStorageVersionValue
+}
+
 // SaveNode saves a FastNode to disk.
 func (ndb *nodeDB) saveFastNodeUnlocked(node *FastNode) error {
 	if node.key == nil {
@@ -260,7 +328,7 @@ func (ndb *nodeDB) SaveBranch(node *Node) []byte {
 }
 
 // resetBatch reset the db batch, keep low memory used
-func (ndb *nodeDB) resetBatch() {
+func (ndb *nodeDB) resetBatch() error {
 	var err error
 	if ndb.opts.Sync {
 		err = ndb.batch.WriteSync()
@@ -268,10 +336,16 @@ func (ndb *nodeDB) resetBatch() {
 		err = ndb.batch.Write()
 	}
 	if err != nil {
-		panic(err)
+		return err
 	}
-	ndb.batch.Close()
+	err = ndb.batch.Close()
+	if err != nil {
+		return err
+	}
+
 	ndb.batch = ndb.db.NewBatch()
+
+	return nil
 }
 
 // DeleteVersion deletes a tree version from disk.
diff --git a/nodedb_test.go b/nodedb_test.go
index 8e891a15e..6232eed71 100644
--- a/nodedb_test.go
+++ b/nodedb_test.go
@@ -2,8 +2,15 @@ package iavl
 
 import (
 	"encoding/binary"
+	"errors"
 	"math/rand"
 	"testing"
+
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/require"
+	db "github.com/tendermint/tm-db"
+
+	"github.com/cosmos/iavl/mock"
 )
 
 func BenchmarkNodeKey(b *testing.B) {
@@ -22,6 +29,73 @@ func BenchmarkOrphanKey(b *testing.B) {
 	}
 }
 
+func TestNewNoDbChain_ChainVersionInDb_Success(t *testing.T) {
+	const expectedVersion = fastStorageVersionValue
+	
+	ctrl := gomock.NewController(t)
+	dbMock := mock.NewMockDB(ctrl)
+
+	dbMock.EXPECT().Get(gomock.Any()).Return([]byte(expectedVersion), nil).Times(1)
+	dbMock.EXPECT().NewBatch().Return(nil).Times(1)
+	
+	ndb := newNodeDB(dbMock, 0, nil)
+	require.Equal(t, expectedVersion, ndb.storageVersion)
+}
+
+func TestNewNoDbChain_ErrorInConstructor_DefaultSet(t *testing.T) {
+	const expectedVersion = defaultStorageVersionValue
+	
+	ctrl := gomock.NewController(t)
+	dbMock := mock.NewMockDB(ctrl)
+
+	dbMock.EXPECT().Get(gomock.Any()).Return(nil, errors.New("some db error")).Times(1)
+	dbMock.EXPECT().NewBatch().Return(nil).Times(1)
+	
+	ndb := newNodeDB(dbMock, 0, nil)
+	require.Equal(t, expectedVersion, string(ndb.getStorageVersion()))
+}
+
+func TestNewNoDbChain_DoesNotExist_DefaultSet(t *testing.T) {
+	const expectedVersion = defaultStorageVersionValue
+	
+	ctrl := gomock.NewController(t)
+	dbMock := mock.NewMockDB(ctrl)
+
+	dbMock.EXPECT().Get(gomock.Any()).Return(nil, nil).Times(1)
+	dbMock.EXPECT().NewBatch().Return(nil).Times(1)
+	
+	ndb := newNodeDB(dbMock, 0, nil)
+	require.Equal(t, expectedVersion, string(ndb.getStorageVersion()))
+}
+
+func TestSetChainVersion_Success(t *testing.T) {
+	const expectedVersion = fastStorageVersionValue
+	
+	db := db.NewMemDB()
+	
+	ndb := newNodeDB(db, 0, nil)
+	require.Equal(t, defaultStorageVersionValue, string(ndb.getStorageVersion()))
+
+	err := ndb.setStorageVersion(expectedVersion)
+	require.NoError(t, err)
+	require.Equal(t, expectedVersion, string(ndb.getStorageVersion()))
+}
+
+func TestSetChainVersion_Failure_OldKept(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	
+	dbMock := mock.NewMockDB(ctrl)
+	dbMock.EXPECT().Get(gomock.Any()).Return([]byte(defaultStorageVersionValue), nil).Times(1)
+	dbMock.EXPECT().NewBatch().Return(nil).Times(1)
+	dbMock.EXPECT().Set(gomock.Any(), gomock.Any()).Return(errors.New("some db error")).Times(1)
+
+	ndb := newNodeDB(dbMock, 0, nil)
+	require.Equal(t, defaultStorageVersionValue, string(ndb.getStorageVersion()))
+
+	ndb.setStorageVersion(fastStorageVersionValue)
+	require.Equal(t, defaultStorageVersionValue, string(ndb.getStorageVersion()))
+}
+
 func makeHashes(b *testing.B, seed int64) [][]byte {
 	b.StopTimer()
 	rnd := rand.NewSource(seed)
diff --git a/testutils_test.go b/testutils_test.go
index e2d58e5b7..a4265d251 100644
--- a/testutils_test.go
+++ b/testutils_test.go
@@ -179,6 +179,9 @@ func getRandomizedTreeAndMirror(t *testing.T) (*MutableTree, map[string]string)
 }
 
 func randomizeTreeAndMirror(t *testing.T, tree *MutableTree, mirror map[string]string) {
+	if mirror == nil {
+		mirror = make(map[string]string)
+	}
 	const keyValLength = 5
 
 	numberOfSets := 1000
diff --git a/tree_random_test.go b/tree_random_test.go
index edc78b4c3..522fa01b8 100644
--- a/tree_random_test.go
+++ b/tree_random_test.go
@@ -7,6 +7,7 @@ import (
 	"math/rand"
 	"os"
 	"sort"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -333,20 +334,28 @@ func assertEmptyDatabase(t *testing.T, tree *MutableTree) {
 	require.NoError(t, err)
 
 	var (
-		firstKey []byte
-		count    int
+		foundKeys []string
 	)
 	for ; iter.Valid(); iter.Next() {
-		count++
-		if firstKey == nil {
-			firstKey = iter.Key()
-		}
+		foundKeys = append(foundKeys, string(iter.Key()))
 	}
 	require.NoError(t, iter.Error())
-	require.EqualValues(t, 1, count, "Found %v database entries, expected 1", count)
+	require.EqualValues(t, 2, len(foundKeys), "Found %v database entries, expected 1", len(foundKeys)) // 1 for storage version and 1 for root
+
+	firstKey := foundKeys[0]
+	secondKey := foundKeys[1]
+
+	require.True(t, strings.HasPrefix(firstKey, metadataKeyFormat.Prefix()))
+	require.True(t, strings.HasPrefix(secondKey, rootKeyFormat.Prefix()))
+
+	require.Equal(t, string(metadataKeyFormat.KeyBytes([]byte(storageVersionKey))), firstKey, "Unexpected storage version key")
+
+	storageVersionValue, err := tree.ndb.db.Get([]byte(firstKey))
+	require.NoError(t, err)
+	require.Equal(t, []byte(fastStorageVersionValue), storageVersionValue)
 
 	var foundVersion int64
-	rootKeyFormat.Scan(firstKey, &foundVersion)
+	rootKeyFormat.Scan([]byte(secondKey), &foundVersion)
 	require.Equal(t, version, foundVersion, "Unexpected root version")
 }
 
diff --git a/tree_test.go b/tree_test.go
index a3c42b0bd..a07801704 100644
--- a/tree_test.go
+++ b/tree_test.go
@@ -383,9 +383,10 @@ func TestVersionedTree(t *testing.T) {
 	tree, err := NewMutableTree(d, 0)
 	require.NoError(err)
 
-	// We start with zero keys in the databse.
-	require.Equal(0, tree.ndb.size())
+	// We start with one key in the database that represents storage version.
+	require.Equal(1, tree.ndb.size())
 	require.True(tree.IsEmpty())
+	require.Equal(fastStorageVersionValue, tree.GetStorageVersion())
 
 	// version 0