mirror of
https://github.com/tendermint/tendermint.git
synced 2026-09-10 10:06:20 +00:00
proxy: collapse triforcated client
This commit is contained in:
tycho garen
+37
-174
@@ -10,233 +10,96 @@ import (
|
||||
"github.com/tendermint/tendermint/abci/types"
|
||||
)
|
||||
|
||||
//go:generate ../../scripts/mockery_generate.sh AppConnConsensus|AppConnMempool|AppConnQuery|AppConnSnapshot
|
||||
|
||||
//----------------------------------------------------------------------------------------
|
||||
// Enforce which abci msgs can be sent on a connection at the type level
|
||||
|
||||
type AppConnConsensus interface {
|
||||
Error() error
|
||||
|
||||
InitChain(context.Context, types.RequestInitChain) (*types.ResponseInitChain, error)
|
||||
|
||||
PrepareProposal(context.Context, types.RequestPrepareProposal) (*types.ResponsePrepareProposal, error)
|
||||
ProcessProposal(context.Context, types.RequestProcessProposal) (*types.ResponseProcessProposal, error)
|
||||
ExtendVote(context.Context, types.RequestExtendVote) (*types.ResponseExtendVote, error)
|
||||
VerifyVoteExtension(context.Context, types.RequestVerifyVoteExtension) (*types.ResponseVerifyVoteExtension, error)
|
||||
FinalizeBlock(context.Context, types.RequestFinalizeBlock) (*types.ResponseFinalizeBlock, error)
|
||||
Commit(context.Context) (*types.ResponseCommit, error)
|
||||
}
|
||||
|
||||
type AppConnMempool interface {
|
||||
Error() error
|
||||
|
||||
CheckTx(context.Context, types.RequestCheckTx) (*types.ResponseCheckTx, error)
|
||||
|
||||
Flush(context.Context) error
|
||||
}
|
||||
|
||||
type AppConnQuery interface {
|
||||
Error() error
|
||||
|
||||
Echo(context.Context, string) (*types.ResponseEcho, error)
|
||||
Info(context.Context, types.RequestInfo) (*types.ResponseInfo, error)
|
||||
Query(context.Context, types.RequestQuery) (*types.ResponseQuery, error)
|
||||
}
|
||||
|
||||
type AppConnSnapshot interface {
|
||||
Error() error
|
||||
|
||||
ListSnapshots(context.Context, types.RequestListSnapshots) (*types.ResponseListSnapshots, error)
|
||||
OfferSnapshot(context.Context, types.RequestOfferSnapshot) (*types.ResponseOfferSnapshot, error)
|
||||
LoadSnapshotChunk(context.Context, types.RequestLoadSnapshotChunk) (*types.ResponseLoadSnapshotChunk, error)
|
||||
ApplySnapshotChunk(context.Context, types.RequestApplySnapshotChunk) (*types.ResponseApplySnapshotChunk, error)
|
||||
}
|
||||
|
||||
//-----------------------------------------------------------------------------------------
|
||||
// Implements AppConnConsensus (subset of abciclient.Client)
|
||||
|
||||
type appConnConsensus struct {
|
||||
type proxyClient struct {
|
||||
metrics *Metrics
|
||||
appConn abciclient.Client
|
||||
abciclient.Client
|
||||
}
|
||||
|
||||
var _ AppConnConsensus = (*appConnConsensus)(nil)
|
||||
|
||||
func NewAppConnConsensus(appConn abciclient.Client, metrics *Metrics) AppConnConsensus {
|
||||
return &appConnConsensus{
|
||||
func newProxyClient(appConn abciclient.Client, metrics *Metrics) abciclient.Client {
|
||||
return &proxyClient{
|
||||
metrics: metrics,
|
||||
appConn: appConn,
|
||||
Client: appConn,
|
||||
}
|
||||
}
|
||||
|
||||
func (app *appConnConsensus) Error() error {
|
||||
return app.appConn.Error()
|
||||
}
|
||||
|
||||
func (app *appConnConsensus) InitChain(
|
||||
ctx context.Context,
|
||||
req types.RequestInitChain,
|
||||
) (*types.ResponseInitChain, error) {
|
||||
func (app *proxyClient) InitChain(ctx context.Context, req types.RequestInitChain) (*types.ResponseInitChain, error) {
|
||||
defer addTimeSample(app.metrics.MethodTiming.With("method", "init_chain", "type", "sync"))()
|
||||
return app.appConn.InitChain(ctx, req)
|
||||
return app.Client.InitChain(ctx, req)
|
||||
}
|
||||
|
||||
func (app *appConnConsensus) PrepareProposal(
|
||||
ctx context.Context,
|
||||
req types.RequestPrepareProposal,
|
||||
) (*types.ResponsePrepareProposal, error) {
|
||||
func (app *proxyClient) PrepareProposal(ctx context.Context, req types.RequestPrepareProposal) (*types.ResponsePrepareProposal, error) {
|
||||
defer addTimeSample(app.metrics.MethodTiming.With("method", "prepare_proposal", "type", "sync"))()
|
||||
return app.appConn.PrepareProposal(ctx, req)
|
||||
return app.Client.PrepareProposal(ctx, req)
|
||||
}
|
||||
|
||||
func (app *appConnConsensus) ProcessProposal(
|
||||
ctx context.Context,
|
||||
req types.RequestProcessProposal,
|
||||
) (*types.ResponseProcessProposal, error) {
|
||||
func (app *proxyClient) ProcessProposal(ctx context.Context, req types.RequestProcessProposal) (*types.ResponseProcessProposal, error) {
|
||||
defer addTimeSample(app.metrics.MethodTiming.With("method", "process_proposal", "type", "sync"))()
|
||||
return app.appConn.ProcessProposal(ctx, req)
|
||||
return app.Client.ProcessProposal(ctx, req)
|
||||
}
|
||||
|
||||
func (app *appConnConsensus) ExtendVote(
|
||||
ctx context.Context,
|
||||
req types.RequestExtendVote,
|
||||
) (*types.ResponseExtendVote, error) {
|
||||
func (app *proxyClient) ExtendVote(ctx context.Context, req types.RequestExtendVote) (*types.ResponseExtendVote, error) {
|
||||
defer addTimeSample(app.metrics.MethodTiming.With("method", "extend_vote", "type", "sync"))()
|
||||
return app.appConn.ExtendVote(ctx, req)
|
||||
return app.Client.ExtendVote(ctx, req)
|
||||
}
|
||||
|
||||
func (app *appConnConsensus) VerifyVoteExtension(
|
||||
ctx context.Context,
|
||||
req types.RequestVerifyVoteExtension,
|
||||
) (*types.ResponseVerifyVoteExtension, error) {
|
||||
func (app *proxyClient) VerifyVoteExtension(ctx context.Context, req types.RequestVerifyVoteExtension) (*types.ResponseVerifyVoteExtension, error) {
|
||||
defer addTimeSample(app.metrics.MethodTiming.With("method", "verify_vote_extension", "type", "sync"))()
|
||||
return app.appConn.VerifyVoteExtension(ctx, req)
|
||||
return app.Client.VerifyVoteExtension(ctx, req)
|
||||
}
|
||||
|
||||
func (app *appConnConsensus) FinalizeBlock(
|
||||
ctx context.Context,
|
||||
req types.RequestFinalizeBlock,
|
||||
) (*types.ResponseFinalizeBlock, error) {
|
||||
func (app *proxyClient) FinalizeBlock(ctx context.Context, req types.RequestFinalizeBlock) (*types.ResponseFinalizeBlock, error) {
|
||||
defer addTimeSample(app.metrics.MethodTiming.With("method", "finalize_block", "type", "sync"))()
|
||||
return app.appConn.FinalizeBlock(ctx, req)
|
||||
return app.Client.FinalizeBlock(ctx, req)
|
||||
}
|
||||
|
||||
func (app *appConnConsensus) Commit(ctx context.Context) (*types.ResponseCommit, error) {
|
||||
func (app *proxyClient) Commit(ctx context.Context) (*types.ResponseCommit, error) {
|
||||
defer addTimeSample(app.metrics.MethodTiming.With("method", "commit", "type", "sync"))()
|
||||
return app.appConn.Commit(ctx)
|
||||
return app.Client.Commit(ctx)
|
||||
}
|
||||
|
||||
//------------------------------------------------
|
||||
// Implements AppConnMempool (subset of abciclient.Client)
|
||||
|
||||
type appConnMempool struct {
|
||||
metrics *Metrics
|
||||
appConn abciclient.Client
|
||||
}
|
||||
|
||||
func NewAppConnMempool(appConn abciclient.Client, metrics *Metrics) AppConnMempool {
|
||||
return &appConnMempool{
|
||||
metrics: metrics,
|
||||
appConn: appConn,
|
||||
}
|
||||
}
|
||||
|
||||
func (app *appConnMempool) Error() error {
|
||||
return app.appConn.Error()
|
||||
}
|
||||
|
||||
func (app *appConnMempool) Flush(ctx context.Context) error {
|
||||
func (app *proxyClient) Flush(ctx context.Context) error {
|
||||
defer addTimeSample(app.metrics.MethodTiming.With("method", "flush", "type", "sync"))()
|
||||
return app.appConn.Flush(ctx)
|
||||
return app.Client.Flush(ctx)
|
||||
}
|
||||
|
||||
func (app *appConnMempool) CheckTx(ctx context.Context, req types.RequestCheckTx) (*types.ResponseCheckTx, error) {
|
||||
func (app *proxyClient) CheckTx(ctx context.Context, req types.RequestCheckTx) (*types.ResponseCheckTx, error) {
|
||||
defer addTimeSample(app.metrics.MethodTiming.With("method", "check_tx", "type", "sync"))()
|
||||
return app.appConn.CheckTx(ctx, req)
|
||||
return app.Client.CheckTx(ctx, req)
|
||||
}
|
||||
|
||||
//------------------------------------------------
|
||||
// Implements AppConnQuery (subset of abciclient.Client)
|
||||
|
||||
type appConnQuery struct {
|
||||
metrics *Metrics
|
||||
appConn abciclient.Client
|
||||
}
|
||||
|
||||
func NewAppConnQuery(appConn abciclient.Client, metrics *Metrics) AppConnQuery {
|
||||
return &appConnQuery{
|
||||
metrics: metrics,
|
||||
appConn: appConn,
|
||||
}
|
||||
}
|
||||
|
||||
func (app *appConnQuery) Error() error {
|
||||
return app.appConn.Error()
|
||||
}
|
||||
|
||||
func (app *appConnQuery) Echo(ctx context.Context, msg string) (*types.ResponseEcho, error) {
|
||||
func (app *proxyClient) Echo(ctx context.Context, msg string) (*types.ResponseEcho, error) {
|
||||
defer addTimeSample(app.metrics.MethodTiming.With("method", "echo", "type", "sync"))()
|
||||
return app.appConn.Echo(ctx, msg)
|
||||
return app.Client.Echo(ctx, msg)
|
||||
}
|
||||
|
||||
func (app *appConnQuery) Info(ctx context.Context, req types.RequestInfo) (*types.ResponseInfo, error) {
|
||||
func (app *proxyClient) Info(ctx context.Context, req types.RequestInfo) (*types.ResponseInfo, error) {
|
||||
defer addTimeSample(app.metrics.MethodTiming.With("method", "info", "type", "sync"))()
|
||||
return app.appConn.Info(ctx, req)
|
||||
return app.Client.Info(ctx, req)
|
||||
}
|
||||
|
||||
func (app *appConnQuery) Query(ctx context.Context, reqQuery types.RequestQuery) (*types.ResponseQuery, error) {
|
||||
func (app *proxyClient) Query(ctx context.Context, reqQuery types.RequestQuery) (*types.ResponseQuery, error) {
|
||||
defer addTimeSample(app.metrics.MethodTiming.With("method", "query", "type", "sync"))()
|
||||
return app.appConn.Query(ctx, reqQuery)
|
||||
return app.Client.Query(ctx, reqQuery)
|
||||
}
|
||||
|
||||
//------------------------------------------------
|
||||
// Implements AppConnSnapshot (subset of abciclient.Client)
|
||||
|
||||
type appConnSnapshot struct {
|
||||
metrics *Metrics
|
||||
appConn abciclient.Client
|
||||
}
|
||||
|
||||
func NewAppConnSnapshot(appConn abciclient.Client, metrics *Metrics) AppConnSnapshot {
|
||||
return &appConnSnapshot{
|
||||
metrics: metrics,
|
||||
appConn: appConn,
|
||||
}
|
||||
}
|
||||
|
||||
func (app *appConnSnapshot) Error() error {
|
||||
return app.appConn.Error()
|
||||
}
|
||||
|
||||
func (app *appConnSnapshot) ListSnapshots(
|
||||
ctx context.Context,
|
||||
req types.RequestListSnapshots,
|
||||
) (*types.ResponseListSnapshots, error) {
|
||||
func (app *proxyClient) ListSnapshots(ctx context.Context, req types.RequestListSnapshots) (*types.ResponseListSnapshots, error) {
|
||||
defer addTimeSample(app.metrics.MethodTiming.With("method", "list_snapshots", "type", "sync"))()
|
||||
return app.appConn.ListSnapshots(ctx, req)
|
||||
return app.Client.ListSnapshots(ctx, req)
|
||||
}
|
||||
|
||||
func (app *appConnSnapshot) OfferSnapshot(
|
||||
ctx context.Context,
|
||||
req types.RequestOfferSnapshot,
|
||||
) (*types.ResponseOfferSnapshot, error) {
|
||||
func (app *proxyClient) OfferSnapshot(ctx context.Context, req types.RequestOfferSnapshot) (*types.ResponseOfferSnapshot, error) {
|
||||
defer addTimeSample(app.metrics.MethodTiming.With("method", "offer_snapshot", "type", "sync"))()
|
||||
return app.appConn.OfferSnapshot(ctx, req)
|
||||
return app.Client.OfferSnapshot(ctx, req)
|
||||
}
|
||||
|
||||
func (app *appConnSnapshot) LoadSnapshotChunk(
|
||||
ctx context.Context,
|
||||
req types.RequestLoadSnapshotChunk) (*types.ResponseLoadSnapshotChunk, error) {
|
||||
func (app *proxyClient) LoadSnapshotChunk(ctx context.Context, req types.RequestLoadSnapshotChunk) (*types.ResponseLoadSnapshotChunk, error) {
|
||||
defer addTimeSample(app.metrics.MethodTiming.With("method", "load_snapshot_chunk", "type", "sync"))()
|
||||
return app.appConn.LoadSnapshotChunk(ctx, req)
|
||||
return app.Client.LoadSnapshotChunk(ctx, req)
|
||||
}
|
||||
|
||||
func (app *appConnSnapshot) ApplySnapshotChunk(
|
||||
ctx context.Context,
|
||||
req types.RequestApplySnapshotChunk) (*types.ResponseApplySnapshotChunk, error) {
|
||||
func (app *proxyClient) ApplySnapshotChunk(ctx context.Context, req types.RequestApplySnapshotChunk) (*types.ResponseApplySnapshotChunk, error) {
|
||||
defer addTimeSample(app.metrics.MethodTiming.With("method", "apply_snapshot_chunk", "type", "sync"))()
|
||||
return app.appConn.ApplySnapshotChunk(ctx, req)
|
||||
return app.Client.ApplySnapshotChunk(ctx, req)
|
||||
}
|
||||
|
||||
// addTimeSample returns a function that, when called, adds an observation to m.
|
||||
|
||||
@@ -10,57 +10,9 @@ import (
|
||||
"github.com/tendermint/tendermint/libs/service"
|
||||
)
|
||||
|
||||
// AppConns is the Tendermint's interface to the application that consists of
|
||||
// multiple connections.
|
||||
type AppConns interface {
|
||||
service.Service
|
||||
|
||||
// Mempool connection
|
||||
Mempool() AppConnMempool
|
||||
// Consensus connection
|
||||
Consensus() AppConnConsensus
|
||||
// Query connection
|
||||
Query() AppConnQuery
|
||||
// Snapshot connection
|
||||
Snapshot() AppConnSnapshot
|
||||
}
|
||||
|
||||
// NewAppConns calls NewMultiAppConn.
|
||||
func NewAppConns(clientCreator abciclient.Creator, logger log.Logger, metrics *Metrics) AppConns {
|
||||
return NewMultiAppConn(clientCreator, logger, metrics)
|
||||
}
|
||||
|
||||
// multiAppConn implements AppConns.
|
||||
//
|
||||
// A multiAppConn is made of a few appConns and manages their underlying abci
|
||||
// clients.
|
||||
// TODO: on app restart, clients must reboot together
|
||||
type multiAppConn struct {
|
||||
service.BaseService
|
||||
logger log.Logger
|
||||
|
||||
metrics *Metrics
|
||||
consensusConn AppConnConsensus
|
||||
mempoolConn AppConnMempool
|
||||
queryConn AppConnQuery
|
||||
snapshotConn AppConnSnapshot
|
||||
|
||||
client stoppableClient
|
||||
|
||||
clientCreator abciclient.Creator
|
||||
}
|
||||
|
||||
// TODO: this is a totally internal and quasi permanent shim for
|
||||
// clients. eventually we can have a single client and have some kind
|
||||
// of reasonable lifecycle witout needing an explicit stop method.
|
||||
type stoppableClient interface {
|
||||
abciclient.Client
|
||||
Stop()
|
||||
}
|
||||
|
||||
// NewMultiAppConn makes all necessary abci connections to the application.
|
||||
func NewMultiAppConn(clientCreator abciclient.Creator, logger log.Logger, metrics *Metrics) AppConns {
|
||||
multiAppConn := &multiAppConn{
|
||||
// New creates a proxy application interface.
|
||||
func New(clientCreator abciclient.Creator, logger log.Logger, metrics *Metrics) abciclient.Client {
|
||||
multiAppConn := &proxyConn{
|
||||
logger: logger,
|
||||
metrics: metrics,
|
||||
clientCreator: clientCreator,
|
||||
@@ -69,16 +21,38 @@ func NewMultiAppConn(clientCreator abciclient.Creator, logger log.Logger, metric
|
||||
return multiAppConn
|
||||
}
|
||||
|
||||
func (app *multiAppConn) Mempool() AppConnMempool { return app.mempoolConn }
|
||||
func (app *multiAppConn) Consensus() AppConnConsensus { return app.consensusConn }
|
||||
func (app *multiAppConn) Query() AppConnQuery { return app.queryConn }
|
||||
func (app *multiAppConn) Snapshot() AppConnSnapshot { return app.snapshotConn }
|
||||
// proxyConn implements provides the application connection.
|
||||
type proxyConn struct {
|
||||
service.BaseService
|
||||
abciclient.Client
|
||||
|
||||
func (app *multiAppConn) OnStart(ctx context.Context) error {
|
||||
logger log.Logger
|
||||
|
||||
metrics *Metrics
|
||||
|
||||
clientCreator abciclient.Creator
|
||||
}
|
||||
|
||||
func (app *proxyConn) OnStop() { tryCallStop(app.Client) }
|
||||
func (app *proxyConn) IsRunning() bool { return app.Client.IsRunning() }
|
||||
func (app *proxyConn) Start(ctx context.Context) error { return app.BaseService.Start(ctx) }
|
||||
func (app *proxyConn) Wait() { app.BaseService.Wait() }
|
||||
|
||||
func tryCallStop(client abciclient.Client) {
|
||||
if client == nil {
|
||||
return
|
||||
}
|
||||
|
||||
if sc, ok := client.(interface{ Stop() }); ok {
|
||||
sc.Stop()
|
||||
}
|
||||
}
|
||||
|
||||
func (app *proxyConn) OnStart(ctx context.Context) error {
|
||||
var err error
|
||||
defer func() {
|
||||
if err != nil {
|
||||
app.client.Stop()
|
||||
tryCallStop(app.Client)
|
||||
}
|
||||
}()
|
||||
|
||||
@@ -88,24 +62,18 @@ func (app *multiAppConn) OnStart(ctx context.Context) error {
|
||||
return err
|
||||
}
|
||||
|
||||
app.queryConn = NewAppConnQuery(client, app.metrics)
|
||||
app.snapshotConn = NewAppConnSnapshot(client, app.metrics)
|
||||
app.mempoolConn = NewAppConnMempool(client, app.metrics)
|
||||
app.consensusConn = NewAppConnConsensus(client, app.metrics)
|
||||
|
||||
app.client = client.(stoppableClient)
|
||||
|
||||
app.Client = newProxyClient(client, app.metrics)
|
||||
// Kill Tendermint if the ABCI application crashes.
|
||||
go func() {
|
||||
if !client.IsRunning() {
|
||||
if !app.Client.IsRunning() {
|
||||
return
|
||||
}
|
||||
app.client.Wait()
|
||||
app.Client.Wait()
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
|
||||
if err := app.client.Error(); err != nil {
|
||||
if err := app.Client.Error(); err != nil {
|
||||
app.logger.Error("client connection terminated. Did the application crash? Please restart tendermint",
|
||||
"err", err)
|
||||
if killErr := kill(); killErr != nil {
|
||||
@@ -119,8 +87,6 @@ func (app *multiAppConn) OnStart(ctx context.Context) error {
|
||||
return client.Start(ctx)
|
||||
}
|
||||
|
||||
func (app *multiAppConn) OnStop() { app.client.Stop() }
|
||||
|
||||
func kill() error {
|
||||
p, err := os.FindProcess(os.Getpid())
|
||||
if err != nil {
|
||||
|
||||
@@ -42,7 +42,7 @@ func TestAppConns_Start_Stop(t *testing.T) {
|
||||
return cl, nil
|
||||
}
|
||||
|
||||
appConns := NewAppConns(creator, log.TestingLogger(), NopMetrics())
|
||||
appConns := New(creator, log.TestingLogger(), NopMetrics())
|
||||
|
||||
err := appConns.Start(ctx)
|
||||
require.NoError(t, err)
|
||||
@@ -84,7 +84,7 @@ func TestAppConns_Failure(t *testing.T) {
|
||||
return cl, nil
|
||||
}
|
||||
|
||||
appConns := NewAppConns(creator, log.TestingLogger(), NopMetrics())
|
||||
appConns := New(creator, log.TestingLogger(), NopMetrics())
|
||||
|
||||
err := appConns.Start(ctx)
|
||||
require.NoError(t, err)
|
||||
|
||||
Reference in New Issue
Block a user