Merge pull request #6057 from oasisprotocol/ptrus/stable/20.12.x/archive

[BACKPORT/20.12.x] Add consensus archive mode support

Author: ptrus

.changelog/4539.feature.md

@@ -0,0 +1,6 @@
+Add archive mode support
+
+Node started in archive mode only serves existing consensus and runtime
+states. The node has all unneeded consensus and P2P functionality disabled so
+it wont participate in the network. Archive mode can be set using the
+`consensus.tendermint.mode` setting.

go/consensus/api/api.go

@@ -4,6 +4,7 @@ package api
 
 import (
 	"context"
+	"fmt"
 	"strings"
 	"time"
 
@@ -34,6 +35,61 @@ const (
 	HeightLatest int64 = 0
 )
 
+// Mode is the consensus node mode.
+type Mode string
+
+const (
+	// ModeFull is the name of the full node consensus mode.
+	ModeFull Mode = "full"
+	// ModeSeed is the name of the seed-only node consensus mode.
+	ModeSeed Mode = "seed"
+	// ModeArchive is the name of the archive node consensus mode.
+	ModeArchive Mode = "archive"
+)
+
+// MarshalText encodes a Mode into text form.
+func (m Mode) MarshalText() ([]byte, error) {
+	switch m {
+	case ModeFull:
+		return []byte(ModeFull.String()), nil
+	case ModeSeed:
+		return []byte(ModeSeed.String()), nil
+	case ModeArchive:
+		return []byte(ModeArchive.String()), nil
+	default:
+		return nil, fmt.Errorf("invalid mode: %s", string(m))
+	}
+}
+
+// UnmarshalText decodes a text marshaled consensus mode.
+func (m *Mode) UnmarshalText(text []byte) error {
+	switch string(text) {
+	case ModeFull.String():
+		*m = ModeFull
+	case ModeSeed.String():
+		*m = ModeSeed
+	case ModeArchive.String():
+		*m = ModeArchive
+	default:
+		return fmt.Errorf("invalid consensus mode: %s", string(text))
+	}
+	return nil
+}
+
+// String returns a string representation of the mode.
+func (m Mode) String() string {
+	switch m {
+	case ModeFull:
+		return string(ModeFull)
+	case ModeSeed:
+		return string(ModeSeed)
+	case ModeArchive:
+		return string(ModeArchive)
+	default:
+		return fmt.Sprintf("[unknown consensus mode: %s]", string(m))
+	}
+}
+
 var (
 	// ErrNoCommittedBlocks is the error returned when there are no committed
 	// blocks and as such no state can be queried.

go/consensus/api/api_test.go

@@ -0,0 +1,37 @@
+package api
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestConsensusMode(t *testing.T) {
+	require := require.New(t)
+
+	// Test valid Modes.
+	for _, k := range []Mode{
+		ModeArchive,
+		ModeFull,
+		ModeSeed,
+	} {
+		enc, err := k.MarshalText()
+		require.NoError(err, "MarshalText")
+
+		var s Mode
+		err = s.UnmarshalText(enc)
+		require.NoError(err, "UnmarshalText")
+
+		require.Equal(k, s, "consensus mode should round-trip")
+	}
+
+	// Test invalid Mode.
+	sr := Mode("abc")
+	require.Equal("[unknown consensus mode: abc]", sr.String())
+	enc, err := sr.MarshalText()
+	require.Nil(enc, "MarshalText on invalid consensus mode should be nil")
+	require.Error(err, "MarshalText on invalid consensus mode should error")
+
+	err = sr.UnmarshalText([]byte("invalid consensus mode"))
+	require.Error(err, "UnmarshalText on invalid consensus mode should error")
+}

go/consensus/api/submission.go

@@ -44,6 +44,12 @@ func (pd *staticPriceDiscovery) GasPrice(ctx context.Context) (*quantity.Quantit
 	return pd.price.Clone(), nil
 }
 
+type noOpPriceDiscovery struct{}
+
+func (pd *noOpPriceDiscovery) GasPrice(ctx context.Context) (*quantity.Quantity, error) {
+	return nil, transaction.ErrMethodNotSupported
+}
+
 // SubmissionManager is a transaction submission manager interface.
 type SubmissionManager interface {
 	// SignAndSubmitTx populates the nonce and fee fields in the transaction, signs the transaction
@@ -170,3 +176,11 @@ func NewSubmissionManager(backend ClientBackend, priceDiscovery PriceDiscovery,
 func SignAndSubmitTx(ctx context.Context, backend Backend, signer signature.Signer, tx *transaction.Transaction) error {
 	return backend.SubmissionManager().SignAndSubmitTx(ctx, signer, tx)
 }
+
+// NoOpSubmissionManager implements a submission manager that doesn't support submitting transactions.
+type NoOpSubmissionManager struct{}
+
+// Implements SubmissionManager.
+func (m *NoOpSubmissionManager) SignAndSubmitTx(ctx context.Context, signer signature.Signer, tx *transaction.Transaction) error {
+	return transaction.ErrMethodNotSupported
+}

go/consensus/api/transaction/transaction.go

@@ -23,6 +23,9 @@ var (
 	// ErrInvalidNonce is the error returned when a nonce is invalid.
 	ErrInvalidNonce = errors.New(moduleName, 1, "transaction: invalid nonce")
 
+	// ErrMethodNotSupported is the error returned if transaction method is not supported.
+	ErrMethodNotSupported = errors.New(moduleName, 5, "transaction: method not supported")
+
 	// SignatureContext is the context used for signing transactions.
 	SignatureContext = signature.NewContext("oasis-core/consensus: tx", signature.WithChainSeparation())
 

go/consensus/tendermint/api/api.go

@@ -199,7 +199,7 @@ type Backend interface {
 
 	// WatchTendermintBlocks returns a stream of Tendermint blocks as they are
 	// returned via the `EventDataNewBlock` query.
-	WatchTendermintBlocks() (<-chan *tmtypes.Block, *pubsub.Subscription)
+	WatchTendermintBlocks() (<-chan *tmtypes.Block, *pubsub.Subscription, error)
 
 	// GetLastRetainedVersion returns the earliest retained version the ABCI
 	// state.
@@ -308,8 +308,7 @@ type ServiceClient interface {
 
 // BaseServiceClient is a default ServiceClient implementation that provides noop implementations of
 // all the delivery methods. Implementations should override them as needed.
-type BaseServiceClient struct {
-}
+type BaseServiceClient struct{}
 
 // Implements ServiceClient.
 func (bsc *BaseServiceClient) DeliverBlock(ctx context.Context, height int64) error {

go/consensus/tendermint/full/archive.go

@@ -0,0 +1,230 @@
+package full
+
+import (
+	"context"
+	"fmt"
+	"path/filepath"
+	"sync"
+	"time"
+
+	"github.com/spf13/viper"
+	abcicli "github.com/tendermint/tendermint/abci/client"
+	tmconfig "github.com/tendermint/tendermint/config"
+	tmsync "github.com/tendermint/tendermint/libs/sync"
+	tmnode "github.com/tendermint/tendermint/node"
+	tmproxy "github.com/tendermint/tendermint/proxy"
+	tmcore "github.com/tendermint/tendermint/rpc/core"
+	tmrpctypes "github.com/tendermint/tendermint/rpc/jsonrpc/types"
+	"github.com/tendermint/tendermint/store"
+
+	"github.com/oasisprotocol/oasis-core/go/common/identity"
+	"github.com/oasisprotocol/oasis-core/go/common/logging"
+	cmservice "github.com/oasisprotocol/oasis-core/go/common/service"
+	consensusAPI "github.com/oasisprotocol/oasis-core/go/consensus/api"
+	"github.com/oasisprotocol/oasis-core/go/consensus/api/transaction"
+	"github.com/oasisprotocol/oasis-core/go/consensus/tendermint/abci"
+	"github.com/oasisprotocol/oasis-core/go/consensus/tendermint/api"
+	tmcommon "github.com/oasisprotocol/oasis-core/go/consensus/tendermint/common"
+	"github.com/oasisprotocol/oasis-core/go/consensus/tendermint/db"
+	genesisAPI "github.com/oasisprotocol/oasis-core/go/genesis/api"
+	cmbackground "github.com/oasisprotocol/oasis-core/go/oasis-node/cmd/common/background"
+)
+
+var _ api.Backend = (*archiveService)(nil)
+
+type archiveService struct {
+	sync.Mutex
+	commonNode
+
+	abciClient abcicli.Client
+
+	isStarted bool
+
+	startedCh chan struct{}
+	quitCh    chan struct{}
+
+	stopOnce sync.Once
+}
+
+func (srv *archiveService) started() bool {
+	srv.Lock()
+	defer srv.Unlock()
+
+	return srv.isStarted
+}
+
+// Start starts the service.
+func (srv *archiveService) Start() error {
+	if srv.started() {
+		return fmt.Errorf("tendermint: service already started")
+	}
+
+	if err := srv.commonNode.Start(); err != nil {
+		return err
+	}
+
+	if err := srv.abciClient.Start(); err != nil {
+		return err
+	}
+
+	// Make sure the quit channel is closed when the node shuts down.
+	go func() {
+		select {
+		case <-srv.quitCh:
+		case <-srv.mux.Quit():
+			select {
+			case <-srv.quitCh:
+			default:
+				close(srv.quitCh)
+			}
+		}
+	}()
+
+	srv.Lock()
+	srv.isStarted = true
+	srv.Unlock()
+	close(srv.startedCh)
+
+	return nil
+}
+
+// Stop halts the service.
+func (srv *archiveService) Stop() {
+	if !srv.started() {
+		return
+	}
+	srv.stopOnce.Do(func() {
+		if err := srv.abciClient.Stop(); err != nil {
+			srv.Logger.Error("error on stopping abci client", "err", err)
+		}
+		srv.commonNode.Stop()
+	})
+}
+
+// Quit returns a channel that will be closed when the service terminates.
+func (srv *archiveService) Quit() <-chan struct{} {
+	return srv.quitCh
+}
+
+// Implements Backend.
+func (srv *archiveService) Synced() <-chan struct{} {
+	// Archive node is always considered synced.
+	ch := make(chan struct{})
+	close(ch)
+	return ch
+}
+
+// Implements Backend.
+func (srv *archiveService) EstimateGas(ctx context.Context, req *consensusAPI.EstimateGasRequest) (transaction.Gas, error) {
+	return 0, consensusAPI.ErrUnsupported
+}
+
+// Implements Backend.
+func (srv *archiveService) GetSignerNonce(ctx context.Context, req *consensusAPI.GetSignerNonceRequest) (uint64, error) {
+	return 0, consensusAPI.ErrUnsupported
+}
+
+// New creates a new archive-only consensus service.
+func NewArchive(
+	ctx context.Context,
+	dataDir string,
+	identity *identity.Identity,
+	genesisProvider genesisAPI.Provider,
+) (consensusAPI.Backend, error) {
+	var err error
+
+	srv := &archiveService{
+		commonNode: commonNode{
+			BaseBackgroundService: *cmservice.NewBaseBackgroundService("tendermint"),
+			ctx:                   ctx,
+			rpcCtx:                &tmrpctypes.Context{},
+			identity:              identity,
+			dataDir:               dataDir,
+			svcMgr:                cmbackground.NewServiceManager(logging.GetLogger("tendermint/servicemanager")),
+			startedCh:             make(chan struct{}),
+		},
+		startedCh: make(chan struct{}),
+		quitCh:    make(chan struct{}),
+	}
+	// Common node needs access to parent struct for initializing consensus services.
+	srv.commonNode.parentNode = srv
+
+	doc, err := genesisProvider.GetGenesisDocument()
+	if err != nil {
+		return nil, fmt.Errorf("tendermint/archive: failed to get genesis document: %w", err)
+	}
+	srv.genesis = doc
+
+	appConfig := &abci.ApplicationConfig{
+		DataDir:        filepath.Join(srv.dataDir, tmcommon.StateDir),
+		StorageBackend: db.GetBackendName(),
+		Pruning: abci.PruneConfig{
+			Strategy: abci.PruneNone,
+		},
+		DisableCheckpointer:       true,
+		CheckpointerCheckInterval: 100 * time.Hour, // Disabled.
+		HaltEpochHeight:           srv.genesis.HaltEpoch,
+		OwnTxSigner:               srv.identity.NodeSigner.Public(),
+		InitialHeight:             uint64(srv.genesis.Height),
+		// ReadOnly should actually be preferable for archive but there is a badger issue with read-only:
+		// https://discuss.dgraph.io/t/read-only-log-truncate-required-to-run-db/16444/2
+		ReadOnlyStorage: false,
+	}
+	srv.mux, err = abci.NewApplicationServer(srv.ctx, nil, appConfig)
+	if err != nil {
+		return nil, fmt.Errorf("tendermint/archive: failed to create application server: %w", err)
+	}
+
+	// Setup needed tendermint services.
+	logger := tmcommon.NewLogAdapter(!viper.GetBool(tmcommon.CfgLogDebug))
+	srv.abciClient = abcicli.NewLocalClient(new(tmsync.Mutex), srv.mux.Mux())
+
+	dbProvider, err := db.GetProvider()
+	if err != nil {
+		return nil, err
+	}
+	tmConfig := tmconfig.DefaultConfig()
+	_ = viper.Unmarshal(&tmConfig)
+	tmConfig.SetRoot(filepath.Join(srv.dataDir, tmcommon.StateDir))
+
+	// NOTE: DBContext uses a full tendermint config but the only thing that is actually used
+	// is the data dir field.
+	srv.blockStoreDB, err = dbProvider(&tmnode.DBContext{ID: "blockstore", Config: tmConfig})
+	if err != nil {
+		return nil, err
+	}
+
+	// NOTE: DBContext uses a full tendermint config but the only thing that is actually used
+	// is the data dir field.
+	srv.stateStore, err = dbProvider(&tmnode.DBContext{ID: "state", Config: tmConfig})
+	if err != nil {
+		return nil, err
+	}
+
+	tmGenDoc, err := api.GetTendermintGenesisDocument(genesisProvider)
+	if err != nil {
+		return nil, err
+	}
+
+	// Setup minimal tendermint environment needed to support consensus queries.
+	tmcore.SetEnvironment(&tmcore.Environment{
+		ProxyAppQuery:    tmproxy.NewAppConnQuery(srv.abciClient),
+		ProxyAppMempool:  nil,
+		StateDB:          srv.stateStore,
+		BlockStore:       store.NewBlockStore(srv.blockStoreDB),
+		EvidencePool:     nil,
+		ConsensusState:   nil,
+		GenDoc:           tmGenDoc,
+		Logger:           logger,
+		Config:           *tmConfig.RPC,
+		EventBus:         nil,
+		P2PPeers:         nil,
+		P2PTransport:     nil,
+		PubKey:           nil,
+		TxIndexer:        nil,
+		ConsensusReactor: nil,
+		Mempool:          nil,
+	})
+
+	return srv, srv.initialize()
+}