Add the ipfs dag api object in Blockstore (#356)

* add the ipfs object to the blockstore and remove if from the state object

* fix linter's complaints

* increase TestReactorSelectiveBroadcast sleep times

* increase ensureTimeout to 4 seconds

* only use the dag api object instead of the entire ipfs api object

* increase TestNodeSetPrivValIPC timeout to 400ms

* increase time waited for TestReactorsGossipNoCommittedEvidence again

* increase TestNodeSetPrivValIPC timeout again

* timeout increase

* cleanup remainging mocks

* try insane timeout for TestNodeSetPrivValIPC

* increase the failing precommit timeout

* more cleanup

* remove the unused ipfsAPI from the node

* try a test node that doesn't use the full mocked ipfs node

* implement and use a dag only api provider

* revert crazy timeout

* simplify dag only mock

* remove accidental file

* try to make TestReactorsGossipNoCommittedEvidence less flaky

* use ipld alias instead of format

* remove access to the IPFS dag from the blockstore and add it back to consensus state

* change api provider to only use the dag instead of the core api object

* change alias to ipld in node package

* increase timeouts for TestWALTruncate and timeoutWaitGroup for CI

blockchain/v0/reactor_test.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"
 
+	mdutils "github.com/ipfs/go-merkledag/test"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -71,7 +72,7 @@ func newBlockchainReactor(
 	blockDB := memdb.NewDB()
 	stateDB := memdb.NewDB()
 	stateStore := sm.NewStore(stateDB)
-	blockStore := store.NewBlockStore(blockDB)
+	blockStore := store.NewBlockStore(blockDB, mdutils.Mock())
 
 	state, err := stateStore.LoadFromDBOrGenesisDoc(genDoc)
 	if err != nil {

cmd/tendermint/commands/light.go

@@ -186,12 +186,12 @@ func runProxy(cmd *cobra.Command, args []string) error {
 		cfg.RootDir = dir
 		// TODO(ismail): share badger instance
 		apiProvider := ipfs.Embedded(true, cfg, logger)
-		var coreAPI coreiface.CoreAPI
-		coreAPI, ipfsCloser, err = apiProvider()
+		var dag coreiface.APIDagService
+		dag, ipfsCloser, err = apiProvider()
 		if err != nil {
 			return fmt.Errorf("could not start ipfs API: %w", err)
 		}
-		options = append(options, light.DataAvailabilitySampling(numSamples, coreAPI))
+		options = append(options, light.DataAvailabilitySampling(numSamples, dag))
 	case sequential:
 		options = append(options, light.SequentialVerification())
 	default:

config/config.go

@@ -847,7 +847,7 @@ func TestConsensusConfig() *ConsensusConfig {
 	cfg.TimeoutProposeDelta = 20 * time.Millisecond
 	cfg.TimeoutPrevote = 80 * time.Millisecond
 	cfg.TimeoutPrevoteDelta = 20 * time.Millisecond
-	cfg.TimeoutPrecommit = 80 * time.Millisecond
+	cfg.TimeoutPrecommit = 160 * time.Millisecond
 	cfg.TimeoutPrecommitDelta = 20 * time.Millisecond
 	// NOTE: when modifying, make sure to update time_iota_ms (testGenesisFmt) in toml.go
 	cfg.TimeoutCommit = 80 * time.Millisecond

consensus/byzantine_test.go

@@ -56,7 +56,8 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 		app.InitChain(abci.RequestInitChain{Validators: vals})
 
 		blockDB := memdb.NewDB()
-		blockStore := store.NewBlockStore(blockDB)
+		dag := mdutils.Mock()
+		blockStore := store.NewBlockStore(blockDB, dag)
 
 		// one for mempool, one for consensus
 		mtx := new(tmsync.Mutex)
@@ -78,7 +79,7 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 
 		// Make State
 		blockExec := sm.NewBlockExecutor(stateStore, log.TestingLogger(), proxyAppConnCon, mempool, evpool)
-		cs := NewState(thisConfig.Consensus, state, blockExec, blockStore, mempool, mdutils.Mock(), evpool)
+		cs := NewState(thisConfig.Consensus, state, blockExec, blockStore, mempool, dag, evpool)
 		cs.SetLogger(cs.Logger)
 		// set private validator
 		pv := privVals[i]

consensus/common_test.go

@@ -14,6 +14,7 @@ import (
 	"time"
 
 	"github.com/go-kit/kit/log/term"
+	format "github.com/ipfs/go-ipld-format"
 	mdutils "github.com/ipfs/go-merkledag/test"
 	"github.com/stretchr/testify/require"
 
@@ -52,7 +53,7 @@ type cleanupFunc func()
 var (
 	config                *cfg.Config // NOTE: must be reset for each _test.go file
 	consensusReplayConfig *cfg.Config
-	ensureTimeout         = 2 * time.Second
+	ensureTimeout         = 4 * time.Second
 )
 
 func ensureDir(dir string, mode os.FileMode) {
@@ -352,19 +353,20 @@ func subscribeToVoter(cs *State, addr []byte) <-chan tmpubsub.Message {
 //-------------------------------------------------------------------------------
 // consensus states
 
-func newState(state sm.State, pv types.PrivValidator, app abci.Application) *State {
+func newState(state sm.State, pv types.PrivValidator, app abci.Application, ipfsDagAPI format.DAGService) *State {
 	config := cfg.ResetTestRoot("consensus_state_test")
-	return newStateWithConfig(config, state, pv, app)
+	return newStateWithConfig(config, state, pv, app, ipfsDagAPI)
 }
 
 func newStateWithConfig(
 	thisConfig *cfg.Config,
 	state sm.State,
 	pv types.PrivValidator,
 	app abci.Application,
+	ipfsDagAPI format.DAGService,
 ) *State {
 	blockDB := memdb.NewDB()
-	return newStateWithConfigAndBlockStore(thisConfig, state, pv, app, blockDB)
+	return newStateWithConfigAndBlockStore(thisConfig, state, pv, app, blockDB, ipfsDagAPI)
 }
 
 func newStateWithConfigAndBlockStore(
@@ -373,9 +375,10 @@ func newStateWithConfigAndBlockStore(
 	pv types.PrivValidator,
 	app abci.Application,
 	blockDB dbm.DB,
+	dag format.DAGService,
 ) *State {
 	// Get BlockStore
-	blockStore := store.NewBlockStore(blockDB)
+	blockStore := store.NewBlockStore(blockDB, dag)
 
 	// one for mempool, one for consensus
 	mtx := new(tmsync.Mutex)
@@ -399,7 +402,7 @@ func newStateWithConfigAndBlockStore(
 	}
 
 	blockExec := sm.NewBlockExecutor(stateStore, log.TestingLogger(), proxyAppConnCon, mempool, evpool)
-	cs := NewState(thisConfig.Consensus, state, blockExec, blockStore, mempool, mdutils.Mock(), evpool)
+	cs := NewState(thisConfig.Consensus, state, blockExec, blockStore, mempool, dag, evpool)
 	cs.SetLogger(log.TestingLogger().With("module", "consensus"))
 	cs.SetPrivValidator(pv)
 
@@ -431,7 +434,7 @@ func randState(nValidators int) (*State, []*validatorStub) {
 
 	vss := make([]*validatorStub, nValidators)
 
-	cs := newState(state, privVals[0], counter.NewApplication(true))
+	cs := newState(state, privVals[0], counter.NewApplication(true), mdutils.Mock())
 
 	for i := 0; i < nValidators; i++ {
 		vss[i] = newValidatorStub(privVals[i], int32(i))
@@ -704,7 +707,7 @@ func randConsensusNet(
 		vals := types.TM2PB.ValidatorUpdates(state.Validators)
 		app.InitChain(abci.RequestInitChain{Validators: vals})
 
-		css[i] = newStateWithConfigAndBlockStore(thisConfig, state, privVals[i], app, stateDB)
+		css[i] = newStateWithConfigAndBlockStore(thisConfig, state, privVals[i], app, stateDB, mdutils.Mock())
 		css[i].SetTimeoutTicker(tickerFunc())
 		css[i].SetLogger(logger.With("validator", i, "module", "consensus"))
 	}
@@ -767,7 +770,7 @@ func randConsensusNetWithPeers(
 		app.InitChain(abci.RequestInitChain{Validators: vals})
 		// sm.SaveState(stateDB,state)	//height 1's validatorsInfo already saved in LoadStateFromDBOrGenesisDoc above
 
-		css[i] = newStateWithConfig(thisConfig, state, privVal, app)
+		css[i] = newStateWithConfig(thisConfig, state, privVal, app, mdutils.Mock())
 		css[i].SetTimeoutTicker(tickerFunc())
 		css[i].SetLogger(logger.With("validator", i, "module", "consensus"))
 	}

consensus/mempool_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	mdutils "github.com/ipfs/go-merkledag/test"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -29,7 +30,7 @@ func TestMempoolNoProgressUntilTxsAvailable(t *testing.T) {
 
 	config.Consensus.CreateEmptyBlocks = false
 	state, privVals := randGenesisState(1, false, 10)
-	cs := newStateWithConfig(config, state, privVals[0], NewCounterApplication())
+	cs := newStateWithConfig(config, state, privVals[0], NewCounterApplication(), mdutils.Mock())
 	assertMempool(cs.txNotifier).EnableTxsAvailable()
 	height, round := cs.Height, cs.Round
 	newBlockCh := subscribe(cs.eventBus, types.EventQueryNewBlock)
@@ -49,7 +50,7 @@ func TestMempoolProgressAfterCreateEmptyBlocksInterval(t *testing.T) {
 
 	config.Consensus.CreateEmptyBlocksInterval = ensureTimeout
 	state, privVals := randGenesisState(1, false, 10)
-	cs := newStateWithConfig(config, state, privVals[0], NewCounterApplication())
+	cs := newStateWithConfig(config, state, privVals[0], NewCounterApplication(), mdutils.Mock())
 
 	assertMempool(cs.txNotifier).EnableTxsAvailable()
 
@@ -67,7 +68,7 @@ func TestMempoolProgressInHigherRound(t *testing.T) {
 
 	config.Consensus.CreateEmptyBlocks = false
 	state, privVals := randGenesisState(1, false, 10)
-	cs := newStateWithConfig(config, state, privVals[0], NewCounterApplication())
+	cs := newStateWithConfig(config, state, privVals[0], NewCounterApplication(), mdutils.Mock())
 
 	assertMempool(cs.txNotifier).EnableTxsAvailable()
 	height, round := cs.Height, cs.Round
@@ -117,7 +118,7 @@ func TestMempoolTxConcurrentWithCommit(t *testing.T) {
 	blockDB := memdb.NewDB()
 	stateStore := sm.NewStore(blockDB)
 
-	cs := newStateWithConfigAndBlockStore(config, state, privVals[0], NewCounterApplication(), blockDB)
+	cs := newStateWithConfigAndBlockStore(config, state, privVals[0], NewCounterApplication(), blockDB, mdutils.Mock())
 	err := stateStore.Save(state)
 	require.NoError(t, err)
 	newBlockHeaderCh := subscribe(cs.eventBus, types.EventQueryNewBlockHeader)
@@ -143,7 +144,7 @@ func TestMempoolRmBadTx(t *testing.T) {
 	blockDB := memdb.NewDB()
 
 	stateStore := sm.NewStore(blockDB)
-	cs := newStateWithConfigAndBlockStore(config, state, privVals[0], app, blockDB)
+	cs := newStateWithConfigAndBlockStore(config, state, privVals[0], app, blockDB, mdutils.Mock())
 	err := stateStore.Save(state)
 	require.NoError(t, err)
 

consensus/reactor_test.go

@@ -155,7 +155,8 @@ func TestReactorWithEvidence(t *testing.T) {
 		// css[i] = newStateWithConfig(thisConfig, state, privVals[i], app)
 
 		blockDB := memdb.NewDB()
-		blockStore := store.NewBlockStore(blockDB)
+		dag := mdutils.Mock()
+		blockStore := store.NewBlockStore(blockDB, dag)
 
 		// one for mempool, one for consensus
 		mtx := new(tmsync.Mutex)
@@ -183,7 +184,7 @@ func TestReactorWithEvidence(t *testing.T) {
 
 		// Make State
 		blockExec := sm.NewBlockExecutor(stateStore, log.TestingLogger(), proxyAppConnCon, mempool, evpool)
-		cs := NewState(thisConfig.Consensus, state, blockExec, blockStore, mempool, mdutils.Mock(), evpool2)
+		cs := NewState(thisConfig.Consensus, state, blockExec, blockStore, mempool, dag, evpool2)
 		cs.SetLogger(log.TestingLogger().With("module", "consensus"))
 		cs.SetPrivValidator(pv)
 
@@ -670,7 +671,7 @@ func timeoutWaitGroup(t *testing.T, n int, f func(int), css []*State) {
 
 	// we're running many nodes in-process, possibly in in a virtual machine,
 	// and spewing debug messages - making a block could take a while,
-	timeout := time.Minute * 4
+	timeout := time.Minute * 8
 
 	select {
 	case <-done:

consensus/replay_file.go

@@ -11,7 +11,6 @@ import (
 	"strings"
 
 	mdutils "github.com/ipfs/go-merkledag/test"
-
 	cfg "github.com/lazyledger/lazyledger-core/config"
 	"github.com/lazyledger/lazyledger-core/libs/db/badgerdb"
 	"github.com/lazyledger/lazyledger-core/libs/log"
@@ -290,7 +289,8 @@ func newConsensusStateForReplay(config cfg.BaseConfig, csConfig *cfg.ConsensusCo
 	if err != nil {
 		tmos.Exit(err.Error())
 	}
-	blockStore := store.NewBlockStore(blockStoreDB)
+	dag := mdutils.Mock()
+	blockStore := store.NewBlockStore(blockStoreDB, dag)
 
 	// Get State
 	stateDB, err := badgerdb.NewDB("state", config.DBDir())
@@ -331,7 +331,7 @@ func newConsensusStateForReplay(config cfg.BaseConfig, csConfig *cfg.ConsensusCo
 	blockExec := sm.NewBlockExecutor(stateStore, log.TestingLogger(), proxyApp.Consensus(), mempool, evpool)
 
 	consensusState := NewState(csConfig, state.Copy(), blockExec,
-		blockStore, mempool, mdutils.Mock(), evpool)
+		blockStore, mempool, dag, evpool)
 
 	consensusState.SetEventBus(eventBus)
 	return consensusState

consensus/replay_test.go

@@ -14,6 +14,7 @@ import (
 	"time"
 
 	"github.com/gogo/protobuf/proto"
+	format "github.com/ipfs/go-ipld-format"
 	mdutils "github.com/ipfs/go-merkledag/test"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -78,6 +79,7 @@ func startNewStateAndWaitForBlock(t *testing.T, consensusReplayConfig *cfg.Confi
 		privValidator,
 		kvstore.NewApplication(),
 		blockDB,
+		mdutils.Mock(),
 	)
 	cs.SetLogger(logger)
 
@@ -130,9 +132,7 @@ func TestWALCrash(t *testing.T) {
 		heightToStop int64
 	}{
 		{"empty block",
-			func(stateDB dbm.DB, cs *State, ctx context.Context) {
-				cs.dag = mdutils.Mock()
-			},
+			func(stateDB dbm.DB, cs *State, ctx context.Context) {},
 			1},
 		{"many non-empty blocks",
 			func(stateDB dbm.DB, cs *State, ctx context.Context) {
@@ -174,6 +174,7 @@ LOOP:
 			privValidator,
 			kvstore.NewApplication(),
 			blockDB,
+			mdutils.Mock(),
 		)
 		cs.SetLogger(logger)
 
@@ -1181,16 +1182,17 @@ func stateAndStore(
 // mock block store
 
 type mockBlockStore struct {
-	config  *cfg.Config
-	params  tmproto.ConsensusParams
-	chain   []*types.Block
-	commits []*types.Commit
-	base    int64
+	config     *cfg.Config
+	params     tmproto.ConsensusParams
+	chain      []*types.Block
+	commits    []*types.Commit
+	base       int64
+	ipfsDagAPI format.DAGService
 }
 
 // TODO: NewBlockStore(db.NewMemDB) ...
 func newMockBlockStore(config *cfg.Config, params tmproto.ConsensusParams) *mockBlockStore {
-	return &mockBlockStore{config, params, nil, nil, 0}
+	return &mockBlockStore{config, params, nil, nil, 0, mdutils.Mock()}
 }
 
 func (bs *mockBlockStore) Height() int64                       { return int64(len(bs.chain)) }

consensus/state_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	mdutils "github.com/ipfs/go-merkledag/test"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -639,7 +640,7 @@ func TestStateLockPOLRelock(t *testing.T) {
 	signAddVotes(cs1, tmproto.PrecommitType, nil, types.PartSetHeader{}, vs2, vs3, vs4)
 
 	// before we timeout to the new round set the new proposal
-	cs2 := newState(cs1.state, vs2, counter.NewApplication(true))
+	cs2 := newState(cs1.state, vs2, counter.NewApplication(true), mdutils.Mock())
 	prop, propBlock := decideProposal(cs2, vs2, vs2.Height, vs2.Round+1)
 	if prop == nil || propBlock == nil {
 		t.Fatal("Failed to create proposal block with vs2")
@@ -825,7 +826,7 @@ func TestStateLockPOLUnlockOnUnknownBlock(t *testing.T) {
 	signAddVotes(cs1, tmproto.PrecommitType, nil, types.PartSetHeader{}, vs2, vs3, vs4)
 
 	// before we timeout to the new round set the new proposal
-	cs2 := newState(cs1.state, vs2, counter.NewApplication(true))
+	cs2 := newState(cs1.state, vs2, counter.NewApplication(true), mdutils.Mock())
 	prop, propBlock := decideProposal(cs2, vs2, vs2.Height, vs2.Round+1)
 	if prop == nil || propBlock == nil {
 		t.Fatal("Failed to create proposal block with vs2")
@@ -869,7 +870,7 @@ func TestStateLockPOLUnlockOnUnknownBlock(t *testing.T) {
 	signAddVotes(cs1, tmproto.PrecommitType, nil, types.PartSetHeader{}, vs2, vs3, vs4)
 
 	// before we timeout to the new round set the new proposal
-	cs3 := newState(cs1.state, vs3, counter.NewApplication(true))
+	cs3 := newState(cs1.state, vs3, counter.NewApplication(true), mdutils.Mock())
 	prop, propBlock = decideProposal(cs3, vs3, vs3.Height, vs3.Round+1)
 	if prop == nil || propBlock == nil {
 		t.Fatal("Failed to create proposal block with vs2")

consensus/wal_test.go

@@ -67,7 +67,7 @@ func TestWALTruncate(t *testing.T) {
 	err = walGenerateNBlocks(t, wal.Group(), 60)
 	require.NoError(t, err)
 
-	time.Sleep(1 * time.Millisecond) // wait groupCheckDuration, make sure RotateFile run
+	time.Sleep(5 * time.Millisecond) // wait groupCheckDuration, make sure RotateFile run
 
 	if err := wal.FlushAndSync(); err != nil {
 		t.Error(err)
@@ -311,8 +311,8 @@ func walGenerateNBlocks(t *testing.T, wr io.Writer, numBlocks int) (err error) {
 	if err = stateStore.Save(state); err != nil {
 		t.Error(err)
 	}
-
-	blockStore := store.NewBlockStore(blockStoreDB)
+	dag := mdutils.Mock()
+	blockStore := store.NewBlockStore(blockStoreDB, dag)
 
 	proxyApp := proxy.NewAppConns(proxy.NewLocalClientCreator(app))
 	proxyApp.SetLogger(logger.With("module", "proxy"))
@@ -339,7 +339,7 @@ func walGenerateNBlocks(t *testing.T, wr io.Writer, numBlocks int) (err error) {
 	evpool := sm.EmptyEvidencePool{}
 	blockExec := sm.NewBlockExecutor(stateStore, log.TestingLogger(), proxyApp.Consensus(), mempool, evpool)
 	require.NoError(t, err)
-	consensusState := NewState(config.Consensus, state.Copy(), blockExec, blockStore, mempool, mdutils.Mock(), evpool)
+	consensusState := NewState(config.Consensus, state.Copy(), blockExec, blockStore, mempool, dag, evpool)
 	consensusState.SetLogger(logger)
 	consensusState.SetEventBus(eventBus)
 	if privValidator != nil && privValidator != (*privval.FilePV)(nil) {

evidence/pool_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	mdutils "github.com/ipfs/go-merkledag/test"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
@@ -395,7 +396,7 @@ func initializeValidatorState(privVal types.PrivValidator, height int64) sm.Stor
 // initializeBlockStore creates a block storage and populates it w/ a dummy
 // block at +height+.
 func initializeBlockStore(db dbm.DB, state sm.State, valAddr []byte) *store.BlockStore {
-	blockStore := store.NewBlockStore(db)
+	blockStore := store.NewBlockStore(db, mdutils.Mock())
 
 	for i := int64(1); i <= state.LastBlockHeight; i++ {
 		lastCommit := makeCommit(i-1, valAddr)