lotus/cmd/lotus-miner/market.go

1050 lines
26 KiB
Go
Raw Normal View History

package main
import (
"bufio"
"context"
2021-09-10 10:26:31 +00:00
"encoding/json"
2020-10-22 20:40:26 +00:00
"errors"
"fmt"
"io"
2020-06-17 00:18:54 +00:00
"os"
"path/filepath"
2020-08-15 00:00:31 +00:00
"sort"
"strconv"
2020-06-17 00:18:54 +00:00
"text/tabwriter"
"time"
tm "github.com/buger/goterm"
"github.com/docker/go-units"
"github.com/fatih/color"
"github.com/ipfs/go-cid"
"github.com/ipfs/go-cidutil/cidenc"
2020-10-22 20:40:26 +00:00
"github.com/libp2p/go-libp2p-core/peer"
"github.com/multiformats/go-multibase"
"github.com/urfave/cli/v2"
"golang.org/x/xerrors"
cborutil "github.com/filecoin-project/go-cbor-util"
datatransfer "github.com/filecoin-project/go-data-transfer"
2020-06-17 00:18:54 +00:00
"github.com/filecoin-project/go-fil-markets/storagemarket"
2020-09-07 03:49:10 +00:00
"github.com/filecoin-project/go-state-types/abi"
2020-06-17 00:18:54 +00:00
2021-09-10 10:21:02 +00:00
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/types"
lcli "github.com/filecoin-project/lotus/cli"
)
var CidBaseFlag = cli.StringFlag{
Name: "cid-base",
Hidden: true,
Value: "base32",
Usage: "Multibase encoding used for version 1 CIDs in output.",
DefaultText: "base32",
}
// GetCidEncoder returns an encoder using the `cid-base` flag if provided, or
// the default (Base32) encoder if not.
func GetCidEncoder(cctx *cli.Context) (cidenc.Encoder, error) {
val := cctx.String("cid-base")
e := cidenc.Encoder{Base: multibase.MustNewEncoder(multibase.Base32)}
if val != "" {
var err error
e.Base, err = multibase.EncoderByName(val)
if err != nil {
return e, err
}
}
return e, nil
}
var storageDealSelectionCmd = &cli.Command{
Name: "selection",
Usage: "Configure acceptance criteria for storage deal proposals",
Subcommands: []*cli.Command{
storageDealSelectionShowCmd,
storageDealSelectionResetCmd,
storageDealSelectionRejectCmd,
},
}
var storageDealSelectionShowCmd = &cli.Command{
Name: "list",
Usage: "List storage deal proposal selection criteria",
Action: func(cctx *cli.Context) error {
smapi, closer, err := lcli.GetMarketsAPI(cctx)
if err != nil {
return err
}
defer closer()
onlineOk, err := smapi.DealsConsiderOnlineStorageDeals(lcli.DaemonContext(cctx))
if err != nil {
return err
}
offlineOk, err := smapi.DealsConsiderOfflineStorageDeals(lcli.DaemonContext(cctx))
if err != nil {
return err
}
fmt.Printf("considering online storage deals: %t\n", onlineOk)
fmt.Printf("considering offline storage deals: %t\n", offlineOk)
return nil
},
}
var storageDealSelectionResetCmd = &cli.Command{
Name: "reset",
Usage: "Reset storage deal proposal selection criteria to default values",
Action: func(cctx *cli.Context) error {
smapi, closer, err := lcli.GetMarketsAPI(cctx)
if err != nil {
return err
}
defer closer()
err = smapi.DealsSetConsiderOnlineStorageDeals(lcli.DaemonContext(cctx), true)
if err != nil {
return err
}
err = smapi.DealsSetConsiderOfflineStorageDeals(lcli.DaemonContext(cctx), true)
if err != nil {
return err
}
err = smapi.DealsSetConsiderVerifiedStorageDeals(lcli.DaemonContext(cctx), true)
if err != nil {
return err
}
err = smapi.DealsSetConsiderUnverifiedStorageDeals(lcli.DaemonContext(cctx), true)
if err != nil {
return err
}
return nil
},
}
var storageDealSelectionRejectCmd = &cli.Command{
Name: "reject",
Usage: "Configure criteria which necessitate automatic rejection",
Flags: []cli.Flag{
&cli.BoolFlag{
Name: "online",
},
&cli.BoolFlag{
Name: "offline",
},
&cli.BoolFlag{
Name: "verified",
},
&cli.BoolFlag{
Name: "unverified",
},
},
Action: func(cctx *cli.Context) error {
smapi, closer, err := lcli.GetMarketsAPI(cctx)
if err != nil {
return err
}
defer closer()
if cctx.Bool("online") {
err = smapi.DealsSetConsiderOnlineStorageDeals(lcli.DaemonContext(cctx), false)
if err != nil {
return err
}
}
if cctx.Bool("offline") {
err = smapi.DealsSetConsiderOfflineStorageDeals(lcli.DaemonContext(cctx), false)
if err != nil {
return err
}
}
if cctx.Bool("verified") {
err = smapi.DealsSetConsiderVerifiedStorageDeals(lcli.DaemonContext(cctx), false)
if err != nil {
return err
}
}
if cctx.Bool("unverified") {
err = smapi.DealsSetConsiderUnverifiedStorageDeals(lcli.DaemonContext(cctx), false)
if err != nil {
return err
}
}
return nil
},
}
var setAskCmd = &cli.Command{
Name: "set-ask",
Usage: "Configure the miner's ask",
Flags: []cli.Flag{
2020-10-07 17:35:57 +00:00
&cli.StringFlag{
Name: "price",
2020-10-07 17:35:57 +00:00
Usage: "Set the price of the ask for unverified deals (specified as FIL / GiB / Epoch) to `PRICE`.",
2020-07-31 19:14:48 +00:00
Required: true,
},
2020-10-07 17:35:57 +00:00
&cli.StringFlag{
2020-07-31 19:14:48 +00:00
Name: "verified-price",
2020-10-07 17:35:57 +00:00
Usage: "Set the price of the ask for verified deals (specified as FIL / GiB / Epoch) to `PRICE`",
Required: true,
2020-07-09 10:36:22 +00:00
},
&cli.StringFlag{
Name: "min-piece-size",
Usage: "Set minimum piece size (w/bit-padding, in bytes) in ask to `SIZE`",
DefaultText: "256B",
Value: "256B",
},
&cli.StringFlag{
Name: "max-piece-size",
Usage: "Set maximum piece size (w/bit-padding, in bytes) in ask to `SIZE`",
DefaultText: "miner sector size",
Value: "0",
},
},
Action: func(cctx *cli.Context) error {
ctx := lcli.DaemonContext(cctx)
minerApi, closer, err := lcli.GetStorageMinerAPI(cctx)
if err != nil {
return err
}
defer closer()
marketsApi, closer, err := lcli.GetMarketsAPI(cctx)
if err != nil {
return err
}
defer closer()
2020-10-07 17:35:57 +00:00
pri, err := types.ParseFIL(cctx.String("price"))
if err != nil {
return err
}
vpri, err := types.ParseFIL(cctx.String("verified-price"))
if err != nil {
return err
}
dur, err := time.ParseDuration("720h0m0s")
if err != nil {
return xerrors.Errorf("cannot parse duration: %w", err)
}
qty := dur.Seconds() / float64(build.BlockDelaySecs)
min, err := units.RAMInBytes(cctx.String("min-piece-size"))
if err != nil {
return xerrors.Errorf("cannot parse min-piece-size to quantity of bytes: %w", err)
}
if min < 256 {
return xerrors.New("minimum piece size (w/bit-padding) is 256B")
2020-06-17 00:45:11 +00:00
}
max, err := units.RAMInBytes(cctx.String("max-piece-size"))
if err != nil {
return xerrors.Errorf("cannot parse max-piece-size to quantity of bytes: %w", err)
}
maddr, err := minerApi.ActorAddress(ctx)
2020-06-17 00:45:11 +00:00
if err != nil {
return err
}
ssize, err := minerApi.ActorSectorSize(ctx, maddr)
2020-06-17 00:45:11 +00:00
if err != nil {
return err
}
smax := int64(ssize)
2020-06-17 00:45:11 +00:00
if max == 0 {
max = smax
}
2020-06-17 00:45:11 +00:00
if max > smax {
2020-06-17 15:57:18 +00:00
return xerrors.Errorf("max piece size (w/bit-padding) %s cannot exceed miner sector size %s", types.SizeStr(types.NewInt(uint64(max))), types.SizeStr(types.NewInt(uint64(smax))))
}
return marketsApi.MarketSetAsk(ctx, types.BigInt(pri), types.BigInt(vpri), abi.ChainEpoch(qty), abi.PaddedPieceSize(min), abi.PaddedPieceSize(max))
},
}
2020-06-17 00:18:54 +00:00
var getAskCmd = &cli.Command{
Name: "get-ask",
Usage: "Print the miner's ask",
Flags: []cli.Flag{},
Action: func(cctx *cli.Context) error {
ctx := lcli.DaemonContext(cctx)
2020-06-17 17:56:42 +00:00
fnapi, closer, err := lcli.GetFullNodeAPI(cctx)
if err != nil {
return err
}
defer closer()
smapi, closer, err := lcli.GetMarketsAPI(cctx)
2020-06-17 00:18:54 +00:00
if err != nil {
return err
}
defer closer()
2020-06-17 17:56:42 +00:00
sask, err := smapi.MarketGetAsk(ctx)
2020-06-17 00:18:54 +00:00
if err != nil {
return err
}
var ask *storagemarket.StorageAsk
if sask != nil && sask.Ask != nil {
ask = sask.Ask
}
w := tabwriter.NewWriter(os.Stdout, 2, 4, 2, ' ', 0)
2020-08-12 11:12:33 +00:00
fmt.Fprintf(w, "Price per GiB/Epoch\tVerified\tMin. Piece Size (padded)\tMax. Piece Size (padded)\tExpiry (Epoch)\tExpiry (Appx. Rem. Time)\tSeq. No.\n")
2020-06-17 00:18:54 +00:00
if ask == nil {
fmt.Fprintf(w, "<miner does not have an ask>\n")
return w.Flush()
2020-06-17 00:18:54 +00:00
}
2020-06-17 17:56:42 +00:00
head, err := fnapi.ChainHead(ctx)
if err != nil {
return err
}
2020-06-17 17:56:42 +00:00
dlt := ask.Expiry - head.Height()
2020-06-17 17:30:16 +00:00
rem := "<expired>"
if dlt > 0 {
rem = (time.Second * time.Duration(int64(dlt)*int64(build.BlockDelaySecs))).String()
2020-06-17 17:30:16 +00:00
}
fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%d\t%s\t%d\n", types.FIL(ask.Price), types.FIL(ask.VerifiedPrice), types.SizeStr(types.NewInt(uint64(ask.MinPieceSize))), types.SizeStr(types.NewInt(uint64(ask.MaxPieceSize))), ask.Expiry, rem, ask.SeqNo)
2020-06-17 00:18:54 +00:00
return w.Flush()
},
}
var storageDealsCmd = &cli.Command{
Name: "storage-deals",
Usage: "Manage storage deals and related configuration",
Subcommands: []*cli.Command{
dealsImportDataCmd,
dealsListCmd,
storageDealSelectionCmd,
setAskCmd,
2020-06-17 00:18:54 +00:00
getAskCmd,
2020-06-18 22:42:24 +00:00
setBlocklistCmd,
getBlocklistCmd,
resetBlocklistCmd,
setSealDurationCmd,
dealsPendingPublish,
dealsRetryPublish,
},
}
var dealsImportDataCmd = &cli.Command{
Name: "import-data",
Usage: "Manually import data for a deal",
ArgsUsage: "<proposal CID> <file>",
Action: func(cctx *cli.Context) error {
api, closer, err := lcli.GetMarketsAPI(cctx)
if err != nil {
return err
}
defer closer()
ctx := lcli.DaemonContext(cctx)
if cctx.Args().Len() < 2 {
return fmt.Errorf("must specify proposal CID and file path")
}
propCid, err := cid.Decode(cctx.Args().Get(0))
if err != nil {
return err
}
fpath := cctx.Args().Get(1)
return api.DealsImportData(ctx, propCid, fpath)
},
}
var dealsListCmd = &cli.Command{
Name: "list",
Usage: "List all deals for this miner",
Flags: []cli.Flag{
2021-09-10 10:26:31 +00:00
&cli.StringFlag{
Name: "format",
2021-09-10 13:18:32 +00:00
Usage: "output format of data, supported: table, json",
2021-09-10 10:26:31 +00:00
Value: "table",
},
&cli.BoolFlag{
Name: "verbose",
Aliases: []string{"v"},
},
&cli.BoolFlag{
Name: "watch",
Usage: "watch deal updates in real-time, rather than a one time list",
},
},
Action: func(cctx *cli.Context) error {
2021-09-10 10:26:31 +00:00
switch cctx.String("format") {
case "table":
return listDealsWithTable(cctx)
case "json":
return listDealsWithJSON(cctx)
}
2021-09-10 10:26:31 +00:00
return fmt.Errorf("unknown format: %s; use `table` or `json`", cctx.String("format"))
},
}
func listDealsWithTable(cctx *cli.Context) error {
api, closer, err := lcli.GetMarketsAPI(cctx)
if err != nil {
return err
}
defer closer()
ctx := lcli.DaemonContext(cctx)
2021-09-10 10:26:31 +00:00
deals, err := api.MarketListIncompleteDeals(ctx)
if err != nil {
return err
}
verbose := cctx.Bool("verbose")
watch := cctx.Bool("watch")
if watch {
updates, err := api.MarketGetDealUpdates(ctx)
if err != nil {
return err
}
2021-09-10 10:26:31 +00:00
for {
tm.Clear()
tm.MoveCursor(1, 1)
2021-09-10 10:26:31 +00:00
err = outputStorageDealsTable(tm.Output, deals, verbose)
if err != nil {
return err
}
2021-09-10 10:26:31 +00:00
tm.Flush()
select {
case <-ctx.Done():
return nil
case updated := <-updates:
var found bool
for i, existing := range deals {
if existing.ProposalCid.Equals(updated.ProposalCid) {
deals[i] = updated
found = true
break
}
}
2021-09-10 10:26:31 +00:00
if !found {
deals = append(deals, updated)
}
}
}
2021-09-10 10:26:31 +00:00
}
2021-09-10 10:26:31 +00:00
return outputStorageDealsTable(os.Stdout, deals, verbose)
}
2021-09-10 10:26:31 +00:00
func outputStorageDealsTable(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())
})
2020-08-15 00:00:31 +00:00
w := tabwriter.NewWriter(out, 2, 4, 2, ' ', 0)
if verbose {
_, _ = fmt.Fprintf(w, "Creation\tVerified\tProposalCid\tDealId\tState\tClient\tSize\tPrice\tDuration\tTransferChannelID\tMessage\n")
} else {
_, _ = fmt.Fprintf(w, "ProposalCid\tDealId\tState\tClient\tSize\tPrice\tDuration\n")
}
for _, deal := range deals {
propcid := deal.ProposalCid.String()
if !verbose {
propcid = "..." + propcid[len(propcid)-8:]
}
fil := types.FIL(types.BigMul(deal.Proposal.StoragePricePerEpoch, types.NewInt(uint64(deal.Proposal.Duration()))))
if verbose {
_, _ = fmt.Fprintf(w, "%s\t%t\t", deal.CreationTime.Time().Format(time.Stamp), deal.Proposal.VerifiedDeal)
}
_, _ = fmt.Fprintf(w, "%s\t%d\t%s\t%s\t%s\t%s\t%s", propcid, deal.DealID, storagemarket.DealStates[deal.State], deal.Proposal.Client, units.BytesSize(float64(deal.Proposal.PieceSize)), fil, deal.Proposal.Duration())
if verbose {
tchid := ""
if deal.TransferChannelId != nil {
tchid = deal.TransferChannelId.String()
}
_, _ = fmt.Fprintf(w, "\t%s", tchid)
_, _ = fmt.Fprintf(w, "\t%s", deal.Message)
}
_, _ = fmt.Fprintln(w)
}
return w.Flush()
}
2020-06-18 22:42:24 +00:00
var getBlocklistCmd = &cli.Command{
Name: "get-blocklist",
Usage: "List the contents of the miner's piece CID blocklist",
Flags: []cli.Flag{
&CidBaseFlag,
},
Action: func(cctx *cli.Context) error {
api, closer, err := lcli.GetMarketsAPI(cctx)
if err != nil {
return err
}
defer closer()
2020-06-18 22:42:24 +00:00
blocklist, err := api.DealsPieceCidBlocklist(lcli.DaemonContext(cctx))
if err != nil {
return err
}
encoder, err := GetCidEncoder(cctx)
if err != nil {
return err
}
2020-06-18 22:42:24 +00:00
for idx := range blocklist {
fmt.Println(encoder.Encode(blocklist[idx]))
}
return nil
},
}
2020-06-18 22:42:24 +00:00
var setBlocklistCmd = &cli.Command{
Name: "set-blocklist",
Usage: "Set the miner's list of blocklisted piece CIDs",
ArgsUsage: "[<path-of-file-containing-newline-delimited-piece-CIDs> (optional, will read from stdin if omitted)]",
Flags: []cli.Flag{},
Action: func(cctx *cli.Context) error {
api, closer, err := lcli.GetMarketsAPI(cctx)
if err != nil {
return err
}
defer closer()
scanner := bufio.NewScanner(os.Stdin)
if cctx.Args().Present() && cctx.Args().First() != "-" {
absPath, err := filepath.Abs(cctx.Args().First())
if err != nil {
return err
}
file, err := os.Open(absPath)
if err != nil {
log.Fatal(err)
}
2020-06-19 16:19:46 +00:00
defer file.Close() //nolint:errcheck
scanner = bufio.NewScanner(file)
}
2020-06-18 22:42:24 +00:00
var blocklist []cid.Cid
for scanner.Scan() {
decoded, err := cid.Decode(scanner.Text())
if err != nil {
return err
}
2020-06-18 22:42:24 +00:00
blocklist = append(blocklist, decoded)
}
err = scanner.Err()
if err != nil {
return err
}
2020-06-18 22:42:24 +00:00
return api.DealsSetPieceCidBlocklist(lcli.DaemonContext(cctx), blocklist)
},
}
2020-06-18 22:42:24 +00:00
var resetBlocklistCmd = &cli.Command{
Name: "reset-blocklist",
Usage: "Remove all entries from the miner's piece CID blocklist",
Flags: []cli.Flag{},
Action: func(cctx *cli.Context) error {
api, closer, err := lcli.GetMarketsAPI(cctx)
if err != nil {
return err
}
defer closer()
2020-06-18 22:42:24 +00:00
return api.DealsSetPieceCidBlocklist(lcli.DaemonContext(cctx), []cid.Cid{})
},
}
var setSealDurationCmd = &cli.Command{
Name: "set-seal-duration",
Usage: "Set the expected time, in minutes, that you expect sealing sectors to take. Deals that start before this duration will be rejected.",
ArgsUsage: "<minutes>",
Action: func(cctx *cli.Context) error {
integrate DAG store and CARv2 in deal-making (#6671) This commit removes badger from the deal-making processes, and moves to a new architecture with the dagstore as the cental component on the miner-side, and CARv2s on the client-side. Every deal that has been handed off to the sealing subsystem becomes a shard in the dagstore. Shards are mounted via the LotusMount, which teaches the dagstore how to load the related piece when serving retrievals. When the miner starts the Lotus for the first time with this patch, we will perform a one-time migration of all active deals into the dagstore. This is a lightweight process, and it consists simply of registering the shards in the dagstore. Shards are backed by the unsealed copy of the piece. This is currently a CARv1. However, the dagstore keeps CARv2 indices for all pieces, so when it's time to acquire a shard to serve a retrieval, the unsealed CARv1 is joined with its index (safeguarded by the dagstore), to form a read-only blockstore, thus taking the place of the monolithic badger. Data transfers have been adjusted to interface directly with CARv2 files. On inbound transfers (client retrievals, miner storage deals), we stream the received data into a CARv2 ReadWrite blockstore. On outbound transfers (client storage deals, miner retrievals), we serve the data off a CARv2 ReadOnly blockstore. Client-side imports are managed by the refactored *imports.Manager component (when not using IPFS integration). Just like it before, we use the go-filestore library to avoid duplicating the data from the original file in the resulting UnixFS DAG (concretely the leaves). However, the target of those imports are what we call "ref-CARv2s": CARv2 files placed under the `$LOTUS_PATH/imports` directory, containing the intermediate nodes in full, and the leaves as positional references to the original file on disk. Client-side retrievals are placed into CARv2 files in the location: `$LOTUS_PATH/retrievals`. A new set of `Dagstore*` JSON-RPC operations and `lotus-miner dagstore` subcommands have been introduced on the miner-side to inspect and manage the dagstore. Despite moving to a CARv2-backed system, the IPFS integration has been respected, and it continues to be possible to make storage deals with data held in an IPFS node, and to perform retrievals directly into an IPFS node. NOTE: because the "staging" and "client" Badger blockstores are no longer used, existing imports on the client will be rendered useless. On startup, Lotus will enumerate all imports and print WARN statements on the log for each import that needs to be reimported. These log lines contain these messages: - import lacks carv2 path; import will not work; please reimport - import has missing/broken carv2; please reimport At the end, we will print a "sanity check completed" message indicating the count of imports found, and how many were deemed broken. Co-authored-by: Aarsh Shah <aarshkshah1992@gmail.com> Co-authored-by: Dirk McCormick <dirkmdev@gmail.com> Co-authored-by: Raúl Kripalani <raul@protocol.ai> Co-authored-by: Dirk McCormick <dirkmdev@gmail.com>
2021-08-16 22:34:32 +00:00
nodeApi, closer, err := lcli.GetMarketsAPI(cctx)
if err != nil {
return err
}
defer closer()
ctx := lcli.ReqContext(cctx)
if cctx.Args().Len() != 1 {
return xerrors.Errorf("must pass duration in minutes")
}
hs, err := strconv.ParseUint(cctx.Args().Get(0), 10, 64)
if err != nil {
return xerrors.Errorf("could not parse duration: %w", err)
}
delay := hs * uint64(time.Minute)
return nodeApi.SectorSetExpectedSealDuration(ctx, time.Duration(delay))
},
}
var dataTransfersCmd = &cli.Command{
Name: "data-transfers",
Usage: "Manage data transfers",
Subcommands: []*cli.Command{
transfersListCmd,
2020-10-22 20:40:26 +00:00
marketRestartTransfer,
marketCancelTransfer,
transfersDiagnosticsCmd,
2020-10-22 20:40:26 +00:00
},
}
var marketRestartTransfer = &cli.Command{
Name: "restart",
Usage: "Force restart a stalled data transfer",
Flags: []cli.Flag{
&cli.StringFlag{
Name: "peerid",
Usage: "narrow to transfer with specific peer",
},
&cli.BoolFlag{
Name: "initiator",
Usage: "specify only transfers where peer is/is not initiator",
Value: false,
},
},
Action: func(cctx *cli.Context) error {
if !cctx.Args().Present() {
return cli.ShowCommandHelp(cctx, cctx.Command.Name)
}
nodeApi, closer, err := lcli.GetMarketsAPI(cctx)
2020-10-22 20:40:26 +00:00
if err != nil {
return err
}
defer closer()
ctx := lcli.ReqContext(cctx)
transferUint, err := strconv.ParseUint(cctx.Args().First(), 10, 64)
if err != nil {
return fmt.Errorf("Error reading transfer ID: %w", err)
}
transferID := datatransfer.TransferID(transferUint)
initiator := cctx.Bool("initiator")
var other peer.ID
if pidstr := cctx.String("peerid"); pidstr != "" {
p, err := peer.Decode(pidstr)
if err != nil {
return err
}
other = p
} else {
channels, err := nodeApi.MarketListDataTransfers(ctx)
if err != nil {
return err
}
found := false
for _, channel := range channels {
if channel.IsInitiator == initiator && channel.TransferID == transferID {
other = channel.OtherPeer
found = true
break
}
}
if !found {
return errors.New("unable to find matching data transfer")
}
}
return nodeApi.MarketRestartDataTransfer(ctx, transferID, other, initiator)
},
}
var marketCancelTransfer = &cli.Command{
Name: "cancel",
Usage: "Force cancel a data transfer",
Flags: []cli.Flag{
&cli.StringFlag{
Name: "peerid",
Usage: "narrow to transfer with specific peer",
},
&cli.BoolFlag{
Name: "initiator",
Usage: "specify only transfers where peer is/is not initiator",
Value: false,
},
&cli.DurationFlag{
Name: "cancel-timeout",
Usage: "time to wait for cancel to be sent to client",
Value: 5 * time.Second,
},
2020-10-22 20:40:26 +00:00
},
Action: func(cctx *cli.Context) error {
if !cctx.Args().Present() {
return cli.ShowCommandHelp(cctx, cctx.Command.Name)
}
nodeApi, closer, err := lcli.GetMarketsAPI(cctx)
2020-10-22 20:40:26 +00:00
if err != nil {
return err
}
defer closer()
ctx := lcli.ReqContext(cctx)
transferUint, err := strconv.ParseUint(cctx.Args().First(), 10, 64)
if err != nil {
return fmt.Errorf("Error reading transfer ID: %w", err)
}
transferID := datatransfer.TransferID(transferUint)
initiator := cctx.Bool("initiator")
var other peer.ID
if pidstr := cctx.String("peerid"); pidstr != "" {
p, err := peer.Decode(pidstr)
if err != nil {
return err
}
other = p
} else {
channels, err := nodeApi.MarketListDataTransfers(ctx)
if err != nil {
return err
}
found := false
for _, channel := range channels {
if channel.IsInitiator == initiator && channel.TransferID == transferID {
other = channel.OtherPeer
found = true
break
}
}
if !found {
return errors.New("unable to find matching data transfer")
}
}
timeoutCtx, cancel := context.WithTimeout(ctx, cctx.Duration("cancel-timeout"))
defer cancel()
return nodeApi.MarketCancelDataTransfer(timeoutCtx, transferID, other, initiator)
},
}
var transfersListCmd = &cli.Command{
Name: "list",
Usage: "List ongoing data transfers for this miner",
Flags: []cli.Flag{
2020-12-28 08:36:19 +00:00
&cli.BoolFlag{
Name: "verbose",
Aliases: []string{"v"},
Usage: "print verbose transfer details",
},
&cli.BoolFlag{
Name: "color",
Usage: "use color in display output",
DefaultText: "depends on output being a TTY",
},
&cli.BoolFlag{
Name: "completed",
Usage: "show completed data transfers",
},
&cli.BoolFlag{
Name: "watch",
Usage: "watch deal updates in real-time, rather than a one time list",
},
2020-10-22 20:40:26 +00:00
&cli.BoolFlag{
Name: "show-failed",
Usage: "show failed/cancelled transfers",
},
},
Action: func(cctx *cli.Context) error {
if cctx.IsSet("color") {
color.NoColor = !cctx.Bool("color")
}
api, closer, err := lcli.GetMarketsAPI(cctx)
if err != nil {
return err
}
defer closer()
ctx := lcli.ReqContext(cctx)
channels, err := api.MarketListDataTransfers(ctx)
if err != nil {
return err
}
2020-12-28 08:36:19 +00:00
verbose := cctx.Bool("verbose")
completed := cctx.Bool("completed")
watch := cctx.Bool("watch")
2020-10-22 20:40:26 +00:00
showFailed := cctx.Bool("show-failed")
if watch {
channelUpdates, err := api.MarketDataTransferUpdates(ctx)
if err != nil {
return err
}
for {
tm.Clear() // Clear current screen
tm.MoveCursor(1, 1)
lcli.OutputDataTransferChannels(tm.Screen, channels, verbose, completed, showFailed)
tm.Flush()
select {
case <-ctx.Done():
return nil
case channelUpdate := <-channelUpdates:
var found bool
for i, existing := range channels {
if existing.TransferID == channelUpdate.TransferID &&
existing.OtherPeer == channelUpdate.OtherPeer &&
existing.IsSender == channelUpdate.IsSender &&
existing.IsInitiator == channelUpdate.IsInitiator {
channels[i] = channelUpdate
found = true
break
}
}
if !found {
channels = append(channels, channelUpdate)
}
}
}
}
lcli.OutputDataTransferChannels(os.Stdout, channels, verbose, completed, showFailed)
return nil
},
}
var transfersDiagnosticsCmd = &cli.Command{
Name: "diagnostics",
Usage: "Get detailed diagnostics on active transfers with a specific peer",
Flags: []cli.Flag{},
Action: func(cctx *cli.Context) error {
if !cctx.Args().Present() {
return cli.ShowCommandHelp(cctx, cctx.Command.Name)
}
api, closer, err := lcli.GetMarketsAPI(cctx)
if err != nil {
return err
}
defer closer()
ctx := lcli.ReqContext(cctx)
2021-12-11 00:49:39 +00:00
targetPeer, err := peer.Decode(cctx.Args().First())
if err != nil {
return err
}
diagnostics, err := api.MarketDataTransferDiagnostics(ctx, targetPeer)
if err != nil {
return err
}
out, err := json.MarshalIndent(diagnostics, "", "\t")
if err != nil {
return err
}
fmt.Println(string(out))
return nil
},
}
var dealsPendingPublish = &cli.Command{
Name: "pending-publish",
Usage: "list deals waiting in publish queue",
Flags: []cli.Flag{
&cli.BoolFlag{
Name: "publish-now",
Usage: "send a publish message now",
},
},
Action: func(cctx *cli.Context) error {
api, closer, err := lcli.GetMarketsAPI(cctx)
if err != nil {
return err
}
defer closer()
ctx := lcli.ReqContext(cctx)
if cctx.Bool("publish-now") {
if err := api.MarketPublishPendingDeals(ctx); err != nil {
return xerrors.Errorf("publishing deals: %w", err)
}
fmt.Println("triggered deal publishing")
return nil
}
pending, err := api.MarketPendingDeals(ctx)
if err != nil {
return xerrors.Errorf("getting pending deals: %w", err)
}
if len(pending.Deals) > 0 {
endsIn := pending.PublishPeriodStart.Add(pending.PublishPeriod).Sub(time.Now())
w := tabwriter.NewWriter(os.Stdout, 2, 4, 2, ' ', 0)
_, _ = fmt.Fprintf(w, "Publish period: %s (ends in %s)\n", pending.PublishPeriod, endsIn.Round(time.Second))
_, _ = fmt.Fprintf(w, "First deal queued at: %s\n", pending.PublishPeriodStart)
_, _ = fmt.Fprintf(w, "Deals will be published at: %s\n", pending.PublishPeriodStart.Add(pending.PublishPeriod))
_, _ = fmt.Fprintf(w, "%d deals queued to be published:\n", len(pending.Deals))
_, _ = fmt.Fprintf(w, "ProposalCID\tClient\tSize\n")
for _, deal := range pending.Deals {
proposalNd, err := cborutil.AsIpld(&deal) // nolint
if err != nil {
return err
}
_, _ = fmt.Fprintf(w, "%s\t%s\t%s\n", proposalNd.Cid(), deal.Proposal.Client, units.BytesSize(float64(deal.Proposal.PieceSize)))
}
return w.Flush()
}
fmt.Println("No deals queued to be published")
return nil
},
}
2021-09-10 10:21:02 +00:00
var dealsRetryPublish = &cli.Command{
Name: "retry-publish",
Usage: "retry publishing a deal",
ArgsUsage: "<proposal CID>",
Action: func(cctx *cli.Context) error {
if !cctx.Args().Present() {
return cli.ShowCommandHelp(cctx, cctx.Command.Name)
}
api, closer, err := lcli.GetMarketsAPI(cctx)
if err != nil {
return err
}
defer closer()
ctx := lcli.ReqContext(cctx)
propcid := cctx.Args().First()
fmt.Printf("retrying deal with proposal-cid: %s\n", propcid)
cid, err := cid.Decode(propcid)
if err != nil {
return err
}
if err := api.MarketRetryPublishDeal(ctx, cid); err != nil {
return xerrors.Errorf("retrying publishing deal: %w", err)
}
fmt.Println("retried to publish deal")
return nil
},
}
2021-09-10 10:26:31 +00:00
func listDealsWithJSON(cctx *cli.Context) error {
node, closer, err := lcli.GetMarketsAPI(cctx)
if err != nil {
return err
}
defer closer()
2021-09-10 10:21:02 +00:00
2021-09-10 10:26:31 +00:00
ctx := lcli.DaemonContext(cctx)
2021-09-10 10:21:02 +00:00
2021-09-10 10:26:31 +00:00
deals, err := node.MarketListIncompleteDeals(ctx)
if err != nil {
return err
}
2021-09-10 10:21:02 +00:00
2021-09-10 10:26:31 +00:00
channels, err := node.MarketListDataTransfers(ctx)
if err != nil {
return err
}
2021-09-10 10:21:02 +00:00
2021-09-10 10:26:31 +00:00
sort.Slice(deals, func(i, j int) bool {
return deals[i].CreationTime.Time().Before(deals[j].CreationTime.Time())
})
2021-09-10 10:21:02 +00:00
2021-09-10 10:26:31 +00:00
channelsByTransferID := map[datatransfer.TransferID]api.DataTransferChannel{}
for _, c := range channels {
channelsByTransferID[c.TransferID] = c
}
2021-09-10 10:21:02 +00:00
2021-09-10 10:26:31 +00:00
w := json.NewEncoder(os.Stdout)
2021-09-10 10:21:02 +00:00
2021-09-10 10:26:31 +00:00
for _, deal := range deals {
val := struct {
DateTime string `json:"datetime"`
VerifiedDeal bool `json:"verified-deal"`
ProposalCID string `json:"proposal-cid"`
DealID abi.DealID `json:"deal-id"`
DealStatus string `json:"deal-status"`
Client string `json:"client"`
PieceSize string `json:"piece-size"`
Price types.FIL `json:"price"`
DurationEpochs abi.ChainEpoch `json:"duration-epochs"`
TransferID *datatransfer.TransferID `json:"transfer-id,omitempty"`
TransferStatus string `json:"transfer-status,omitempty"`
TransferredData string `json:"transferred-data,omitempty"`
}{}
val.DateTime = deal.CreationTime.Time().Format(time.RFC3339)
val.VerifiedDeal = deal.Proposal.VerifiedDeal
val.ProposalCID = deal.ProposalCid.String()
val.DealID = deal.DealID
val.DealStatus = storagemarket.DealStates[deal.State]
val.Client = deal.Proposal.Client.String()
val.PieceSize = units.BytesSize(float64(deal.Proposal.PieceSize))
val.Price = types.FIL(types.BigMul(deal.Proposal.StoragePricePerEpoch, types.NewInt(uint64(deal.Proposal.Duration()))))
val.DurationEpochs = deal.Proposal.Duration()
if deal.TransferChannelId != nil {
if c, ok := channelsByTransferID[deal.TransferChannelId.ID]; ok {
val.TransferID = &c.TransferID
val.TransferStatus = datatransfer.Statuses[c.Status]
val.TransferredData = units.BytesSize(float64(c.Transferred))
2021-09-10 10:21:02 +00:00
}
}
2021-09-10 13:09:37 +00:00
err := w.Encode(val)
if err != nil {
return err
}
2021-09-10 10:26:31 +00:00
}
return nil
2021-09-10 10:21:02 +00:00
}