fix lotus accessor startup

markets/dagstore/lotusaccessor.go

@@ -3,8 +3,8 @@ package dagstore
 import (
 	"context"
 	"io"
-	"sync"
 
+	"github.com/filecoin-project/go-fil-markets/shared"
 	"github.com/ipfs/go-cid"
 	"golang.org/x/xerrors"
 
@@ -15,15 +15,14 @@ import (
 type LotusAccessor interface {
 	FetchUnsealedPiece(ctx context.Context, pieceCid cid.Cid) (io.ReadCloser, error)
 	GetUnpaddedCARSize(ctx context.Context, pieceCid cid.Cid) (uint64, error)
+	Start(ctx context.Context) error
 }
 
 type lotusAccessor struct {
 	pieceStore piecestore.PieceStore
 	rm         retrievalmarket.RetrievalProviderNode
 
-	startLk  sync.Mutex
+	readyMgr *shared.ReadyManager
-	started  bool
-	startErr error
 }
 
 var _ LotusAccessor = (*lotusAccessor)(nil)
@@ -32,17 +31,11 @@ func NewLotusAccessor(store piecestore.PieceStore, rm retrievalmarket.RetrievalP
 	return &lotusAccessor{
 		pieceStore: store,
 		rm:         rm,
+		readyMgr:   shared.NewReadyManager(),
 	}
 }
 
-func (m *lotusAccessor) start(ctx context.Context) error {
+func (m *lotusAccessor) Start(ctx context.Context) error {
-	m.startLk.Lock()
-	defer m.startLk.Unlock()
-
-	if m.started {
-		return m.startErr
-	}
-
 	// Wait for the piece store to startup
 	ready := make(chan error)
 	m.pieceStore.OnReady(func(err error) {
@@ -54,13 +47,17 @@ func (m *lotusAccessor) start(ctx context.Context) error {
 
 	select {
 	case <-ctx.Done():
-		return xerrors.Errorf("context cancelled waiting for piece store startup: %w", ctx.Err())
+		err := xerrors.Errorf("context cancelled waiting for piece store startup: %w", ctx.Err())
-	case err := <-ready:
+		if err := m.readyMgr.FireReady(err); err != nil {
-		// Piece store has started up, check if there was an error
+			log.Warnw("failed to pubish ready event", "err", err)
-		if err != nil {
-			m.startErr = err
-			return err
 		}
+		return err
+
+	case err := <-ready:
+		if err := m.readyMgr.FireReady(err); err != nil {
+			log.Warnw("failed to pubish ready event", "err", err)
+		}
+		return err
 	}
 
 	// Piece store has started up successfully
@@ -68,7 +65,7 @@ func (m *lotusAccessor) start(ctx context.Context) error {
 }
 
 func (m *lotusAccessor) FetchUnsealedPiece(ctx context.Context, pieceCid cid.Cid) (io.ReadCloser, error) {
-	err := m.start(ctx)
+	err := m.readyMgr.AwaitReady()
 	if err != nil {
 		return nil, err
 	}
@@ -118,7 +115,7 @@ func (m *lotusAccessor) FetchUnsealedPiece(ctx context.Context, pieceCid cid.Cid
 }
 
 func (m *lotusAccessor) GetUnpaddedCARSize(ctx context.Context, pieceCid cid.Cid) (uint64, error) {
-	err := m.start(ctx)
+	err := m.readyMgr.AwaitReady()
 	if err != nil {
 		return 0, err
 	}

markets/dagstore/lotusaccessor_test.go

@@ -66,6 +66,7 @@ func TestLotusAccessorFetchUnsealedPiece(t *testing.T) {
 				sectors: mockData,
 			}
 			api := NewLotusAccessor(ps, rpn)
+			require.NoError(t, api.Start(ctx))
 
 			// Add deals to piece store
 			for _, sectorID := range tc.deals {
@@ -101,6 +102,7 @@ func TestLotusAccessorGetUnpaddedCARSize(t *testing.T) {
 	ps := getPieceStore(t)
 	rpn := &mockRPN{}
 	api := NewLotusAccessor(ps, rpn)
+	require.NoError(t, api.Start(ctx))
 
 	// Add a deal with data Length 10
 	dealInfo := piecestore.DealInfo{
@@ -121,14 +123,6 @@ func getPieceStore(t *testing.T) piecestore.PieceStore {
 
 	err = ps.Start(context.Background())
 	require.NoError(t, err)
-	//
-	//ready := make(chan error)
-	//ps.OnReady(func(err error) {
-	//	ready <- err
-	//})
-	//err = <-ready
-	//require.NoError(t, err)
-	//
 	return ps
 }
 

markets/dagstore/mocks/mock_lotus_accessor.go

@@ -31,6 +31,10 @@ func NewMockLotusAccessor(ctrl *gomock.Controller) *MockLotusAccessor {
 	return mock
 }
 
+func (mr *MockLotusAccessor) Start(_ context.Context) error {
+	return nil
+}
+
 // EXPECT returns an object that allows the caller to indicate expected use.
 func (m *MockLotusAccessor) EXPECT() *MockLotusAccessorMockRecorder {
 	return m.recorder

node/builder_miner.go

@@ -147,6 +147,7 @@ func ConfigStorageMiner(c interface{}) Option {
 			Override(new(dtypes.RetrievalPricingFunc), modules.RetrievalPricingFunc(cfg.Dealmaking)),
 
 			// DAG Store
+			Override(new(dagstore.LotusAccessor), modules.NewLotusAccessor),
 			Override(new(*dagstore.Wrapper), modules.DagStoreWrapper),
 
 			// Markets (retrieval)

node/modules/storageminer.go

@@ -576,12 +576,28 @@ func BasicDealFilter(user dtypes.StorageDealFilter) func(onlineOk dtypes.Conside
 	}
 }
 
+func NewLotusAccessor(lc fx.Lifecycle,
+	pieceStore dtypes.ProviderPieceStore,
+	rpn retrievalmarket.RetrievalProviderNode,
+) (dagstore.LotusAccessor, error) {
+	mountApi := dagstore.NewLotusAccessor(pieceStore, rpn)
+	lc.Append(fx.Hook{
+		OnStart: func(ctx context.Context) error {
+			return mountApi.Start(ctx)
+		},
+		OnStop: func(context.Context) error {
+			return nil
+		},
+	})
+
+	return mountApi, nil
+}
+
 func DagStoreWrapper(
 	lc fx.Lifecycle,
 	ds dtypes.MetadataDS,
 	r repo.LockedRepo,
-	pieceStore dtypes.ProviderPieceStore,
+	lotusAccessor dagstore.LotusAccessor,
-	rpn retrievalmarket.RetrievalProviderNode,
 ) (*dagstore.Wrapper, error) {
 	dagStoreDir := filepath.Join(r.Path(), dagStore)
 	dagStoreDS := namespace.Wrap(ds, datastore.NewKey("/dagstore/provider"))
@@ -591,8 +607,8 @@ func DagStoreWrapper(
 		Datastore:     dagStoreDS,
 		GCInterval:    5 * time.Minute,
 	}
-	mountApi := dagstore.NewLotusAccessor(pieceStore, rpn)
+
-	dsw, err := dagstore.NewDagStoreWrapper(cfg, mountApi)
+	dsw, err := dagstore.NewDagStoreWrapper(cfg, lotusAccessor)
 	if err != nil {
 		return nil, xerrors.Errorf("failed to create DAG store wrapper: %w", err)
 	}