client retrieval logging

node/impl/client/client.go

@@ -35,11 +35,13 @@ import (
 	"github.com/filecoin-project/go-fil-markets/storagemarket"
 	"github.com/filecoin-project/go-multistore"
 	"github.com/filecoin-project/go-padreader"
-	"github.com/filecoin-project/sector-storage/ffiwrapper"
 	"github.com/filecoin-project/specs-actors/actors/abi"
 	"github.com/filecoin-project/specs-actors/actors/abi/big"
 	"github.com/filecoin-project/specs-actors/actors/builtin/miner"
 
+	marketevents "github.com/filecoin-project/lotus/markets/loggers"
+	"github.com/filecoin-project/sector-storage/ffiwrapper"
+
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/chain/store"
@@ -398,11 +400,24 @@ func (a *API) ClientListImports(ctx context.Context) ([]api.Import, error) {
 	return out, nil
 }
 
-func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref *api.FileRef) error {
+func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref *api.FileRef) (<-chan marketevents.RetrievalEvent, error) {
+	events := make(chan marketevents.RetrievalEvent)
+	go a.clientRetrieve(ctx, order, ref, events)
+	return events, nil
+}
+
+func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref *api.FileRef, events chan marketevents.RetrievalEvent) {
+	defer close(events)
+
+	finish := func(e error) {
+		events <- marketevents.RetrievalEvent{Err: e}
+	}
+
 	if order.MinerPeer.ID == "" {
 		mi, err := a.StateMinerInfo(ctx, order.Miner, types.EmptyTSK)
 		if err != nil {
-			return err
+			finish(err)
+			return
 		}
 
 		order.MinerPeer = retrievalmarket.RetrievalPeer{
@@ -412,7 +427,8 @@ func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
 	}
 
 	if order.Size == 0 {
-		return xerrors.Errorf("cannot make retrieval deal for zero bytes")
+		finish(xerrors.Errorf("cannot make retrieval deal for zero bytes"))
+		return
 	}
 
 	/*id, st, err := a.imgr().NewStore()
@@ -427,6 +443,14 @@ func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
 
 	unsubscribe := a.Retrieval.SubscribeToEvents(func(event rm.ClientEvent, state rm.ClientDealState) {
 		if state.PayloadCID.Equals(order.Root) {
+
+			events <- marketevents.RetrievalEvent{
+				Event:         event,
+				Status:        state.Status,
+				BytesReceived: state.TotalReceived,
+				FundsSpent:    state.FundsSpent,
+			}
+
 			switch state.Status {
 			case rm.DealStatusCompleted:
 				retrievalResult <- nil
@@ -444,12 +468,14 @@ func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
 
 	params, err := rm.NewParamsV1(ppb, order.PaymentInterval, order.PaymentIntervalIncrease, shared.AllSelector(), order.Piece, order.UnsealPrice)
 	if err != nil {
-		return xerrors.Errorf("Error in retrieval params: %s", err)
+		finish(xerrors.Errorf("Error in retrieval params: %s", err))
+		return
 	}
 
 	store, err := a.RetrievalStoreMgr.NewStore()
 	if err != nil {
-		return xerrors.Errorf("Error setting up new store: %w", err)
+		finish(xerrors.Errorf("Error setting up new store: %w", err))
+		return
 	}
 
 	defer func() {
@@ -467,14 +493,18 @@ func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
 		store.StoreID())
 
 	if err != nil {
-		return xerrors.Errorf("Retrieve failed: %w", err)
+		finish(xerrors.Errorf("Retrieve failed: %w", err))
+		return
 	}
+
 	select {
 	case <-ctx.Done():
-		return xerrors.New("Retrieval Timed Out")
+		finish(xerrors.New("Retrieval Timed Out"))
+		return
 	case err := <-retrievalResult:
 		if err != nil {
-			return xerrors.Errorf("Retrieve: %w", err)
+			finish(xerrors.Errorf("Retrieve: %w", err))
+			return
 		}
 	}
 
@@ -482,7 +512,8 @@ func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
 
 	// If ref is nil, it only fetches the data into the configured blockstore.
 	if ref == nil {
-		return nil
+		finish(nil)
+		return
 	}
 
 	rdag := store.DAGService()
@@ -490,24 +521,30 @@ func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
 	if ref.IsCAR {
 		f, err := os.OpenFile(ref.Path, os.O_CREATE|os.O_WRONLY, 0644)
 		if err != nil {
-			return err
+			finish(err)
+			return
 		}
 		err = car.WriteCar(ctx, rdag, []cid.Cid{order.Root}, f)
 		if err != nil {
-			return err
+			finish(err)
+			return
 		}
-		return f.Close()
+		finish(f.Close())
+		return
 	}
 
 	nd, err := rdag.Get(ctx, order.Root)
 	if err != nil {
-		return xerrors.Errorf("ClientRetrieve: %w", err)
+		finish(xerrors.Errorf("ClientRetrieve: %w", err))
+		return
 	}
 	file, err := unixfile.NewUnixfsFile(ctx, rdag, nd)
 	if err != nil {
-		return xerrors.Errorf("ClientRetrieve: %w", err)
+		finish(xerrors.Errorf("ClientRetrieve: %w", err))
+		return
 	}
-	return files.WriteTo(file, ref.Path)
+	finish(files.WriteTo(file, ref.Path))
+	return
 }
 
 func (a *API) ClientQueryAsk(ctx context.Context, p peer.ID, miner address.Address) (*storagemarket.SignedStorageAsk, error) {