proxy: move proxy package to internal (#6953)

internal/blocksync/v0/reactor_test.go

@@ -7,16 +7,17 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	abciclient "github.com/tendermint/tendermint/abci/client"
 	abci "github.com/tendermint/tendermint/abci/types"
 	cfg "github.com/tendermint/tendermint/config"
 	cons "github.com/tendermint/tendermint/internal/consensus"
 	"github.com/tendermint/tendermint/internal/mempool/mock"
 	"github.com/tendermint/tendermint/internal/p2p"
 	"github.com/tendermint/tendermint/internal/p2p/p2ptest"
+	"github.com/tendermint/tendermint/internal/proxy"
 	"github.com/tendermint/tendermint/internal/test/factory"
 	"github.com/tendermint/tendermint/libs/log"
 	bcproto "github.com/tendermint/tendermint/proto/tendermint/blocksync"
-	"github.com/tendermint/tendermint/proxy"
 	sm "github.com/tendermint/tendermint/state"
 	sf "github.com/tendermint/tendermint/state/test/factory"
 	"github.com/tendermint/tendermint/store"
@@ -97,7 +98,7 @@ func (rts *reactorTestSuite) addNode(t *testing.T,
 	t.Helper()
 
 	rts.nodes = append(rts.nodes, nodeID)
-	rts.app[nodeID] = proxy.NewAppConns(proxy.NewLocalClientCreator(&abci.BaseApplication{}))
+	rts.app[nodeID] = proxy.NewAppConns(abciclient.NewLocalCreator(&abci.BaseApplication{}))
 	require.NoError(t, rts.app[nodeID].Start())
 
 	blockDB := dbm.NewMemDB()

internal/blocksync/v2/reactor_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/stretchr/testify/require"
 	dbm "github.com/tendermint/tm-db"
 
+	abciclient "github.com/tendermint/tendermint/abci/client"
 	abci "github.com/tendermint/tendermint/abci/types"
 	cfg "github.com/tendermint/tendermint/config"
 	"github.com/tendermint/tendermint/internal/blocksync/v2/internal/behavior"
@@ -20,11 +21,11 @@ import (
 	"github.com/tendermint/tendermint/internal/mempool/mock"
 	"github.com/tendermint/tendermint/internal/p2p"
 	"github.com/tendermint/tendermint/internal/p2p/conn"
+	"github.com/tendermint/tendermint/internal/proxy"
 	"github.com/tendermint/tendermint/internal/test/factory"
 	"github.com/tendermint/tendermint/libs/log"
 	"github.com/tendermint/tendermint/libs/service"
 	bcproto "github.com/tendermint/tendermint/proto/tendermint/blocksync"
-	"github.com/tendermint/tendermint/proxy"
 	sm "github.com/tendermint/tendermint/state"
 	sf "github.com/tendermint/tendermint/state/test/factory"
 	tmstore "github.com/tendermint/tendermint/store"
@@ -163,7 +164,7 @@ func newTestReactor(t *testing.T, p testReactorParams) *BlockchainReactor {
 		appl = &mockBlockApplier{}
 	} else {
 		app := &testApp{}
-		cc := proxy.NewLocalClientCreator(app)
+		cc := abciclient.NewLocalCreator(app)
 		proxyApp := proxy.NewAppConns(cc)
 		err := proxyApp.Start()
 		require.NoError(t, err)
@@ -187,176 +188,176 @@ func newTestReactor(t *testing.T, p testReactorParams) *BlockchainReactor {
 // future improvement in [#4482](https://github.com/tendermint/tendermint/issues/4482).
 // func TestReactorTerminationScenarios(t *testing.T) {
 
-// 	config := cfg.ResetTestRoot("blockchain_reactor_v2_test")
-// 	defer os.RemoveAll(config.RootDir)
-// 	genDoc, privVals := randGenesisDoc(config.ChainID(), 1, false, 30)
-// 	refStore, _, _ := newReactorStore(genDoc, privVals, 20)
+//	config := cfg.ResetTestRoot("blockchain_reactor_v2_test")
+//	defer os.RemoveAll(config.RootDir)
+//	genDoc, privVals := randGenesisDoc(config.ChainID(), 1, false, 30)
+//	refStore, _, _ := newReactorStore(genDoc, privVals, 20)
 
-// 	params := testReactorParams{
-// 		logger:      log.TestingLogger(),
-// 		genDoc:      genDoc,
-// 		privVals:    privVals,
-// 		startHeight: 10,
-// 		bufferSize:  100,
-// 		mockA:       true,
-// 	}
+//	params := testReactorParams{
+//		logger:      log.TestingLogger(),
+//		genDoc:      genDoc,
+//		privVals:    privVals,
+//		startHeight: 10,
+//		bufferSize:  100,
+//		mockA:       true,
+//	}
 
-// 	type testEvent struct {
-// 		evType string
-// 		peer   string
-// 		height int64
-// 	}
+//	type testEvent struct {
+//		evType string
+//		peer   string
+//		height int64
+//	}
 
-// 	tests := []struct {
-// 		name   string
-// 		params testReactorParams
-// 		msgs   []testEvent
-// 	}{
-// 		{
-// 			name:   "simple termination on max peer height - one peer",
-// 			params: params,
-// 			msgs: []testEvent{
-// 				{evType: "AddPeer", peer: "P1"},
-// 				{evType: "ReceiveS", peer: "P1", height: 13},
-// 				{evType: "BlockReq"},
-// 				{evType: "ReceiveB", peer: "P1", height: 11},
-// 				{evType: "BlockReq"},
-// 				{evType: "BlockReq"},
-// 				{evType: "ReceiveB", peer: "P1", height: 12},
-// 				{evType: "Process"},
-// 				{evType: "ReceiveB", peer: "P1", height: 13},
-// 				{evType: "Process"},
-// 			},
-// 		},
-// 		{
-// 			name:   "simple termination on max peer height - two peers",
-// 			params: params,
-// 			msgs: []testEvent{
-// 				{evType: "AddPeer", peer: "P1"},
-// 				{evType: "AddPeer", peer: "P2"},
-// 				{evType: "ReceiveS", peer: "P1", height: 13},
-// 				{evType: "ReceiveS", peer: "P2", height: 15},
-// 				{evType: "BlockReq"},
-// 				{evType: "BlockReq"},
-// 				{evType: "ReceiveB", peer: "P1", height: 11},
-// 				{evType: "ReceiveB", peer: "P2", height: 12},
-// 				{evType: "Process"},
-// 				{evType: "BlockReq"},
-// 				{evType: "BlockReq"},
-// 				{evType: "ReceiveB", peer: "P1", height: 13},
-// 				{evType: "Process"},
-// 				{evType: "ReceiveB", peer: "P2", height: 14},
-// 				{evType: "Process"},
-// 				{evType: "BlockReq"},
-// 				{evType: "ReceiveB", peer: "P2", height: 15},
-// 				{evType: "Process"},
-// 			},
-// 		},
-// 		{
-// 			name:   "termination on max peer height - two peers, noBlock error",
-// 			params: params,
-// 			msgs: []testEvent{
-// 				{evType: "AddPeer", peer: "P1"},
-// 				{evType: "AddPeer", peer: "P2"},
-// 				{evType: "ReceiveS", peer: "P1", height: 13},
-// 				{evType: "ReceiveS", peer: "P2", height: 15},
-// 				{evType: "BlockReq"},
-// 				{evType: "BlockReq"},
-// 				{evType: "ReceiveNB", peer: "P1", height: 11},
-// 				{evType: "BlockReq"},
-// 				{evType: "ReceiveB", peer: "P2", height: 12},
-// 				{evType: "ReceiveB", peer: "P2", height: 11},
-// 				{evType: "Process"},
-// 				{evType: "BlockReq"},
-// 				{evType: "BlockReq"},
-// 				{evType: "ReceiveB", peer: "P2", height: 13},
-// 				{evType: "Process"},
-// 				{evType: "ReceiveB", peer: "P2", height: 14},
-// 				{evType: "Process"},
-// 				{evType: "BlockReq"},
-// 				{evType: "ReceiveB", peer: "P2", height: 15},
-// 				{evType: "Process"},
-// 			},
-// 		},
-// 		{
-// 			name:   "termination on max peer height - two peers, remove one peer",
-// 			params: params,
-// 			msgs: []testEvent{
-// 				{evType: "AddPeer", peer: "P1"},
-// 				{evType: "AddPeer", peer: "P2"},
-// 				{evType: "ReceiveS", peer: "P1", height: 13},
-// 				{evType: "ReceiveS", peer: "P2", height: 15},
-// 				{evType: "BlockReq"},
-// 				{evType: "BlockReq"},
-// 				{evType: "RemovePeer", peer: "P1"},
-// 				{evType: "BlockReq"},
-// 				{evType: "ReceiveB", peer: "P2", height: 12},
-// 				{evType: "ReceiveB", peer: "P2", height: 11},
-// 				{evType: "Process"},
-// 				{evType: "BlockReq"},
-// 				{evType: "BlockReq"},
-// 				{evType: "ReceiveB", peer: "P2", height: 13},
-// 				{evType: "Process"},
-// 				{evType: "ReceiveB", peer: "P2", height: 14},
-// 				{evType: "Process"},
-// 				{evType: "BlockReq"},
-// 				{evType: "ReceiveB", peer: "P2", height: 15},
-// 				{evType: "Process"},
-// 			},
-// 		},
-// 	}
+//	tests := []struct {
+//		name   string
+//		params testReactorParams
+//		msgs   []testEvent
+//	}{
+//		{
+//			name:   "simple termination on max peer height - one peer",
+//			params: params,
+//			msgs: []testEvent{
+//				{evType: "AddPeer", peer: "P1"},
+//				{evType: "ReceiveS", peer: "P1", height: 13},
+//				{evType: "BlockReq"},
+//				{evType: "ReceiveB", peer: "P1", height: 11},
+//				{evType: "BlockReq"},
+//				{evType: "BlockReq"},
+//				{evType: "ReceiveB", peer: "P1", height: 12},
+//				{evType: "Process"},
+//				{evType: "ReceiveB", peer: "P1", height: 13},
+//				{evType: "Process"},
+//			},
+//		},
+//		{
+//			name:   "simple termination on max peer height - two peers",
+//			params: params,
+//			msgs: []testEvent{
+//				{evType: "AddPeer", peer: "P1"},
+//				{evType: "AddPeer", peer: "P2"},
+//				{evType: "ReceiveS", peer: "P1", height: 13},
+//				{evType: "ReceiveS", peer: "P2", height: 15},
+//				{evType: "BlockReq"},
+//				{evType: "BlockReq"},
+//				{evType: "ReceiveB", peer: "P1", height: 11},
+//				{evType: "ReceiveB", peer: "P2", height: 12},
+//				{evType: "Process"},
+//				{evType: "BlockReq"},
+//				{evType: "BlockReq"},
+//				{evType: "ReceiveB", peer: "P1", height: 13},
+//				{evType: "Process"},
+//				{evType: "ReceiveB", peer: "P2", height: 14},
+//				{evType: "Process"},
+//				{evType: "BlockReq"},
+//				{evType: "ReceiveB", peer: "P2", height: 15},
+//				{evType: "Process"},
+//			},
+//		},
+//		{
+//			name:   "termination on max peer height - two peers, noBlock error",
+//			params: params,
+//			msgs: []testEvent{
+//				{evType: "AddPeer", peer: "P1"},
+//				{evType: "AddPeer", peer: "P2"},
+//				{evType: "ReceiveS", peer: "P1", height: 13},
+//				{evType: "ReceiveS", peer: "P2", height: 15},
+//				{evType: "BlockReq"},
+//				{evType: "BlockReq"},
+//				{evType: "ReceiveNB", peer: "P1", height: 11},
+//				{evType: "BlockReq"},
+//				{evType: "ReceiveB", peer: "P2", height: 12},
+//				{evType: "ReceiveB", peer: "P2", height: 11},
+//				{evType: "Process"},
+//				{evType: "BlockReq"},
+//				{evType: "BlockReq"},
+//				{evType: "ReceiveB", peer: "P2", height: 13},
+//				{evType: "Process"},
+//				{evType: "ReceiveB", peer: "P2", height: 14},
+//				{evType: "Process"},
+//				{evType: "BlockReq"},
+//				{evType: "ReceiveB", peer: "P2", height: 15},
+//				{evType: "Process"},
+//			},
+//		},
+//		{
+//			name:   "termination on max peer height - two peers, remove one peer",
+//			params: params,
+//			msgs: []testEvent{
+//				{evType: "AddPeer", peer: "P1"},
+//				{evType: "AddPeer", peer: "P2"},
+//				{evType: "ReceiveS", peer: "P1", height: 13},
+//				{evType: "ReceiveS", peer: "P2", height: 15},
+//				{evType: "BlockReq"},
+//				{evType: "BlockReq"},
+//				{evType: "RemovePeer", peer: "P1"},
+//				{evType: "BlockReq"},
+//				{evType: "ReceiveB", peer: "P2", height: 12},
+//				{evType: "ReceiveB", peer: "P2", height: 11},
+//				{evType: "Process"},
+//				{evType: "BlockReq"},
+//				{evType: "BlockReq"},
+//				{evType: "ReceiveB", peer: "P2", height: 13},
+//				{evType: "Process"},
+//				{evType: "ReceiveB", peer: "P2", height: 14},
+//				{evType: "Process"},
+//				{evType: "BlockReq"},
+//				{evType: "ReceiveB", peer: "P2", height: 15},
+//				{evType: "Process"},
+//			},
+//		},
+//	}
 
-// 	for _, tt := range tests {
-// 		tt := tt
-// 		t.Run(tt.name, func(t *testing.T) {
-// 			reactor := newTestReactor(params)
-// 			reactor.Start()
-// 			reactor.reporter = behavior.NewMockReporter()
-// 			mockSwitch := &mockSwitchIo{switchedToConsensus: false}
-// 			reactor.io = mockSwitch
-// 			// time for go routines to start
-// 			time.Sleep(time.Millisecond)
+//	for _, tt := range tests {
+//		tt := tt
+//		t.Run(tt.name, func(t *testing.T) {
+//			reactor := newTestReactor(params)
+//			reactor.Start()
+//			reactor.reporter = behavior.NewMockReporter()
+//			mockSwitch := &mockSwitchIo{switchedToConsensus: false}
+//			reactor.io = mockSwitch
+//			// time for go routines to start
+//			time.Sleep(time.Millisecond)
 
-// 			for _, step := range tt.msgs {
-// 				switch step.evType {
-// 				case "AddPeer":
-// 					reactor.scheduler.send(bcAddNewPeer{peerID: p2p.ID(step.peer)})
-// 				case "RemovePeer":
-// 					reactor.scheduler.send(bcRemovePeer{peerID: p2p.ID(step.peer)})
-// 				case "ReceiveS":
-// 					reactor.scheduler.send(bcStatusResponse{
-// 						peerID: p2p.ID(step.peer),
-// 						height: step.height,
-// 						time:   time.Now(),
-// 					})
-// 				case "ReceiveB":
-// 					reactor.scheduler.send(bcBlockResponse{
-// 						peerID: p2p.ID(step.peer),
-// 						block:  refStore.LoadBlock(step.height),
-// 						size:   10,
-// 						time:   time.Now(),
-// 					})
-// 				case "ReceiveNB":
-// 					reactor.scheduler.send(bcNoBlockResponse{
-// 						peerID: p2p.ID(step.peer),
-// 						height: step.height,
-// 						time:   time.Now(),
-// 					})
-// 				case "BlockReq":
-// 					reactor.scheduler.send(rTrySchedule{time: time.Now()})
-// 				case "Process":
-// 					reactor.processor.send(rProcessBlock{})
-// 				}
-// 				// give time for messages to propagate between routines
-// 				time.Sleep(time.Millisecond)
-// 			}
+//			for _, step := range tt.msgs {
+//				switch step.evType {
+//				case "AddPeer":
+//					reactor.scheduler.send(bcAddNewPeer{peerID: p2p.ID(step.peer)})
+//				case "RemovePeer":
+//					reactor.scheduler.send(bcRemovePeer{peerID: p2p.ID(step.peer)})
+//				case "ReceiveS":
+//					reactor.scheduler.send(bcStatusResponse{
+//						peerID: p2p.ID(step.peer),
+//						height: step.height,
+//						time:   time.Now(),
+//					})
+//				case "ReceiveB":
+//					reactor.scheduler.send(bcBlockResponse{
+//						peerID: p2p.ID(step.peer),
+//						block:  refStore.LoadBlock(step.height),
+//						size:   10,
+//						time:   time.Now(),
+//					})
+//				case "ReceiveNB":
+//					reactor.scheduler.send(bcNoBlockResponse{
+//						peerID: p2p.ID(step.peer),
+//						height: step.height,
+//						time:   time.Now(),
+//					})
+//				case "BlockReq":
+//					reactor.scheduler.send(rTrySchedule{time: time.Now()})
+//				case "Process":
+//					reactor.processor.send(rProcessBlock{})
+//				}
+//				// give time for messages to propagate between routines
+//				time.Sleep(time.Millisecond)
+//			}
 
-// 			// time for processor to finish and reactor to switch to consensus
-// 			time.Sleep(20 * time.Millisecond)
-// 			assert.True(t, mockSwitch.hasSwitchedToConsensus())
-// 			reactor.Stop()
-// 		})
-// 	}
+//			// time for processor to finish and reactor to switch to consensus
+//			time.Sleep(20 * time.Millisecond)
+//			assert.True(t, mockSwitch.hasSwitchedToConsensus())
+//			reactor.Stop()
+//		})
+//	}
 // }
 
 func TestReactorHelperMode(t *testing.T) {
@@ -482,7 +483,7 @@ func newReactorStore(
 
 	require.Len(t, privVals, 1)
 	app := &testApp{}
-	cc := proxy.NewLocalClientCreator(app)
+	cc := abciclient.NewLocalCreator(app)
 	proxyApp := proxy.NewAppConns(cc)
 	err := proxyApp.Start()
 	if err != nil {

internal/consensus/byzantine_test.go

@@ -10,7 +10,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	abcicli "github.com/tendermint/tendermint/abci/client"
+	abciclient "github.com/tendermint/tendermint/abci/client"
 	abci "github.com/tendermint/tendermint/abci/types"
 	"github.com/tendermint/tendermint/internal/evidence"
 	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
@@ -62,8 +62,8 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 
 			// one for mempool, one for consensus
 			mtx := new(tmsync.RWMutex)
-			proxyAppConnMem := abcicli.NewLocalClient(mtx, app)
-			proxyAppConnCon := abcicli.NewLocalClient(mtx, app)
+			proxyAppConnMem := abciclient.NewLocalClient(mtx, app)
+			proxyAppConnCon := abciclient.NewLocalClient(mtx, app)
 
 			// Make Mempool
 			mempool := mempoolv0.NewCListMempool(thisConfig.Mempool, proxyAppConnMem, 0)

internal/consensus/common_test.go

@@ -18,7 +18,7 @@ import (
 
 	dbm "github.com/tendermint/tm-db"
 
-	abcicli "github.com/tendermint/tendermint/abci/client"
+	abciclient "github.com/tendermint/tendermint/abci/client"
 	"github.com/tendermint/tendermint/abci/example/kvstore"
 	abci "github.com/tendermint/tendermint/abci/types"
 	cfg "github.com/tendermint/tendermint/config"
@@ -410,8 +410,8 @@ func newStateWithConfigAndBlockStore(
 ) *State {
 	// one for mempool, one for consensus
 	mtx := new(tmsync.RWMutex)
-	proxyAppConnMem := abcicli.NewLocalClient(mtx, app)
-	proxyAppConnCon := abcicli.NewLocalClient(mtx, app)
+	proxyAppConnMem := abciclient.NewLocalClient(mtx, app)
+	proxyAppConnCon := abciclient.NewLocalClient(mtx, app)
 
 	// Make Mempool
 	mempool := mempoolv0.NewCListMempool(thisConfig.Mempool, proxyAppConnMem, 0)

internal/consensus/reactor_test.go

@@ -13,7 +13,7 @@ import (
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
-	abcicli "github.com/tendermint/tendermint/abci/client"
+	abciclient "github.com/tendermint/tendermint/abci/client"
 	"github.com/tendermint/tendermint/abci/example/kvstore"
 	abci "github.com/tendermint/tendermint/abci/types"
 	cfg "github.com/tendermint/tendermint/config"
@@ -347,8 +347,8 @@ func TestReactorWithEvidence(t *testing.T) {
 
 		// one for mempool, one for consensus
 		mtx := new(tmsync.RWMutex)
-		proxyAppConnMem := abcicli.NewLocalClient(mtx, app)
-		proxyAppConnCon := abcicli.NewLocalClient(mtx, app)
+		proxyAppConnMem := abciclient.NewLocalClient(mtx, app)
+		proxyAppConnCon := abciclient.NewLocalClient(mtx, app)
 
 		mempool := mempoolv0.NewCListMempool(thisConfig.Mempool, proxyAppConnMem, 0)
 		mempool.SetLogger(log.TestingLogger().With("module", "mempool"))

internal/consensus/replay.go

@@ -11,8 +11,8 @@ import (
 
 	abci "github.com/tendermint/tendermint/abci/types"
 	"github.com/tendermint/tendermint/crypto/merkle"
+	"github.com/tendermint/tendermint/internal/proxy"
 	"github.com/tendermint/tendermint/libs/log"
-	"github.com/tendermint/tendermint/proxy"
 	sm "github.com/tendermint/tendermint/state"
 	"github.com/tendermint/tendermint/types"
 )

internal/consensus/replay_file.go

@@ -13,10 +13,10 @@ import (
 	dbm "github.com/tendermint/tm-db"
 
 	cfg "github.com/tendermint/tendermint/config"
+	"github.com/tendermint/tendermint/internal/proxy"
 	"github.com/tendermint/tendermint/libs/log"
 	tmos "github.com/tendermint/tendermint/libs/os"
 	tmpubsub "github.com/tendermint/tendermint/libs/pubsub"
-	"github.com/tendermint/tendermint/proxy"
 	sm "github.com/tendermint/tendermint/state"
 	"github.com/tendermint/tendermint/store"
 	"github.com/tendermint/tendermint/types"

internal/consensus/replay_stubs.go

@@ -3,11 +3,12 @@ package consensus
 import (
 	"context"
 
+	abciclient "github.com/tendermint/tendermint/abci/client"
 	abci "github.com/tendermint/tendermint/abci/types"
 	"github.com/tendermint/tendermint/internal/libs/clist"
 	mempl "github.com/tendermint/tendermint/internal/mempool"
+	"github.com/tendermint/tendermint/internal/proxy"
 	tmstate "github.com/tendermint/tendermint/proto/tendermint/state"
-	"github.com/tendermint/tendermint/proxy"
 	"github.com/tendermint/tendermint/types"
 )
 
@@ -53,11 +54,11 @@ func (emptyMempool) CloseWAL()      {}
 // the real app.
 
 func newMockProxyApp(appHash []byte, abciResponses *tmstate.ABCIResponses) proxy.AppConnConsensus {
-	clientCreator := proxy.NewLocalClientCreator(&mockProxyApp{
+	clientCreator := abciclient.NewLocalCreator(&mockProxyApp{
 		appHash:       appHash,
 		abciResponses: abciResponses,
 	})
-	cli, _ := clientCreator.NewABCIClient()
+	cli, _ := clientCreator()
 	err := cli.Start()
 	if err != nil {
 		panic(err)

internal/consensus/replay_test.go

@@ -19,19 +19,20 @@ import (
 	"github.com/stretchr/testify/require"
 	dbm "github.com/tendermint/tm-db"
 
+	abciclient "github.com/tendermint/tendermint/abci/client"
 	"github.com/tendermint/tendermint/abci/example/kvstore"
 	abci "github.com/tendermint/tendermint/abci/types"
 	cfg "github.com/tendermint/tendermint/config"
 	"github.com/tendermint/tendermint/crypto"
 	cryptoenc "github.com/tendermint/tendermint/crypto/encoding"
 	mempl "github.com/tendermint/tendermint/internal/mempool"
+	"github.com/tendermint/tendermint/internal/proxy"
 	"github.com/tendermint/tendermint/internal/test/factory"
 	"github.com/tendermint/tendermint/libs/log"
 	tmrand "github.com/tendermint/tendermint/libs/rand"
 	"github.com/tendermint/tendermint/privval"
 	tmstate "github.com/tendermint/tendermint/proto/tendermint/state"
 	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
-	"github.com/tendermint/tendermint/proxy"
 	sm "github.com/tendermint/tendermint/state"
 	sf "github.com/tendermint/tendermint/state/test/factory"
 	"github.com/tendermint/tendermint/store"
@@ -740,7 +741,7 @@ func testHandshakeReplay(t *testing.T, sim *simulatorTestSuite, nBlocks int, mod
 		filepath.Join(config.DBDir(), fmt.Sprintf("replay_test_%d_%d_a_r%d", nBlocks, mode, rand.Int())))
 	t.Cleanup(func() { require.NoError(t, kvstoreApp.Close()) })
 
-	clientCreator2 := proxy.NewLocalClientCreator(kvstoreApp)
+	clientCreator2 := abciclient.NewLocalCreator(kvstoreApp)
 	if nBlocks > 0 {
 		// run nBlocks against a new client to build up the app state.
 		// use a throwaway tendermint state
@@ -890,7 +891,7 @@ func buildTMStateFromChain(
 	kvstoreApp := kvstore.NewPersistentKVStoreApplication(
 		filepath.Join(config.DBDir(), fmt.Sprintf("replay_test_%d_%d_t", nBlocks, mode)))
 	defer kvstoreApp.Close()
-	clientCreator := proxy.NewLocalClientCreator(kvstoreApp)
+	clientCreator := abciclient.NewLocalCreator(kvstoreApp)
 
 	proxyApp := proxy.NewAppConns(clientCreator)
 	if err := proxyApp.Start(); err != nil {
@@ -958,7 +959,7 @@ func TestHandshakePanicsIfAppReturnsWrongAppHash(t *testing.T) {
 	//		- 0x03
 	{
 		app := &badApp{numBlocks: 3, allHashesAreWrong: true}
-		clientCreator := proxy.NewLocalClientCreator(app)
+		clientCreator := abciclient.NewLocalCreator(app)
 		proxyApp := proxy.NewAppConns(clientCreator)
 		err := proxyApp.Start()
 		require.NoError(t, err)
@@ -982,7 +983,7 @@ func TestHandshakePanicsIfAppReturnsWrongAppHash(t *testing.T) {
 	//		- RANDOM HASH
 	{
 		app := &badApp{numBlocks: 3, onlyLastHashIsWrong: true}
-		clientCreator := proxy.NewLocalClientCreator(app)
+		clientCreator := abciclient.NewLocalCreator(app)
 		proxyApp := proxy.NewAppConns(clientCreator)
 		err := proxyApp.Start()
 		require.NoError(t, err)
@@ -1225,7 +1226,7 @@ func TestHandshakeUpdatesValidators(t *testing.T) {
 	val, _ := factory.RandValidator(true, 10)
 	vals := types.NewValidatorSet([]*types.Validator{val})
 	app := &initChainApp{vals: types.TM2PB.ValidatorUpdates(vals)}
-	clientCreator := proxy.NewLocalClientCreator(app)
+	clientCreator := abciclient.NewLocalCreator(app)
 
 	config := ResetConfig("handshake_test_")
 	t.Cleanup(func() { _ = os.RemoveAll(config.RootDir) })

internal/consensus/wal_generator.go

@@ -13,11 +13,12 @@ import (
 	"github.com/stretchr/testify/require"
 	db "github.com/tendermint/tm-db"
 
+	abciclient "github.com/tendermint/tendermint/abci/client"
 	"github.com/tendermint/tendermint/abci/example/kvstore"
 	cfg "github.com/tendermint/tendermint/config"
+	"github.com/tendermint/tendermint/internal/proxy"
 	"github.com/tendermint/tendermint/libs/log"
 	"github.com/tendermint/tendermint/privval"
-	"github.com/tendermint/tendermint/proxy"
 	sm "github.com/tendermint/tendermint/state"
 	"github.com/tendermint/tendermint/store"
 	"github.com/tendermint/tendermint/types"
@@ -64,7 +65,7 @@ func WALGenerateNBlocks(t *testing.T, wr io.Writer, numBlocks int) (err error) {
 
 	blockStore := store.NewBlockStore(blockStoreDB)
 
-	proxyApp := proxy.NewAppConns(proxy.NewLocalClientCreator(app))
+	proxyApp := proxy.NewAppConns(abciclient.NewLocalCreator(app))
 	proxyApp.SetLogger(logger.With("module", "proxy"))
 	if err := proxyApp.Start(); err != nil {
 		return fmt.Errorf("failed to start proxy app connections: %w", err)

internal/mempool/v0/bench_test.go

@@ -6,14 +6,14 @@ import (
 	"sync/atomic"
 	"testing"
 
+	abciclient "github.com/tendermint/tendermint/abci/client"
 	"github.com/tendermint/tendermint/abci/example/kvstore"
 	"github.com/tendermint/tendermint/internal/mempool"
-	"github.com/tendermint/tendermint/proxy"
 )
 
 func BenchmarkReap(b *testing.B) {
 	app := kvstore.NewApplication()
-	cc := proxy.NewLocalClientCreator(app)
+	cc := abciclient.NewLocalCreator(app)
 	mp, cleanup := newMempoolWithApp(cc)
 	defer cleanup()
 	mp.config.Size = 100000
@@ -34,7 +34,7 @@ func BenchmarkReap(b *testing.B) {
 
 func BenchmarkCheckTx(b *testing.B) {
 	app := kvstore.NewApplication()
-	cc := proxy.NewLocalClientCreator(app)
+	cc := abciclient.NewLocalCreator(app)
 	mp, cleanup := newMempoolWithApp(cc)
 	defer cleanup()
 
@@ -56,7 +56,7 @@ func BenchmarkCheckTx(b *testing.B) {
 
 func BenchmarkParallelCheckTx(b *testing.B) {
 	app := kvstore.NewApplication()
-	cc := proxy.NewLocalClientCreator(app)
+	cc := abciclient.NewLocalCreator(app)
 	mp, cleanup := newMempoolWithApp(cc)
 	defer cleanup()
 
@@ -81,7 +81,7 @@ func BenchmarkParallelCheckTx(b *testing.B) {
 
 func BenchmarkCheckDuplicateTx(b *testing.B) {
 	app := kvstore.NewApplication()
-	cc := proxy.NewLocalClientCreator(app)
+	cc := abciclient.NewLocalCreator(app)
 	mp, cleanup := newMempoolWithApp(cc)
 	defer cleanup()
 

internal/mempool/v0/cache_test.go

@@ -7,16 +7,16 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	abciclient "github.com/tendermint/tendermint/abci/client"
 	"github.com/tendermint/tendermint/abci/example/kvstore"
 	abci "github.com/tendermint/tendermint/abci/types"
 	"github.com/tendermint/tendermint/internal/mempool"
-	"github.com/tendermint/tendermint/proxy"
 	"github.com/tendermint/tendermint/types"
 )
 
 func TestCacheAfterUpdate(t *testing.T) {
 	app := kvstore.NewApplication()
-	cc := proxy.NewLocalClientCreator(app)
+	cc := abciclient.NewLocalCreator(app)
 	mp, cleanup := newMempoolWithApp(cc)
 	defer cleanup()
 

internal/mempool/v0/clist_mempool.go

@@ -12,9 +12,9 @@ import (
 	"github.com/tendermint/tendermint/internal/libs/clist"
 	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
 	"github.com/tendermint/tendermint/internal/mempool"
+	"github.com/tendermint/tendermint/internal/proxy"
 	"github.com/tendermint/tendermint/libs/log"
 	tmmath "github.com/tendermint/tendermint/libs/math"
-	"github.com/tendermint/tendermint/proxy"
 	"github.com/tendermint/tendermint/types"
 )
 

internal/mempool/v0/clist_mempool_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	abciclient "github.com/tendermint/tendermint/abci/client"
 	"github.com/tendermint/tendermint/abci/example/kvstore"
 	abciserver "github.com/tendermint/tendermint/abci/server"
 	abci "github.com/tendermint/tendermint/abci/types"
@@ -23,7 +24,6 @@ import (
 	"github.com/tendermint/tendermint/libs/log"
 	tmrand "github.com/tendermint/tendermint/libs/rand"
 	"github.com/tendermint/tendermint/libs/service"
-	"github.com/tendermint/tendermint/proxy"
 	"github.com/tendermint/tendermint/types"
 )
 
@@ -31,12 +31,12 @@ import (
 // test.
 type cleanupFunc func()
 
-func newMempoolWithApp(cc proxy.ClientCreator) (*CListMempool, cleanupFunc) {
+func newMempoolWithApp(cc abciclient.Creator) (*CListMempool, cleanupFunc) {
 	return newMempoolWithAppAndConfig(cc, cfg.ResetTestRoot("mempool_test"))
 }
 
-func newMempoolWithAppAndConfig(cc proxy.ClientCreator, config *cfg.Config) (*CListMempool, cleanupFunc) {
-	appConnMem, _ := cc.NewABCIClient()
+func newMempoolWithAppAndConfig(cc abciclient.Creator, config *cfg.Config) (*CListMempool, cleanupFunc) {
+	appConnMem, _ := cc()
 	appConnMem.SetLogger(log.TestingLogger().With("module", "abci-client", "connection", "mempool"))
 	err := appConnMem.Start()
 	if err != nil {
@@ -92,7 +92,7 @@ func checkTxs(t *testing.T, mp mempool.Mempool, count int, peerID uint16) types.
 
 func TestReapMaxBytesMaxGas(t *testing.T) {
 	app := kvstore.NewApplication()
-	cc := proxy.NewLocalClientCreator(app)
+	cc := abciclient.NewLocalCreator(app)
 	mp, cleanup := newMempoolWithApp(cc)
 	defer cleanup()
 
@@ -141,7 +141,7 @@ func TestReapMaxBytesMaxGas(t *testing.T) {
 
 func TestMempoolFilters(t *testing.T) {
 	app := kvstore.NewApplication()
-	cc := proxy.NewLocalClientCreator(app)
+	cc := abciclient.NewLocalCreator(app)
 	mp, cleanup := newMempoolWithApp(cc)
 	defer cleanup()
 	emptyTxArr := []types.Tx{[]byte{}}
@@ -180,7 +180,7 @@ func TestMempoolFilters(t *testing.T) {
 
 func TestMempoolUpdate(t *testing.T) {
 	app := kvstore.NewApplication()
-	cc := proxy.NewLocalClientCreator(app)
+	cc := abciclient.NewLocalCreator(app)
 	mp, cleanup := newMempoolWithApp(cc)
 	defer cleanup()
 
@@ -216,7 +216,7 @@ func TestMempoolUpdate(t *testing.T) {
 
 func TestMempool_KeepInvalidTxsInCache(t *testing.T) {
 	app := kvstore.NewApplication()
-	cc := proxy.NewLocalClientCreator(app)
+	cc := abciclient.NewLocalCreator(app)
 	wcfg := cfg.DefaultConfig()
 	wcfg.Mempool.KeepInvalidTxsInCache = true
 	mp, cleanup := newMempoolWithAppAndConfig(cc, wcfg)
@@ -264,7 +264,7 @@ func TestMempool_KeepInvalidTxsInCache(t *testing.T) {
 
 func TestTxsAvailable(t *testing.T) {
 	app := kvstore.NewApplication()
-	cc := proxy.NewLocalClientCreator(app)
+	cc := abciclient.NewLocalCreator(app)
 	mp, cleanup := newMempoolWithApp(cc)
 	defer cleanup()
 	mp.EnableTxsAvailable()
@@ -308,12 +308,12 @@ func TestTxsAvailable(t *testing.T) {
 
 func TestSerialReap(t *testing.T) {
 	app := kvstore.NewApplication()
-	cc := proxy.NewLocalClientCreator(app)
+	cc := abciclient.NewLocalCreator(app)
 
 	mp, cleanup := newMempoolWithApp(cc)
 	defer cleanup()
 
-	appConnCon, _ := cc.NewABCIClient()
+	appConnCon, _ := cc()
 	appConnCon.SetLogger(log.TestingLogger().With("module", "abci-client", "connection", "consensus"))
 	err := appConnCon.Start()
 	require.Nil(t, err)
@@ -419,7 +419,7 @@ func TestSerialReap(t *testing.T) {
 
 func TestMempool_CheckTxChecksTxSize(t *testing.T) {
 	app := kvstore.NewApplication()
-	cc := proxy.NewLocalClientCreator(app)
+	cc := abciclient.NewLocalCreator(app)
 	mempl, cleanup := newMempoolWithApp(cc)
 	defer cleanup()
 
@@ -464,7 +464,7 @@ func TestMempool_CheckTxChecksTxSize(t *testing.T) {
 
 func TestMempoolTxsBytes(t *testing.T) {
 	app := kvstore.NewApplication()
-	cc := proxy.NewLocalClientCreator(app)
+	cc := abciclient.NewLocalCreator(app)
 	config := cfg.ResetTestRoot("mempool_test")
 	config.Mempool.MaxTxsBytes = 10
 	mp, cleanup := newMempoolWithAppAndConfig(cc, config)
@@ -507,7 +507,7 @@ func TestMempoolTxsBytes(t *testing.T) {
 
 	// 6. zero after tx is rechecked and removed due to not being valid anymore
 	app2 := kvstore.NewApplication()
-	cc = proxy.NewLocalClientCreator(app2)
+	cc = abciclient.NewLocalCreator(app2)
 	mp, cleanup = newMempoolWithApp(cc)
 	defer cleanup()
 
@@ -518,7 +518,7 @@ func TestMempoolTxsBytes(t *testing.T) {
 	require.NoError(t, err)
 	assert.EqualValues(t, 8, mp.SizeBytes())
 
-	appConnCon, _ := cc.NewABCIClient()
+	appConnCon, _ := cc()
 	appConnCon.SetLogger(log.TestingLogger().With("module", "abci-client", "connection", "consensus"))
 	err = appConnCon.Start()
 	require.Nil(t, err)
@@ -597,10 +597,10 @@ func newRemoteApp(
 	addr string,
 	app abci.Application,
 ) (
-	clientCreator proxy.ClientCreator,
+	clientCreator abciclient.Creator,
 	server service.Service,
 ) {
-	clientCreator = proxy.NewRemoteClientCreator(addr, "socket", true)
+	clientCreator = abciclient.NewRemoteCreator(addr, "socket", true)
 
 	// Start server
 	server = abciserver.NewSocketServer(addr, app)

internal/mempool/v0/reactor_test.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	abciclient "github.com/tendermint/tendermint/abci/client"
 	"github.com/tendermint/tendermint/abci/example/kvstore"
 	abci "github.com/tendermint/tendermint/abci/types"
 	cfg "github.com/tendermint/tendermint/config"
@@ -17,7 +18,6 @@ import (
 	"github.com/tendermint/tendermint/libs/log"
 	tmrand "github.com/tendermint/tendermint/libs/rand"
 	protomem "github.com/tendermint/tendermint/proto/tendermint/mempool"
-	"github.com/tendermint/tendermint/proxy"
 	"github.com/tendermint/tendermint/types"
 )
 
@@ -55,7 +55,7 @@ func setup(t *testing.T, cfg *cfg.MempoolConfig, numNodes int, chBuf uint) *reac
 
 	for nodeID := range rts.network.Nodes {
 		rts.kvstores[nodeID] = kvstore.NewApplication()
-		cc := proxy.NewLocalClientCreator(rts.kvstores[nodeID])
+		cc := abciclient.NewLocalCreator(rts.kvstores[nodeID])
 
 		mempool, memCleanup := newMempoolWithApp(cc)
 		t.Cleanup(memCleanup)

internal/mempool/v1/mempool.go

@@ -12,9 +12,9 @@ import (
 	"github.com/tendermint/tendermint/internal/libs/clist"
 	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
 	"github.com/tendermint/tendermint/internal/mempool"
+	"github.com/tendermint/tendermint/internal/proxy"
 	"github.com/tendermint/tendermint/libs/log"
 	tmmath "github.com/tendermint/tendermint/libs/math"
-	"github.com/tendermint/tendermint/proxy"
 	"github.com/tendermint/tendermint/types"
 )
 

internal/mempool/v1/mempool_test.go

@@ -15,13 +15,14 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/require"
+
+	abciclient "github.com/tendermint/tendermint/abci/client"
 	"github.com/tendermint/tendermint/abci/example/code"
 	"github.com/tendermint/tendermint/abci/example/kvstore"
 	abci "github.com/tendermint/tendermint/abci/types"
 	"github.com/tendermint/tendermint/config"
 	"github.com/tendermint/tendermint/internal/mempool"
 	"github.com/tendermint/tendermint/libs/log"
-	"github.com/tendermint/tendermint/proxy"
 	"github.com/tendermint/tendermint/types"
 )
 
@@ -76,12 +77,12 @@ func setup(t testing.TB, cacheSize int, options ...TxMempoolOption) *TxMempool {
 	t.Helper()
 
 	app := &application{kvstore.NewApplication()}
-	cc := proxy.NewLocalClientCreator(app)
+	cc := abciclient.NewLocalCreator(app)
 
 	cfg := config.ResetTestRoot(strings.ReplaceAll(t.Name(), "/", "|"))
 	cfg.Mempool.CacheSize = cacheSize
 
-	appConnMem, err := cc.NewABCIClient()
+	appConnMem, err := cc()
 	require.NoError(t, err)
 	require.NoError(t, appConnMem.Start())
 

internal/proxy/app_conn.go

@@ -0,0 +1,215 @@
+package proxy
+
+import (
+	"context"
+
+	abciclient "github.com/tendermint/tendermint/abci/client"
+	"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 {
+	SetResponseCallback(abciclient.Callback)
+	Error() error
+
+	InitChainSync(context.Context, types.RequestInitChain) (*types.ResponseInitChain, error)
+
+	BeginBlockSync(context.Context, types.RequestBeginBlock) (*types.ResponseBeginBlock, error)
+	DeliverTxAsync(context.Context, types.RequestDeliverTx) (*abciclient.ReqRes, error)
+	EndBlockSync(context.Context, types.RequestEndBlock) (*types.ResponseEndBlock, error)
+	CommitSync(context.Context) (*types.ResponseCommit, error)
+}
+
+type AppConnMempool interface {
+	SetResponseCallback(abciclient.Callback)
+	Error() error
+
+	CheckTxAsync(context.Context, types.RequestCheckTx) (*abciclient.ReqRes, error)
+	CheckTxSync(context.Context, types.RequestCheckTx) (*types.ResponseCheckTx, error)
+
+	FlushAsync(context.Context) (*abciclient.ReqRes, error)
+	FlushSync(context.Context) error
+}
+
+type AppConnQuery interface {
+	Error() error
+
+	EchoSync(context.Context, string) (*types.ResponseEcho, error)
+	InfoSync(context.Context, types.RequestInfo) (*types.ResponseInfo, error)
+	QuerySync(context.Context, types.RequestQuery) (*types.ResponseQuery, error)
+}
+
+type AppConnSnapshot interface {
+	Error() error
+
+	ListSnapshotsSync(context.Context, types.RequestListSnapshots) (*types.ResponseListSnapshots, error)
+	OfferSnapshotSync(context.Context, types.RequestOfferSnapshot) (*types.ResponseOfferSnapshot, error)
+	LoadSnapshotChunkSync(context.Context, types.RequestLoadSnapshotChunk) (*types.ResponseLoadSnapshotChunk, error)
+	ApplySnapshotChunkSync(context.Context, types.RequestApplySnapshotChunk) (*types.ResponseApplySnapshotChunk, error)
+}
+
+//-----------------------------------------------------------------------------------------
+// Implements AppConnConsensus (subset of abciclient.Client)
+
+type appConnConsensus struct {
+	appConn abciclient.Client
+}
+
+func NewAppConnConsensus(appConn abciclient.Client) AppConnConsensus {
+	return &appConnConsensus{
+		appConn: appConn,
+	}
+}
+
+func (app *appConnConsensus) SetResponseCallback(cb abciclient.Callback) {
+	app.appConn.SetResponseCallback(cb)
+}
+
+func (app *appConnConsensus) Error() error {
+	return app.appConn.Error()
+}
+
+func (app *appConnConsensus) InitChainSync(
+	ctx context.Context,
+	req types.RequestInitChain,
+) (*types.ResponseInitChain, error) {
+	return app.appConn.InitChainSync(ctx, req)
+}
+
+func (app *appConnConsensus) BeginBlockSync(
+	ctx context.Context,
+	req types.RequestBeginBlock,
+) (*types.ResponseBeginBlock, error) {
+	return app.appConn.BeginBlockSync(ctx, req)
+}
+
+func (app *appConnConsensus) DeliverTxAsync(
+	ctx context.Context,
+	req types.RequestDeliverTx,
+) (*abciclient.ReqRes, error) {
+	return app.appConn.DeliverTxAsync(ctx, req)
+}
+
+func (app *appConnConsensus) EndBlockSync(
+	ctx context.Context,
+	req types.RequestEndBlock,
+) (*types.ResponseEndBlock, error) {
+	return app.appConn.EndBlockSync(ctx, req)
+}
+
+func (app *appConnConsensus) CommitSync(ctx context.Context) (*types.ResponseCommit, error) {
+	return app.appConn.CommitSync(ctx)
+}
+
+//------------------------------------------------
+// Implements AppConnMempool (subset of abciclient.Client)
+
+type appConnMempool struct {
+	appConn abciclient.Client
+}
+
+func NewAppConnMempool(appConn abciclient.Client) AppConnMempool {
+	return &appConnMempool{
+		appConn: appConn,
+	}
+}
+
+func (app *appConnMempool) SetResponseCallback(cb abciclient.Callback) {
+	app.appConn.SetResponseCallback(cb)
+}
+
+func (app *appConnMempool) Error() error {
+	return app.appConn.Error()
+}
+
+func (app *appConnMempool) FlushAsync(ctx context.Context) (*abciclient.ReqRes, error) {
+	return app.appConn.FlushAsync(ctx)
+}
+
+func (app *appConnMempool) FlushSync(ctx context.Context) error {
+	return app.appConn.FlushSync(ctx)
+}
+
+func (app *appConnMempool) CheckTxAsync(ctx context.Context, req types.RequestCheckTx) (*abciclient.ReqRes, error) {
+	return app.appConn.CheckTxAsync(ctx, req)
+}
+
+func (app *appConnMempool) CheckTxSync(ctx context.Context, req types.RequestCheckTx) (*types.ResponseCheckTx, error) {
+	return app.appConn.CheckTxSync(ctx, req)
+}
+
+//------------------------------------------------
+// Implements AppConnQuery (subset of abciclient.Client)
+
+type appConnQuery struct {
+	appConn abciclient.Client
+}
+
+func NewAppConnQuery(appConn abciclient.Client) AppConnQuery {
+	return &appConnQuery{
+		appConn: appConn,
+	}
+}
+
+func (app *appConnQuery) Error() error {
+	return app.appConn.Error()
+}
+
+func (app *appConnQuery) EchoSync(ctx context.Context, msg string) (*types.ResponseEcho, error) {
+	return app.appConn.EchoSync(ctx, msg)
+}
+
+func (app *appConnQuery) InfoSync(ctx context.Context, req types.RequestInfo) (*types.ResponseInfo, error) {
+	return app.appConn.InfoSync(ctx, req)
+}
+
+func (app *appConnQuery) QuerySync(ctx context.Context, reqQuery types.RequestQuery) (*types.ResponseQuery, error) {
+	return app.appConn.QuerySync(ctx, reqQuery)
+}
+
+//------------------------------------------------
+// Implements AppConnSnapshot (subset of abciclient.Client)
+
+type appConnSnapshot struct {
+	appConn abciclient.Client
+}
+
+func NewAppConnSnapshot(appConn abciclient.Client) AppConnSnapshot {
+	return &appConnSnapshot{
+		appConn: appConn,
+	}
+}
+
+func (app *appConnSnapshot) Error() error {
+	return app.appConn.Error()
+}
+
+func (app *appConnSnapshot) ListSnapshotsSync(
+	ctx context.Context,
+	req types.RequestListSnapshots,
+) (*types.ResponseListSnapshots, error) {
+	return app.appConn.ListSnapshotsSync(ctx, req)
+}
+
+func (app *appConnSnapshot) OfferSnapshotSync(
+	ctx context.Context,
+	req types.RequestOfferSnapshot,
+) (*types.ResponseOfferSnapshot, error) {
+	return app.appConn.OfferSnapshotSync(ctx, req)
+}
+
+func (app *appConnSnapshot) LoadSnapshotChunkSync(
+	ctx context.Context,
+	req types.RequestLoadSnapshotChunk) (*types.ResponseLoadSnapshotChunk, error) {
+	return app.appConn.LoadSnapshotChunkSync(ctx, req)
+}
+
+func (app *appConnSnapshot) ApplySnapshotChunkSync(
+	ctx context.Context,
+	req types.RequestApplySnapshotChunk) (*types.ResponseApplySnapshotChunk, error) {
+	return app.appConn.ApplySnapshotChunkSync(ctx, req)
+}

internal/proxy/app_conn_test.go

@@ -0,0 +1,186 @@
+package proxy
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+
+	abciclient "github.com/tendermint/tendermint/abci/client"
+	"github.com/tendermint/tendermint/abci/example/kvstore"
+	"github.com/tendermint/tendermint/abci/server"
+	"github.com/tendermint/tendermint/abci/types"
+	"github.com/tendermint/tendermint/libs/log"
+	tmrand "github.com/tendermint/tendermint/libs/rand"
+)
+
+//----------------------------------------
+
+type appConnTestI interface {
+	EchoAsync(ctx context.Context, msg string) (*abciclient.ReqRes, error)
+	FlushSync(context.Context) error
+	InfoSync(context.Context, types.RequestInfo) (*types.ResponseInfo, error)
+}
+
+type appConnTest struct {
+	appConn abciclient.Client
+}
+
+func newAppConnTest(appConn abciclient.Client) appConnTestI {
+	return &appConnTest{appConn}
+}
+
+func (app *appConnTest) EchoAsync(ctx context.Context, msg string) (*abciclient.ReqRes, error) {
+	return app.appConn.EchoAsync(ctx, msg)
+}
+
+func (app *appConnTest) FlushSync(ctx context.Context) error {
+	return app.appConn.FlushSync(ctx)
+}
+
+func (app *appConnTest) InfoSync(ctx context.Context, req types.RequestInfo) (*types.ResponseInfo, error) {
+	return app.appConn.InfoSync(ctx, req)
+}
+
+//----------------------------------------
+
+var SOCKET = "socket"
+
+func TestEcho(t *testing.T) {
+	sockPath := fmt.Sprintf("unix:///tmp/echo_%v.sock", tmrand.Str(6))
+	clientCreator := abciclient.NewRemoteCreator(sockPath, SOCKET, true)
+
+	// Start server
+	s := server.NewSocketServer(sockPath, kvstore.NewApplication())
+	s.SetLogger(log.TestingLogger().With("module", "abci-server"))
+	if err := s.Start(); err != nil {
+		t.Fatalf("Error starting socket server: %v", err.Error())
+	}
+	t.Cleanup(func() {
+		if err := s.Stop(); err != nil {
+			t.Error(err)
+		}
+	})
+
+	// Start client
+	cli, err := clientCreator()
+	if err != nil {
+		t.Fatalf("Error creating ABCI client: %v", err.Error())
+	}
+	cli.SetLogger(log.TestingLogger().With("module", "abci-client"))
+	if err := cli.Start(); err != nil {
+		t.Fatalf("Error starting ABCI client: %v", err.Error())
+	}
+
+	proxy := newAppConnTest(cli)
+	t.Log("Connected")
+
+	ctx := context.Background()
+	for i := 0; i < 1000; i++ {
+		_, err = proxy.EchoAsync(ctx, fmt.Sprintf("echo-%v", i))
+		if err != nil {
+			t.Error(err)
+		}
+		// flush sometimes
+		if i%128 == 0 {
+			if err := proxy.FlushSync(ctx); err != nil {
+				t.Error(err)
+			}
+		}
+	}
+	if err := proxy.FlushSync(ctx); err != nil {
+		t.Error(err)
+	}
+}
+
+func BenchmarkEcho(b *testing.B) {
+	b.StopTimer() // Initialize
+	sockPath := fmt.Sprintf("unix:///tmp/echo_%v.sock", tmrand.Str(6))
+	clientCreator := abciclient.NewRemoteCreator(sockPath, SOCKET, true)
+
+	// Start server
+	s := server.NewSocketServer(sockPath, kvstore.NewApplication())
+	s.SetLogger(log.TestingLogger().With("module", "abci-server"))
+	if err := s.Start(); err != nil {
+		b.Fatalf("Error starting socket server: %v", err.Error())
+	}
+	b.Cleanup(func() {
+		if err := s.Stop(); err != nil {
+			b.Error(err)
+		}
+	})
+
+	// Start client
+	cli, err := clientCreator()
+	if err != nil {
+		b.Fatalf("Error creating ABCI client: %v", err.Error())
+	}
+	cli.SetLogger(log.TestingLogger().With("module", "abci-client"))
+	if err := cli.Start(); err != nil {
+		b.Fatalf("Error starting ABCI client: %v", err.Error())
+	}
+
+	proxy := newAppConnTest(cli)
+	b.Log("Connected")
+	echoString := strings.Repeat(" ", 200)
+	b.StartTimer() // Start benchmarking tests
+
+	ctx := context.Background()
+	for i := 0; i < b.N; i++ {
+		_, err = proxy.EchoAsync(ctx, echoString)
+		if err != nil {
+			b.Error(err)
+		}
+		// flush sometimes
+		if i%128 == 0 {
+			if err := proxy.FlushSync(ctx); err != nil {
+				b.Error(err)
+			}
+		}
+	}
+	if err := proxy.FlushSync(ctx); err != nil {
+		b.Error(err)
+	}
+
+	b.StopTimer()
+	// info := proxy.InfoSync(types.RequestInfo{""})
+	// b.Log("N: ", b.N, info)
+}
+
+func TestInfo(t *testing.T) {
+	sockPath := fmt.Sprintf("unix:///tmp/echo_%v.sock", tmrand.Str(6))
+	clientCreator := abciclient.NewRemoteCreator(sockPath, SOCKET, true)
+
+	// Start server
+	s := server.NewSocketServer(sockPath, kvstore.NewApplication())
+	s.SetLogger(log.TestingLogger().With("module", "abci-server"))
+	if err := s.Start(); err != nil {
+		t.Fatalf("Error starting socket server: %v", err.Error())
+	}
+	t.Cleanup(func() {
+		if err := s.Stop(); err != nil {
+			t.Error(err)
+		}
+	})
+
+	// Start client
+	cli, err := clientCreator()
+	if err != nil {
+		t.Fatalf("Error creating ABCI client: %v", err.Error())
+	}
+	cli.SetLogger(log.TestingLogger().With("module", "abci-client"))
+	if err := cli.Start(); err != nil {
+		t.Fatalf("Error starting ABCI client: %v", err.Error())
+	}
+
+	proxy := newAppConnTest(cli)
+	t.Log("Connected")
+
+	resInfo, err := proxy.InfoSync(context.Background(), RequestInfo)
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+	if resInfo.Data != "{\"size\":0}" {
+		t.Error("Expected ResponseInfo with one element '{\"size\":0}' but got something else")
+	}
+}

internal/proxy/client.go

@@ -0,0 +1,34 @@
+package proxy
+
+import (
+	"io"
+
+	abciclient "github.com/tendermint/tendermint/abci/client"
+	"github.com/tendermint/tendermint/abci/example/kvstore"
+	"github.com/tendermint/tendermint/abci/types"
+)
+
+// DefaultClientCreator returns a default ClientCreator, which will create a
+// local client if addr is one of: 'kvstore',
+// 'persistent_kvstore' or 'noop', otherwise - a remote client.
+//
+// The Closer is a noop except for persistent_kvstore applications,
+// which will clean up the store.
+func DefaultClientCreator(addr, transport, dbDir string) (abciclient.Creator, io.Closer) {
+	switch addr {
+	case "kvstore":
+		return abciclient.NewLocalCreator(kvstore.NewApplication()), noopCloser{}
+	case "persistent_kvstore":
+		app := kvstore.NewPersistentKVStoreApplication(dbDir)
+		return abciclient.NewLocalCreator(app), app
+	case "noop":
+		return abciclient.NewLocalCreator(types.NewBaseApplication()), noopCloser{}
+	default:
+		mustConnect := false // loop retrying
+		return abciclient.NewRemoteCreator(addr, transport, mustConnect), noopCloser{}
+	}
+}
+
+type noopCloser struct{}
+
+func (noopCloser) Close() error { return nil }

internal/proxy/mocks/app_conn_consensus.go

@@ -0,0 +1,152 @@
+// Code generated by mockery. DO NOT EDIT.
+
+package mocks
+
+import (
+	context "context"
+
+	abciclient "github.com/tendermint/tendermint/abci/client"
+
+	mock "github.com/stretchr/testify/mock"
+
+	types "github.com/tendermint/tendermint/abci/types"
+)
+
+// AppConnConsensus is an autogenerated mock type for the AppConnConsensus type
+type AppConnConsensus struct {
+	mock.Mock
+}
+
+// BeginBlockSync provides a mock function with given fields: _a0, _a1
+func (_m *AppConnConsensus) BeginBlockSync(_a0 context.Context, _a1 types.RequestBeginBlock) (*types.ResponseBeginBlock, error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *types.ResponseBeginBlock
+	if rf, ok := ret.Get(0).(func(context.Context, types.RequestBeginBlock) *types.ResponseBeginBlock); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*types.ResponseBeginBlock)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, types.RequestBeginBlock) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// CommitSync provides a mock function with given fields: _a0
+func (_m *AppConnConsensus) CommitSync(_a0 context.Context) (*types.ResponseCommit, error) {
+	ret := _m.Called(_a0)
+
+	var r0 *types.ResponseCommit
+	if rf, ok := ret.Get(0).(func(context.Context) *types.ResponseCommit); ok {
+		r0 = rf(_a0)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*types.ResponseCommit)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context) error); ok {
+		r1 = rf(_a0)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// DeliverTxAsync provides a mock function with given fields: _a0, _a1
+func (_m *AppConnConsensus) DeliverTxAsync(_a0 context.Context, _a1 types.RequestDeliverTx) (*abciclient.ReqRes, error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *abciclient.ReqRes
+	if rf, ok := ret.Get(0).(func(context.Context, types.RequestDeliverTx) *abciclient.ReqRes); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*abciclient.ReqRes)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, types.RequestDeliverTx) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// EndBlockSync provides a mock function with given fields: _a0, _a1
+func (_m *AppConnConsensus) EndBlockSync(_a0 context.Context, _a1 types.RequestEndBlock) (*types.ResponseEndBlock, error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *types.ResponseEndBlock
+	if rf, ok := ret.Get(0).(func(context.Context, types.RequestEndBlock) *types.ResponseEndBlock); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*types.ResponseEndBlock)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, types.RequestEndBlock) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Error provides a mock function with given fields:
+func (_m *AppConnConsensus) Error() error {
+	ret := _m.Called()
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func() error); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// InitChainSync provides a mock function with given fields: _a0, _a1
+func (_m *AppConnConsensus) InitChainSync(_a0 context.Context, _a1 types.RequestInitChain) (*types.ResponseInitChain, error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *types.ResponseInitChain
+	if rf, ok := ret.Get(0).(func(context.Context, types.RequestInitChain) *types.ResponseInitChain); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*types.ResponseInitChain)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, types.RequestInitChain) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// SetResponseCallback provides a mock function with given fields: _a0
+func (_m *AppConnConsensus) SetResponseCallback(_a0 abciclient.Callback) {
+	_m.Called(_a0)
+}

internal/proxy/mocks/app_conn_mempool.go

@@ -0,0 +1,120 @@
+// Code generated by mockery. DO NOT EDIT.
+
+package mocks
+
+import (
+	context "context"
+
+	abciclient "github.com/tendermint/tendermint/abci/client"
+
+	mock "github.com/stretchr/testify/mock"
+
+	types "github.com/tendermint/tendermint/abci/types"
+)
+
+// AppConnMempool is an autogenerated mock type for the AppConnMempool type
+type AppConnMempool struct {
+	mock.Mock
+}
+
+// CheckTxAsync provides a mock function with given fields: _a0, _a1
+func (_m *AppConnMempool) CheckTxAsync(_a0 context.Context, _a1 types.RequestCheckTx) (*abciclient.ReqRes, error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *abciclient.ReqRes
+	if rf, ok := ret.Get(0).(func(context.Context, types.RequestCheckTx) *abciclient.ReqRes); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*abciclient.ReqRes)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, types.RequestCheckTx) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// CheckTxSync provides a mock function with given fields: _a0, _a1
+func (_m *AppConnMempool) CheckTxSync(_a0 context.Context, _a1 types.RequestCheckTx) (*types.ResponseCheckTx, error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *types.ResponseCheckTx
+	if rf, ok := ret.Get(0).(func(context.Context, types.RequestCheckTx) *types.ResponseCheckTx); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*types.ResponseCheckTx)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, types.RequestCheckTx) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Error provides a mock function with given fields:
+func (_m *AppConnMempool) Error() error {
+	ret := _m.Called()
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func() error); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// FlushAsync provides a mock function with given fields: _a0
+func (_m *AppConnMempool) FlushAsync(_a0 context.Context) (*abciclient.ReqRes, error) {
+	ret := _m.Called(_a0)
+
+	var r0 *abciclient.ReqRes
+	if rf, ok := ret.Get(0).(func(context.Context) *abciclient.ReqRes); ok {
+		r0 = rf(_a0)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*abciclient.ReqRes)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context) error); ok {
+		r1 = rf(_a0)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// FlushSync provides a mock function with given fields: _a0
+func (_m *AppConnMempool) FlushSync(_a0 context.Context) error {
+	ret := _m.Called(_a0)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context) error); ok {
+		r0 = rf(_a0)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// SetResponseCallback provides a mock function with given fields: _a0
+func (_m *AppConnMempool) SetResponseCallback(_a0 abciclient.Callback) {
+	_m.Called(_a0)
+}

internal/proxy/mocks/app_conn_query.go

@@ -0,0 +1,99 @@
+// Code generated by mockery. DO NOT EDIT.
+
+package mocks
+
+import (
+	context "context"
+
+	mock "github.com/stretchr/testify/mock"
+
+	types "github.com/tendermint/tendermint/abci/types"
+)
+
+// AppConnQuery is an autogenerated mock type for the AppConnQuery type
+type AppConnQuery struct {
+	mock.Mock
+}
+
+// EchoSync provides a mock function with given fields: _a0, _a1
+func (_m *AppConnQuery) EchoSync(_a0 context.Context, _a1 string) (*types.ResponseEcho, error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *types.ResponseEcho
+	if rf, ok := ret.Get(0).(func(context.Context, string) *types.ResponseEcho); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*types.ResponseEcho)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, string) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Error provides a mock function with given fields:
+func (_m *AppConnQuery) Error() error {
+	ret := _m.Called()
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func() error); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// InfoSync provides a mock function with given fields: _a0, _a1
+func (_m *AppConnQuery) InfoSync(_a0 context.Context, _a1 types.RequestInfo) (*types.ResponseInfo, error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *types.ResponseInfo
+	if rf, ok := ret.Get(0).(func(context.Context, types.RequestInfo) *types.ResponseInfo); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*types.ResponseInfo)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, types.RequestInfo) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// QuerySync provides a mock function with given fields: _a0, _a1
+func (_m *AppConnQuery) QuerySync(_a0 context.Context, _a1 types.RequestQuery) (*types.ResponseQuery, error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *types.ResponseQuery
+	if rf, ok := ret.Get(0).(func(context.Context, types.RequestQuery) *types.ResponseQuery); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*types.ResponseQuery)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, types.RequestQuery) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}

internal/proxy/mocks/app_conn_snapshot.go

@@ -0,0 +1,122 @@
+// Code generated by mockery. DO NOT EDIT.
+
+package mocks
+
+import (
+	context "context"
+
+	mock "github.com/stretchr/testify/mock"
+
+	types "github.com/tendermint/tendermint/abci/types"
+)
+
+// AppConnSnapshot is an autogenerated mock type for the AppConnSnapshot type
+type AppConnSnapshot struct {
+	mock.Mock
+}
+
+// ApplySnapshotChunkSync provides a mock function with given fields: _a0, _a1
+func (_m *AppConnSnapshot) ApplySnapshotChunkSync(_a0 context.Context, _a1 types.RequestApplySnapshotChunk) (*types.ResponseApplySnapshotChunk, error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *types.ResponseApplySnapshotChunk
+	if rf, ok := ret.Get(0).(func(context.Context, types.RequestApplySnapshotChunk) *types.ResponseApplySnapshotChunk); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*types.ResponseApplySnapshotChunk)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, types.RequestApplySnapshotChunk) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Error provides a mock function with given fields:
+func (_m *AppConnSnapshot) Error() error {
+	ret := _m.Called()
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func() error); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// ListSnapshotsSync provides a mock function with given fields: _a0, _a1
+func (_m *AppConnSnapshot) ListSnapshotsSync(_a0 context.Context, _a1 types.RequestListSnapshots) (*types.ResponseListSnapshots, error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *types.ResponseListSnapshots
+	if rf, ok := ret.Get(0).(func(context.Context, types.RequestListSnapshots) *types.ResponseListSnapshots); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*types.ResponseListSnapshots)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, types.RequestListSnapshots) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// LoadSnapshotChunkSync provides a mock function with given fields: _a0, _a1
+func (_m *AppConnSnapshot) LoadSnapshotChunkSync(_a0 context.Context, _a1 types.RequestLoadSnapshotChunk) (*types.ResponseLoadSnapshotChunk, error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *types.ResponseLoadSnapshotChunk
+	if rf, ok := ret.Get(0).(func(context.Context, types.RequestLoadSnapshotChunk) *types.ResponseLoadSnapshotChunk); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*types.ResponseLoadSnapshotChunk)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, types.RequestLoadSnapshotChunk) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// OfferSnapshotSync provides a mock function with given fields: _a0, _a1
+func (_m *AppConnSnapshot) OfferSnapshotSync(_a0 context.Context, _a1 types.RequestOfferSnapshot) (*types.ResponseOfferSnapshot, error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *types.ResponseOfferSnapshot
+	if rf, ok := ret.Get(0).(func(context.Context, types.RequestOfferSnapshot) *types.ResponseOfferSnapshot); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*types.ResponseOfferSnapshot)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, types.RequestOfferSnapshot) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}

internal/proxy/multi_app_conn.go

@@ -0,0 +1,200 @@
+package proxy
+
+import (
+	"fmt"
+	"os"
+	"syscall"
+
+	abciclient "github.com/tendermint/tendermint/abci/client"
+	tmlog "github.com/tendermint/tendermint/libs/log"
+	"github.com/tendermint/tendermint/libs/service"
+)
+
+const (
+	connConsensus = "consensus"
+	connMempool   = "mempool"
+	connQuery     = "query"
+	connSnapshot  = "snapshot"
+)
+
+// 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) AppConns {
+	return NewMultiAppConn(clientCreator)
+}
+
+// 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
+
+	consensusConn AppConnConsensus
+	mempoolConn   AppConnMempool
+	queryConn     AppConnQuery
+	snapshotConn  AppConnSnapshot
+
+	consensusConnClient abciclient.Client
+	mempoolConnClient   abciclient.Client
+	queryConnClient     abciclient.Client
+	snapshotConnClient  abciclient.Client
+
+	clientCreator abciclient.Creator
+}
+
+// NewMultiAppConn makes all necessary abci connections to the application.
+func NewMultiAppConn(clientCreator abciclient.Creator) AppConns {
+	multiAppConn := &multiAppConn{
+		clientCreator: clientCreator,
+	}
+	multiAppConn.BaseService = *service.NewBaseService(nil, "multiAppConn", multiAppConn)
+	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
+}
+
+func (app *multiAppConn) OnStart() error {
+	c, err := app.abciClientFor(connQuery)
+	if err != nil {
+		return err
+	}
+	app.queryConnClient = c
+	app.queryConn = NewAppConnQuery(c)
+
+	c, err = app.abciClientFor(connSnapshot)
+	if err != nil {
+		app.stopAllClients()
+		return err
+	}
+	app.snapshotConnClient = c
+	app.snapshotConn = NewAppConnSnapshot(c)
+
+	c, err = app.abciClientFor(connMempool)
+	if err != nil {
+		app.stopAllClients()
+		return err
+	}
+	app.mempoolConnClient = c
+	app.mempoolConn = NewAppConnMempool(c)
+
+	c, err = app.abciClientFor(connConsensus)
+	if err != nil {
+		app.stopAllClients()
+		return err
+	}
+	app.consensusConnClient = c
+	app.consensusConn = NewAppConnConsensus(c)
+
+	// Kill Tendermint if the ABCI application crashes.
+	go app.killTMOnClientError()
+
+	return nil
+}
+
+func (app *multiAppConn) OnStop() {
+	app.stopAllClients()
+}
+
+func (app *multiAppConn) killTMOnClientError() {
+	killFn := func(conn string, err error, logger tmlog.Logger) {
+		logger.Error(
+			fmt.Sprintf("%s connection terminated. Did the application crash? Please restart tendermint", conn),
+			"err", err)
+		if killErr := kill(); killErr != nil {
+			logger.Error("Failed to kill this process - please do so manually", "err", killErr)
+		}
+	}
+
+	select {
+	case <-app.consensusConnClient.Quit():
+		if err := app.consensusConnClient.Error(); err != nil {
+			killFn(connConsensus, err, app.Logger)
+		}
+	case <-app.mempoolConnClient.Quit():
+		if err := app.mempoolConnClient.Error(); err != nil {
+			killFn(connMempool, err, app.Logger)
+		}
+	case <-app.queryConnClient.Quit():
+		if err := app.queryConnClient.Error(); err != nil {
+			killFn(connQuery, err, app.Logger)
+		}
+	case <-app.snapshotConnClient.Quit():
+		if err := app.snapshotConnClient.Error(); err != nil {
+			killFn(connSnapshot, err, app.Logger)
+		}
+	}
+}
+
+func (app *multiAppConn) stopAllClients() {
+	if app.consensusConnClient != nil {
+		if err := app.consensusConnClient.Stop(); err != nil {
+			app.Logger.Error("error while stopping consensus client", "error", err)
+		}
+	}
+	if app.mempoolConnClient != nil {
+		if err := app.mempoolConnClient.Stop(); err != nil {
+			app.Logger.Error("error while stopping mempool client", "error", err)
+		}
+	}
+	if app.queryConnClient != nil {
+		if err := app.queryConnClient.Stop(); err != nil {
+			app.Logger.Error("error while stopping query client", "error", err)
+		}
+	}
+	if app.snapshotConnClient != nil {
+		if err := app.snapshotConnClient.Stop(); err != nil {
+			app.Logger.Error("error while stopping snapshot client", "error", err)
+		}
+	}
+}
+
+func (app *multiAppConn) abciClientFor(conn string) (abciclient.Client, error) {
+	c, err := app.clientCreator()
+	if err != nil {
+		return nil, fmt.Errorf("error creating ABCI client (%s connection): %w", conn, err)
+	}
+	c.SetLogger(app.Logger.With("module", "abci-client", "connection", conn))
+	if err := c.Start(); err != nil {
+		return nil, fmt.Errorf("error starting ABCI client (%s connection): %w", conn, err)
+	}
+	return c, nil
+}
+
+func kill() error {
+	p, err := os.FindProcess(os.Getpid())
+	if err != nil {
+		return err
+	}
+
+	return p.Signal(syscall.SIGTERM)
+}

internal/proxy/multi_app_conn_test.go

@@ -0,0 +1,94 @@
+package proxy
+
+import (
+	"errors"
+	"os"
+	"os/signal"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+
+	abciclient "github.com/tendermint/tendermint/abci/client"
+	abcimocks "github.com/tendermint/tendermint/abci/client/mocks"
+)
+
+func TestAppConns_Start_Stop(t *testing.T) {
+	quitCh := make(<-chan struct{})
+
+	clientMock := &abcimocks.Client{}
+	clientMock.On("SetLogger", mock.Anything).Return().Times(4)
+	clientMock.On("Start").Return(nil).Times(4)
+	clientMock.On("Stop").Return(nil).Times(4)
+	clientMock.On("Quit").Return(quitCh).Times(4)
+
+	creatorCallCount := 0
+	creator := func() (abciclient.Client, error) {
+		creatorCallCount++
+		return clientMock, nil
+	}
+
+	appConns := NewAppConns(creator)
+
+	err := appConns.Start()
+	require.NoError(t, err)
+
+	time.Sleep(100 * time.Millisecond)
+
+	err = appConns.Stop()
+	require.NoError(t, err)
+
+	clientMock.AssertExpectations(t)
+	assert.Equal(t, 4, creatorCallCount)
+}
+
+// Upon failure, we call tmos.Kill
+func TestAppConns_Failure(t *testing.T) {
+	ok := make(chan struct{})
+	c := make(chan os.Signal, 1)
+	signal.Notify(c, syscall.SIGTERM)
+	go func() {
+		for range c {
+			close(ok)
+		}
+	}()
+
+	quitCh := make(chan struct{})
+	var recvQuitCh <-chan struct{} // nolint:gosimple
+	recvQuitCh = quitCh
+
+	clientMock := &abcimocks.Client{}
+	clientMock.On("SetLogger", mock.Anything).Return()
+	clientMock.On("Start").Return(nil)
+	clientMock.On("Stop").Return(nil)
+
+	clientMock.On("Quit").Return(recvQuitCh)
+	clientMock.On("Error").Return(errors.New("EOF")).Once()
+
+	creator := func() (abciclient.Client, error) {
+		return clientMock, nil
+	}
+
+	appConns := NewAppConns(creator)
+
+	err := appConns.Start()
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		if err := appConns.Stop(); err != nil {
+			t.Error(err)
+		}
+	})
+
+	// simulate failure
+	close(quitCh)
+
+	select {
+	case <-ok:
+		t.Log("SIGTERM successfully received")
+	case <-time.After(5 * time.Second):
+		t.Fatal("expected process to receive SIGTERM signal")
+	}
+}

internal/proxy/version.go

@@ -0,0 +1,16 @@
+package proxy
+
+import (
+	abci "github.com/tendermint/tendermint/abci/types"
+	"github.com/tendermint/tendermint/version"
+)
+
+// RequestInfo contains all the information for sending
+// the abci.RequestInfo message during handshake with the app.
+// It contains only compile-time version information.
+var RequestInfo = abci.RequestInfo{
+	Version:      version.TMVersion,
+	BlockVersion: version.BlockProtocol,
+	P2PVersion:   version.P2PProtocol,
+	AbciVersion:  version.ABCIVersion,
+}

internal/statesync/reactor.go

@@ -14,12 +14,12 @@ import (
 	"github.com/tendermint/tendermint/config"
 	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
 	"github.com/tendermint/tendermint/internal/p2p"
+	"github.com/tendermint/tendermint/internal/proxy"
 	"github.com/tendermint/tendermint/libs/log"
 	"github.com/tendermint/tendermint/libs/service"
 	"github.com/tendermint/tendermint/light"
 	"github.com/tendermint/tendermint/light/provider"
 	ssproto "github.com/tendermint/tendermint/proto/tendermint/statesync"
-	"github.com/tendermint/tendermint/proxy"
 	sm "github.com/tendermint/tendermint/state"
 	"github.com/tendermint/tendermint/store"
 	"github.com/tendermint/tendermint/types"

internal/statesync/reactor_test.go

@@ -16,14 +16,14 @@ import (
 	abci "github.com/tendermint/tendermint/abci/types"
 	"github.com/tendermint/tendermint/config"
 	"github.com/tendermint/tendermint/internal/p2p"
+	"github.com/tendermint/tendermint/internal/proxy"
+	proxymocks "github.com/tendermint/tendermint/internal/proxy/mocks"
 	"github.com/tendermint/tendermint/internal/statesync/mocks"
 	"github.com/tendermint/tendermint/internal/test/factory"
 	"github.com/tendermint/tendermint/libs/log"
 	"github.com/tendermint/tendermint/light/provider"
 	ssproto "github.com/tendermint/tendermint/proto/tendermint/statesync"
 	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
-	"github.com/tendermint/tendermint/proxy"
-	proxymocks "github.com/tendermint/tendermint/proxy/mocks"
 	smmocks "github.com/tendermint/tendermint/state/mocks"
 	"github.com/tendermint/tendermint/store"
 	"github.com/tendermint/tendermint/types"

internal/statesync/syncer.go

@@ -11,10 +11,10 @@ import (
 	"github.com/tendermint/tendermint/config"
 	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
 	"github.com/tendermint/tendermint/internal/p2p"
+	"github.com/tendermint/tendermint/internal/proxy"
 	"github.com/tendermint/tendermint/libs/log"
 	"github.com/tendermint/tendermint/light"
 	ssproto "github.com/tendermint/tendermint/proto/tendermint/statesync"
-	"github.com/tendermint/tendermint/proxy"
 	sm "github.com/tendermint/tendermint/state"
 	"github.com/tendermint/tendermint/types"
 )

internal/statesync/syncer_test.go

@@ -13,10 +13,10 @@ import (
 
 	abci "github.com/tendermint/tendermint/abci/types"
 	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
+	"github.com/tendermint/tendermint/internal/proxy"
+	proxymocks "github.com/tendermint/tendermint/internal/proxy/mocks"
 	"github.com/tendermint/tendermint/internal/statesync/mocks"
 	ssproto "github.com/tendermint/tendermint/proto/tendermint/statesync"
-	"github.com/tendermint/tendermint/proxy"
-	proxymocks "github.com/tendermint/tendermint/proxy/mocks"
 	sm "github.com/tendermint/tendermint/state"
 	"github.com/tendermint/tendermint/types"
 	"github.com/tendermint/tendermint/version"