go/consensus/tendermint: Use MKVS for storing application state

.changelog/1898.breaking.md

@@ -0,0 +1 @@
+go/consensus/tendermint: Use MKVS for storing application state

go/common/version/version.go

@@ -70,7 +70,7 @@ var (
 	//
 	// NOTE: Any change in the major or minor versions are considered
 	//       breaking changes for the protocol.
-	ConsensusProtocol = Version{Major: 0, Minor: 23, Patch: 0}
+	ConsensusProtocol = Version{Major: 0, Minor: 24, Patch: 0}
 
 	// Tendermint exposes the tendermint core version.
 	Tendermint = parseSemVerStr(version.TMCoreSemVer)

go/consensus/tendermint/abci/context.go

@@ -3,14 +3,15 @@ package abci
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"time"
 
-	"github.com/tendermint/iavl"
 	"github.com/tendermint/tendermint/abci/types"
 
 	"github.com/oasislabs/oasis-core/go/common/crypto/signature"
 	"github.com/oasislabs/oasis-core/go/common/logging"
 	"github.com/oasislabs/oasis-core/go/consensus/tendermint/api"
+	mkvs "github.com/oasislabs/oasis-core/go/storage/mkvs/urkel"
 )
 
 type contextKey struct{}
@@ -57,7 +58,7 @@ func (m ContextMode) String() string {
 
 // Context is the context of processing a transaction/block.
 type Context struct {
-	ctx context.Context
+	context.Context
 
 	mode        ContextMode
 	currentTime time.Time
@@ -69,22 +70,13 @@ type Context struct {
 	txSigner signature.PublicKey
 
 	appState    ApplicationState
-	state       *iavl.MutableTree
+	state       mkvs.Tree
 	blockHeight int64
 	blockCtx    *BlockContext
 
-	logger *logging.Logger
-}
+	stateCheckpoint *StateCheckpoint
 
-// NewMockContext creates a new mock context for use in tests.
-func NewMockContext(mode ContextMode, now time.Time) *Context {
-	return &Context{
-		ctx:           context.Background(),
-		mode:          mode,
-		currentTime:   now,
-		gasAccountant: NewNopGasAccountant(),
-		logger:        logging.GetLogger("consensus/tendermint/abci").With("mode", mode),
-	}
+	logger *logging.Logger
 }
 
 // FromCtx extracts an ABCI context from a context.Context if one has been
@@ -99,25 +91,26 @@ func FromCtx(ctx context.Context) *Context {
 // After calling this method, the context should no longer be used.
 func (c *Context) Close() {
 	if c.IsSimulation() {
-		c.state.Rollback()
+		if tree, ok := c.state.(mkvs.ClosableTree); ok {
+			tree.Close()
+		}
 	}
 
 	c.events = nil
 	c.appState = nil
 	c.state = nil
 	c.blockCtx = nil
+
+	if c.stateCheckpoint != nil {
+		panic("context: open checkpoint was never committed or discarded")
+	}
 }
 
 // Logger returns the logger associated with this context.
 func (c *Context) Logger() *logging.Logger {
 	return c.logger
 }
 
-// Ctx returns a context.Context that is associated with this ABCI context.
-func (c *Context) Ctx() context.Context {
-	return c.ctx
-}
-
 // Mode returns the context mode.
 func (c *Context) Mode() ContextMode {
 	return c.mode
@@ -225,8 +218,11 @@ func (c *Context) Now() time.Time {
 	return c.currentTime
 }
 
-// State returns the mutable state tree.
-func (c *Context) State() *iavl.MutableTree {
+// State returns the state tree associated with this context.
+func (c *Context) State() mkvs.KeyValueTree {
+	if c.stateCheckpoint != nil {
+		return c.stateCheckpoint.overlay
+	}
 	return c.state
 }
 
@@ -253,34 +249,47 @@ func (c *Context) BlockContext() *BlockContext {
 	return c.blockCtx
 }
 
-// NewStateCheckpoint creates a new state checkpoint.
-func (c *Context) NewStateCheckpoint() *StateCheckpoint {
-	return &StateCheckpoint{
-		ImmutableTree: *c.State().ImmutableTree,
-		ctx:           c,
+// StartCheckpoint starts a new state checkpoint. Any further updates to the context's state will
+// be performed against the checkpoint and will only be committed in case of an explicit Commit.
+//
+// The caller must make sure to call either Close or Commit on the checkpoint, otherwise this will
+// leak resources.
+func (c *Context) StartCheckpoint() *StateCheckpoint {
+	if c.stateCheckpoint != nil {
+		panic("context: nested checkpoints are not allowed")
 	}
+	c.stateCheckpoint = &StateCheckpoint{
+		ctx:     c,
+		overlay: mkvs.NewOverlay(c.state),
+	}
+	return c.stateCheckpoint
 }
 
 // StateCheckpoint is a state checkpoint that can be used to rollback state.
 type StateCheckpoint struct {
-	iavl.ImmutableTree
-
-	ctx *Context
+	ctx     *Context
+	overlay mkvs.OverlayTree
 }
 
-// Close releases resources associated with the checkpoint.
+// Close releases resources associated with the checkpoint without committing it.
 func (sc *StateCheckpoint) Close() {
+	if sc.ctx == nil {
+		return
+	}
+	sc.overlay.Close()
+	sc.ctx.stateCheckpoint = nil
 	sc.ctx = nil
 }
 
-// Rollback rolls back the active state to the one from the checkpoint.
-func (sc *StateCheckpoint) Rollback() {
+// Commit commits any changes performed since the checkpoint was created.
+func (sc *StateCheckpoint) Commit() {
 	if sc.ctx == nil {
 		return
 	}
-	st := sc.ctx.State()
-	st.Rollback()
-	st.ImmutableTree = &sc.ImmutableTree
+	if err := sc.overlay.Commit(sc.ctx); err != nil {
+		panic(fmt.Errorf("context: failed to commit checkpoint: %w", err))
+	}
+	sc.Close()
 }
 
 // BlockContextKey is an interface for a block context key.

go/consensus/tendermint/abci/context_test.go

@@ -0,0 +1,82 @@
+package abci
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestContext(t *testing.T) {
+	require := require.New(t)
+
+	now := time.Unix(1580461674, 0)
+	appState := NewMockApplicationState(MockApplicationStateConfig{})
+	ctx := appState.NewContext(ContextBeginBlock, now)
+	defer ctx.Close()
+
+	// Add some state.
+	tree := ctx.State()
+	err := tree.Insert(ctx, []byte("key"), []byte("value"))
+	require.NoError(err, "Insert")
+
+	// Test checkpoints.
+	cp := ctx.StartCheckpoint()
+	// Should panic on nested checkpoints.
+	require.Panics(func() { ctx.StartCheckpoint() })
+	overlay := ctx.State()
+	require.NotEqual(&tree, &overlay, "new State() should return the overlay")
+
+	// Existing state should be there.
+	value, err := overlay.Get(ctx, []byte("key"))
+	require.NoError(err, "Get")
+	require.EqualValues([]byte("value"), value)
+
+	// Add some state to the overlay.
+	err = overlay.Insert(ctx, []byte("blah"), []byte("value2"))
+	require.NoError(err, "Insert")
+	err = overlay.Remove(ctx, []byte("key"))
+	require.NoError(err, "Remove")
+
+	// Make sure updates didn't leak.
+	value, err = tree.Get(ctx, []byte("key"))
+	require.NoError(err, "Get")
+	require.EqualValues([]byte("value"), value, "updates should not leak outside checkpoint")
+	value, err = tree.Get(ctx, []byte("blah"))
+	require.NoError(err, "Get")
+	require.Nil(value, "updates should not leak outside checkpoint")
+
+	// Commit checkpoint.
+	cp.Commit()
+	newTree := ctx.State()
+	require.Equal(&tree, &newTree, "new State() should return the original tree")
+
+	// Make sure updates were applied.
+	value, err = tree.Get(ctx, []byte("key"))
+	require.NoError(err, "Get")
+	require.Nil(value, "updates should have been applied")
+	value, err = tree.Get(ctx, []byte("blah"))
+	require.NoError(err, "Get")
+	require.EqualValues([]byte("value2"), value, "updates should have been applied")
+
+	ctx.Close()
+}
+
+type testBlockContextKey struct{}
+
+func (k testBlockContextKey) NewDefault() interface{} {
+	return 42
+}
+
+func TestBlockContext(t *testing.T) {
+	require := require.New(t)
+
+	bc := NewBlockContext()
+
+	value := bc.Get(testBlockContextKey{})
+	require.EqualValues(42, value, "block context key should have default value")
+
+	bc.Set(testBlockContextKey{}, 21)
+	value = bc.Get(testBlockContextKey{})
+	require.EqualValues(21, value, "block context key should have correct value")
+}

go/consensus/tendermint/abci/errors.go

@@ -0,0 +1,32 @@
+package abci
+
+import (
+	"errors"
+	"fmt"
+)
+
+type errorUnavailableState struct {
+	inner error
+}
+
+func (e *errorUnavailableState) Error() string {
+	return fmt.Sprintf("unavailable/corrupted state: %s", e.inner.Error())
+}
+
+func (e *errorUnavailableState) Unwrap() error {
+	return e.inner
+}
+
+// UnavailableStateError wraps an error in an unavailable state error.
+func UnavailableStateError(err error) error {
+	if err == nil {
+		return nil
+	}
+	return &errorUnavailableState{err}
+}
+
+// IsUnavailableStateError returns true if any error in err's chain is an unavailable state error.
+func IsUnavailableStateError(err error) bool {
+	var e *errorUnavailableState
+	return errors.As(err, &e)
+}

go/consensus/tendermint/abci/mux.go

@@ -32,7 +32,6 @@ import (
 
 const (
 	stateKeyGenesisDigest   = "OasisGenesisDigest"
-	stateKeyGenesisRequest  = "OasisGenesisRequest"
 	stateKeyInitChainEvents = "OasisInitChainEvents"
 
 	metricsUpdateInterval = 10 * time.Second
@@ -57,6 +56,7 @@ var (
 // ApplicationConfig is the configuration for the consensus application.
 type ApplicationConfig struct {
 	DataDir         string
+	StorageBackend  string
 	Pruning         PruneConfig
 	HaltEpochHeight epochtime.EpochTime
 	MinGasPrice     uint64
@@ -218,11 +218,6 @@ func (a *ApplicationServer) RegisterHaltHook(hook func(ctx context.Context, bloc
 	a.mux.registerHaltHook(hook)
 }
 
-// Pruner returns the ABCI state pruner.
-func (a *ApplicationServer) Pruner() StatePruner {
-	return a.mux.state.statePruner
-}
-
 // SetEpochtime sets the mux epochtime.
 //
 // Epochtime must be set before the multiplexer can be used.
@@ -396,7 +391,10 @@ func (mux *abciMux) InitChain(req types.RequestInitChain) types.ResponseInitChai
 	tmp := bytes.NewBuffer(nil)
 	_ = types.WriteMessage(&req, tmp)
 	genesisDigest := sha512.Sum512_256(tmp.Bytes())
-	mux.state.deliverTxTree.Set([]byte(stateKeyGenesisDigest), genesisDigest[:])
+	err = mux.state.deliverTxTree.Insert(mux.state.ctx, []byte(stateKeyGenesisDigest), genesisDigest[:])
+	if err != nil {
+		panic(err)
+	}
 
 	resp := mux.BaseApplication.InitChain(req)
 
@@ -422,11 +420,6 @@ func (mux *abciMux) InitChain(req types.RequestInitChain) types.ResponseInitChai
 		mux.logger.Debug("Genesis hook dispatch complete")
 	}()
 
-	// TODO: remove stateKeyGenesisRequest here, see oasis-core#2426
-	b, _ = req.Marshal()
-	mux.state.deliverTxTree.Set([]byte(stateKeyGenesisRequest), b)
-	mux.state.checkTxTree.Set([]byte(stateKeyGenesisRequest), b)
-
 	// Call InitChain() on all applications.
 	mux.logger.Debug("InitChain: initializing applications")
 
@@ -452,7 +445,10 @@ func (mux *abciMux) InitChain(req types.RequestInitChain) types.ResponseInitChai
 	// Since returning emitted events doesn't work for InitChain() response yet,
 	// we store those and return them in BeginBlock().
 	evBinary := cbor.Marshal(ctx.GetEvents())
-	mux.state.deliverTxTree.Set([]byte(stateKeyInitChainEvents), evBinary)
+	err = mux.state.deliverTxTree.Insert(mux.state.ctx, []byte(stateKeyInitChainEvents), evBinary)
+	if err != nil {
+		panic(err)
+	}
 
 	return resp
 }
@@ -483,7 +479,7 @@ func (mux *abciMux) BeginBlock(req types.RequestBeginBlock) types.ResponseBeginB
 	ctx := mux.state.NewContext(ContextBeginBlock, mux.currentTime)
 	defer ctx.Close()
 
-	currentEpoch, err := mux.state.GetCurrentEpoch(ctx.Ctx())
+	currentEpoch, err := mux.state.GetCurrentEpoch(ctx)
 	if err != nil {
 		panic("mux: can't get current epoch in BeginBlock")
 	}
@@ -547,14 +543,19 @@ func (mux *abciMux) BeginBlock(req types.RequestBeginBlock) types.ResponseBeginB
 	// During the first block, also collect and prepend application events
 	// generated during InitChain to BeginBlock events.
 	if mux.state.BlockHeight() == 0 {
-		_, evBinary := mux.state.deliverTxTree.Get([]byte(stateKeyInitChainEvents))
+		evBinary, err := mux.state.deliverTxTree.Get(ctx, []byte(stateKeyInitChainEvents))
+		if err != nil {
+			panic(fmt.Errorf("mux: BeginBlock: failed to query init chain events: %w", err))
+		}
 		if evBinary != nil {
 			var events []types.Event
 			_ = cbor.Unmarshal(evBinary, &events)
 
 			response.Events = append(events, response.Events...)
 
-			mux.state.deliverTxTree.Remove([]byte(stateKeyInitChainEvents))
+			if err := mux.state.deliverTxTree.Remove(ctx, []byte(stateKeyInitChainEvents)); err != nil {
+				panic(fmt.Errorf("mux: BeginBlock: failed to remove init chain events: %w", err))
+			}
 		}
 	}
 
@@ -729,6 +730,14 @@ func (mux *abciMux) DeliverTx(req types.RequestDeliverTx) types.ResponseDeliverT
 	defer ctx.Close()
 
 	if err := mux.executeTx(ctx, req.Tx); err != nil {
+		if IsUnavailableStateError(err) {
+			// Make sure to not commit any transactions which include results based on unavailable
+			// and/or corrupted state -- doing so can further corrupt state.
+			ctx.Logger().Error("unavailable and/or corrupted state detected during tx processing",
+				"err", err,
+			)
+			panic(err)
+		}
 		module, code := errors.Code(err)
 
 		return types.ResponseDeliverTx{