Merge pull request #36 from filecoin-project/fix/retrieval

Fix/retrieval

api/api.go

@@ -144,7 +144,7 @@ type MarketFullNode interface {
 	GetDeals(ctx context.Context, miner address.Address, pageIndex, pageSize int) ([]*types.DealInfo, error)                                             //perm:read
 	GetUnPackedDeals(ctx context.Context, miner address.Address, spec *types.GetDealSpec) ([]*types.DealInfoIncludePath, error)                          //perm:read
 	AssignUnPackedDeals(ctx context.Context, miner address.Address, ssize abi.SectorSize, spec *types.GetDealSpec) ([]*types.DealInfoIncludePath, error) //perm:write
-
+	UpdateStorageDealStatus(ctx context.Context, dealProposalCid cid.Cid, state storagemarket.StorageDealStatus) error                                   //perm:write
 	//market event
 	ResponseMarketEvent(ctx context.Context, resp *types2.ResponseEvent) error                                            //perm:read
 	ListenMarketEvent(ctx context.Context, policy *marketevent.MarketRegisterPolicy) (<-chan *types2.RequestEvent, error) //perm:read

api/clients/modules.go

@@ -208,7 +208,7 @@ func NewIMarketEvent(stream *marketevent.MarketEventStream) (MarketRequestEvent,
 	return stream, nil
 }
 
-var ClientsOpts = func(server bool, mCfg *config.Messager, signerCfg *config.Signer, mysqlCfg *config.Mysql) builder.Option {
+var ClientsOpts = func(server bool, mCfg *config.Messager, signerCfg *config.Signer) builder.Option {
 	opts := builder.Options(
 		builder.ApplyIf(
 			func(s *builder.Settings) bool {

api/impl/impl.go

@@ -2,6 +2,7 @@ package impl
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"github.com/filecoin-project/venus-market/models/repo"
 	"os"
@@ -140,6 +141,19 @@ func (m MarketNodeImpl) MarketListRetrievalDeals(ctx context.Context, mAddr addr
 			}
 		}
 		// todo: 按miner过滤交易
+		out = append(out, *deal)
+	}
+	for _, mm := range deals {
+		xxxx, err := json.Marshal(mm)
+		if err != nil {
+			fmt.Println(err)
+		}
+
+		var dd retrievalmarket.ProviderDealState
+		err = json.Unmarshal(xxxx, &dd)
+		if err != nil {
+			fmt.Println(err)
+		}
 	}
 
 	return out, nil
@@ -175,6 +189,10 @@ func (m MarketNodeImpl) MarketListIncompleteDeals(ctx context.Context, mAddr add
 	return resDeals, nil
 }
 
+func (m MarketNodeImpl) UpdateStorageDealStatus(ctx context.Context, dealProposal cid.Cid, state storagemarket.StorageDealStatus) error {
+	return m.Repo.StorageDealRepo().UpdateDealStatus(dealProposal, state)
+}
+
 func (m MarketNodeImpl) MarketSetAsk(ctx context.Context, mAddr address.Address, price vTypes.BigInt, verifiedPrice vTypes.BigInt, duration abi.ChainEpoch, minPieceSize abi.PaddedPieceSize, maxPieceSize abi.PaddedPieceSize) error {
 	options := []storagemarket.StorageAskOption{
 		storagemarket.MinPieceSize(minPieceSize),

cli/client.go

@@ -1202,6 +1202,12 @@ var clientRetrieveCmd = &cli.Command{
 			IsCAR: cctx.Bool("car"),
 		}
 
+		fmt.Println("Size:", order.Size)
+		fmt.Println("Unseal Price:", order.UnsealPrice)
+		fmt.Println("Total Fee:", order.Total)
+
+		data, _ := json.MarshalIndent(order, " ", "\t")
+		fmt.Println(string(data))
 		updates, err := mapi.ClientRetrieveWithEvents(ctx, *order, ref)
 		if err != nil {
 			return xerrors.Errorf("error setting up retrieval: %w", err)

cli/market.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/venus-market/storageprovider"
 	"io"
 	"log"
 	"os"
@@ -321,6 +322,7 @@ var StorageDealsCmd = &cli.Command{
 	Subcommands: []*cli.Command{
 		dealsImportDataCmd,
 		dealsListCmd,
+		updateStorageDealStateCmd,
 		storageDealSelectionCmd,
 		setAskCmd,
 		getAskCmd,
@@ -440,6 +442,36 @@ var dealsListCmd = &cli.Command{
 	},
 }
 
+var updateStorageDealStateCmd = &cli.Command{
+	Name:  "update",
+	Usage: "update deal status",
+	Flags: []cli.Flag{
+		&cli.StringFlag{
+			Name:     "proposalcid",
+			Required: true,
+		},
+		&cli.StringFlag{
+			Name:     "state",
+			Required: true,
+		},
+	},
+	Action: func(cctx *cli.Context) error {
+		api, closer, err := NewMarketNode(cctx)
+		if err != nil {
+			return err
+		}
+		defer closer()
+
+		ctx := DaemonContext(cctx)
+		proposalCid, err := cid.Decode(cctx.String("proposalcid"))
+		if err != nil {
+			return err
+		}
+		state := storageprovider.StringToStorageState[cctx.String("state")]
+		return api.UpdateStorageDealStatus(ctx, proposalCid, state)
+	},
+}
+
 func outputStorageDeals(out io.Writer, deals []storagemarket.MinerDeal, verbose bool) error {
 	sort.Slice(deals, func(i, j int) bool {
 		return deals[i].CreationTime.Time().Before(deals[j].CreationTime.Time())

cli/retrieval-deals.go

@@ -148,22 +148,23 @@ var retrievalDealsListCmd = &cli.Command{
 		if err != nil {
 			return err
 		}
-
 		w := tabwriter.NewWriter(os.Stdout, 2, 4, 2, ' ', 0)
 
-		_, _ = fmt.Fprintf(w, "Receiver\tDealID\tPayload\tState\tPricePerByte\tBytesSent\tMessage\n")
+		_, _ = fmt.Fprintf(w, "Receiver\tDealID\tPayload\tState\tPricePerByte\tBytesSent\tPaied\tInterval\tMessage\n")
 
 		for _, deal := range deals {
 			payloadCid := deal.PayloadCID.String()
 
 			_, _ = fmt.Fprintf(w,
-				"%s\t%d\t%s\t%s\t%s\t%d\t%s\n",
+				"%s\t%d\t%s\t%s\t%s\t%d\t%d\t%d\t%s\n",
 				deal.Receiver.String(),
 				deal.ID,
 				"..."+payloadCid[len(payloadCid)-8:],
 				retrievalmarket.DealStatuses[deal.Status],
 				deal.PricePerByte.String(),
 				deal.TotalSent,
+				deal.FundsReceived,
+				deal.CurrentInterval,
 				deal.Message,
 			)
 		}
@@ -212,7 +213,10 @@ var retrievalSetAskCmd = &cli.Command{
 
 		ask, err := api.MarketGetRetrievalAsk(ctx, mAddr)
 		if err != nil {
-			return err
+			if err.Error() != "record not found" {
+				return err
+			}
+			ask = &retrievalmarket.Ask{}
 		}
 
 		if cctx.IsSet("price") {

client/modules.go

@@ -78,7 +78,7 @@ func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs network.Gra
 	dtRetryParams := dtnet.RetryParameters(time.Second, 5*time.Minute, 15, 5)
 	net := dtnet.NewFromLibp2pHost(h, dtRetryParams)
 
-	transport := dtgstransport.NewTransport(h.ID(), gs)
+	transport := dtgstransport.NewTransport(h.ID(), gs, net)
 	err := os.MkdirAll(filepath.Join(string(*homeDir), "data-transfer"), 0755) //nolint: gosec
 	if err != nil && !os.IsExist(err) {
 		return nil, err

cmd/market-client/main.go

@@ -158,8 +158,8 @@ func marketClient(cctx *cli.Context) error {
 
 		config.ConfigClientOpts(cfg),
 
-		clients2.ClientsOpts(false, &cfg.Messager, &cfg.Signer, &config.Mysql{}),
-		models.DBOptions(false, &config.Mysql{}),
+		clients2.ClientsOpts(false, &cfg.Messager, &cfg.Signer),
+		models.DBOptions(false, nil),
 		network.NetworkOpts(false, cfg.SimultaneousTransfers),
 		paychmgr.PaychOpts,
 		fundmgr.FundMgrOpts,

cmd/venus-market/main.go

@@ -193,7 +193,7 @@ func daemon(cctx *cli.Context) error {
 		minermgr2.MinerMgrOpts(cfg),
 
 		//clients
-		clients.ClientsOpts(true, &cfg.Messager, &cfg.Signer, &cfg.Mysql),
+		clients.ClientsOpts(true, &cfg.Messager, &cfg.Signer),
 		models.DBOptions(true, &cfg.Mysql),
 		network.NetworkOpts(true, cfg.SimultaneousTransfers),
 		piecestorage.PieceStorageOpts(cfg),
@@ -265,7 +265,7 @@ func flagData(cctx *cli.Context, cfg *config.MarketConfig) error {
 			if err != nil {
 				return xerrors.Errorf("flag provide a wrong address %s %w", addrStr, err)
 			}
-			cfg.MinerAddress = append(cfg.MinerAddress, addr)
+			cfg.MinerAddress = append(cfg.MinerAddress, config.Address(addr))
 		}
 	}
 	return nil

config/config.go

@@ -99,7 +99,7 @@ type RetrievalPricingDefault struct {
 }
 
 type AddressConfig struct {
-	DealPublishControl []address.Address
+	DealPublishControl []Address
 
 	// DisableOwnerFallback disables usage of the owner address for messages
 	// sent automatically
@@ -165,7 +165,8 @@ type MarketConfig struct {
 	AddressConfig AddressConfig
 	DAGStore      DAGStoreConfig
 
-	MinerAddress []address.Address
+	MinerAddress            []Address
+	RetrievalPaymentAddress Address
 	// When enabled, the miner can accept online deals
 	ConsiderOnlineStorageDeals bool
 	// When enabled, the miner can accept offline deals
@@ -266,3 +267,11 @@ func (addr *Address) UnmarshalText(text []byte) error {
 func (dur Address) MarshalText() ([]byte, error) {
 	return []byte(address.Address(dur).String()), nil
 }
+
+func ConvertConfigAddress(addrs []Address) []address.Address {
+	addrs2 := make([]address.Address, len(addrs))
+	for index, addr := range addrs {
+		addrs2[index] = address.Address(addr)
+	}
+	return addrs2
+}

go.mod

@@ -15,7 +15,7 @@ require (
 	github.com/filecoin-project/go-bitfield v0.2.4
 	github.com/filecoin-project/go-cbor-util v0.0.0-20201016124514-d0bbec7bfcc4
 	github.com/filecoin-project/go-commp-utils v0.1.1-0.20210427191551-70bf140d31c7
-	github.com/filecoin-project/go-data-transfer v1.10.1
+	github.com/filecoin-project/go-data-transfer v1.11.4
 	github.com/filecoin-project/go-fil-commcid v0.1.0
 	github.com/filecoin-project/go-fil-commp-hashhash v0.1.0
 	github.com/filecoin-project/go-fil-markets v1.12.0
@@ -50,7 +50,7 @@ require (
 	github.com/ipfs/go-ds-badger2 v0.1.1-0.20200708190120-187fc06f714e
 	github.com/ipfs/go-ds-leveldb v0.4.2
 	github.com/ipfs/go-ds-measure v0.1.0
-	github.com/ipfs/go-graphsync v0.9.1
+	github.com/ipfs/go-graphsync v0.10.0
 	github.com/ipfs/go-ipfs-blockstore v1.0.4
 	github.com/ipfs/go-ipfs-blocksutil v0.0.1
 	github.com/ipfs/go-ipfs-chunker v0.0.5
@@ -63,7 +63,7 @@ require (
 	github.com/ipfs/go-merkledag v0.3.2
 	github.com/ipfs/go-metrics-interface v0.0.1
 	github.com/ipfs/go-unixfs v0.2.6
-	github.com/ipld/go-car v0.3.1-0.20210601190600-f512dac51e8e
+	github.com/ipld/go-car v0.3.2-0.20211001225732-32d0d9933823
 	github.com/ipld/go-car/v2 v2.0.3-0.20210811121346-c514a30114d7
 	github.com/ipld/go-ipld-prime v0.12.3
 	github.com/libp2p/go-buffer-pool v0.0.2
@@ -104,6 +104,7 @@ require (
 
 replace (
 	github.com/filecoin-project/filecoin-ffi => ./extern/filecoin-ffi
+	github.com/filecoin-project/go-fil-markets => /Users/lijunlong/code/go-fil-markets
 	github.com/filecoin-project/go-jsonrpc => github.com/ipfs-force-community/go-jsonrpc v0.1.4-0.20210721095535-a67dff16de21
 	github.com/ipfs/go-ipfs-cmds => github.com/ipfs-force-community/go-ipfs-cmds v0.6.1-0.20210521090123-4587df7fa0ab
 )