Merge pull request #3379 from filecoin-project/feat/add-list-deals-watch-option

Add watch option to list-deals

api/api_full.go

@@ -243,6 +243,8 @@ type FullNode interface {
 	ClientGetDealInfo(context.Context, cid.Cid) (*DealInfo, error)
 	// ClientListDeals returns information about the deals made by the local client.
 	ClientListDeals(ctx context.Context) ([]DealInfo, error)
+	// ClientGetDealUpdates returns the status of updated deals
+	ClientGetDealUpdates(ctx context.Context) (<-chan DealInfo, error)
 	// ClientHasLocal indicates whether a certain CID is locally stored.
 	ClientHasLocal(ctx context.Context, root cid.Cid) (bool, error)
 	// ClientFindData identifies peers that have a certain file, and returns QueryOffers (one per peer).

api/apistruct/struct.go

@@ -139,6 +139,7 @@ type FullNodeStruct struct {
 		ClientStartDeal           func(ctx context.Context, params *api.StartDealParams) (*cid.Cid, error)                                          `perm:"admin"`
 		ClientGetDealInfo         func(context.Context, cid.Cid) (*api.DealInfo, error)                                                             `perm:"read"`
 		ClientListDeals           func(ctx context.Context) ([]api.DealInfo, error)                                                                 `perm:"write"`
+		ClientGetDealUpdates      func(ctx context.Context) (<-chan api.DealInfo, error)                                                            `perm:"read"`
 		ClientRetrieve            func(ctx context.Context, order api.RetrievalOrder, ref *api.FileRef) error                                       `perm:"admin"`
 		ClientRetrieveWithEvents  func(ctx context.Context, order api.RetrievalOrder, ref *api.FileRef) (<-chan marketevents.RetrievalEvent, error) `perm:"admin"`
 		ClientQueryAsk            func(ctx context.Context, p peer.ID, miner address.Address) (*storagemarket.SignedStorageAsk, error)              `perm:"read"`
@@ -433,6 +434,10 @@ func (c *FullNodeStruct) ClientListDeals(ctx context.Context) ([]api.DealInfo, e
 	return c.Internal.ClientListDeals(ctx)
 }
 
+func (c *FullNodeStruct) ClientGetDealUpdates(ctx context.Context) (<-chan api.DealInfo, error) {
+	return c.Internal.ClientGetDealUpdates(ctx)
+}
+
 func (c *FullNodeStruct) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref *api.FileRef) error {
 	return c.Internal.ClientRetrieve(ctx, order, ref)
 }

cli/client.go

@@ -1,6 +1,7 @@
 package cli
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -978,6 +979,10 @@ var clientListDeals = &cli.Command{
 			Usage: "use color in display output",
 			Value: true,
 		},
+		&cli.BoolFlag{
+			Name:  "watch",
+			Usage: "watch deal updates in real-time, rather than a one time list",
+		},
 	},
 	Action: func(cctx *cli.Context) error {
 		api, closer, err := GetFullNodeAPI(cctx)
@@ -987,81 +992,97 @@ var clientListDeals = &cli.Command{
 		defer closer()
 		ctx := ReqContext(cctx)
 
-		head, err := api.ChainHead(ctx)
+		verbose := cctx.Bool("verbose")
-		if err != nil {
+		color := cctx.Bool("color")
-			return err
+		watch := cctx.Bool("watch")
-		}
 
 		localDeals, err := api.ClientListDeals(ctx)
 		if err != nil {
 			return err
 		}
 
-		sort.Slice(localDeals, func(i, j int) bool {
+		if watch {
-			return localDeals[i].CreationTime.Before(localDeals[j].CreationTime)
+			updates, err := api.ClientGetDealUpdates(ctx)
-		})
+			if err != nil {
+				return err
+			}
 
-		var deals []deal
+			for {
-		for _, v := range localDeals {
+				tm.Clear()
-			if v.DealID == 0 {
+				tm.MoveCursor(1, 1)
-				deals = append(deals, deal{
+
-					LocalDeal: v,
+				err = outputStorageDeals(ctx, tm.Screen, api, localDeals, verbose, color)
-					OnChainDealState: market.DealState{
-						SectorStartEpoch: -1,
-						LastUpdatedEpoch: -1,
-						SlashEpoch:       -1,
-					},
-				})
-			} else {
-				onChain, err := api.StateMarketStorageDeal(ctx, v.DealID, head.Key())
 				if err != nil {
-					deals = append(deals, deal{LocalDeal: v})
+					return err
-				} else {
+				}
-					deals = append(deals, deal{
+
-						LocalDeal:        v,
+				tm.Flush()
-						OnChainDealState: onChain.State,
+
-					})
+				select {
+				case <-ctx.Done():
+					return nil
+				case updated := <-updates:
+					var found bool
+					for i, existing := range localDeals {
+						if existing.ProposalCid.Equals(updated.ProposalCid) {
+							localDeals[i] = updated
+							found = true
+							break
+						}
+					}
+					if !found {
+						localDeals = append(localDeals, updated)
+					}
 				}
 			}
 		}
 
-		color := cctx.Bool("color")
+		return outputStorageDeals(ctx, os.Stdout, api, localDeals, cctx.Bool("verbose"), cctx.Bool("color"))
+	},
+}
 
-		if cctx.Bool("verbose") {
+func dealFromDealInfo(ctx context.Context, full api.FullNode, head *types.TipSet, v api.DealInfo) deal {
-			w := tabwriter.NewWriter(os.Stdout, 2, 4, 2, ' ', 0)
+	if v.DealID == 0 {
-			fmt.Fprintf(w, "Created\tDealCid\tDealId\tProvider\tState\tOn Chain?\tSlashed?\tPieceCID\tSize\tPrice\tDuration\tMessage\n")
+		return deal{
-			for _, d := range deals {
+			LocalDeal: v,
-				onChain := "N"
+			OnChainDealState: market.DealState{
-				if d.OnChainDealState.SectorStartEpoch != -1 {
+				SectorStartEpoch: -1,
-					onChain = fmt.Sprintf("Y (epoch %d)", d.OnChainDealState.SectorStartEpoch)
+				LastUpdatedEpoch: -1,
-				}
+				SlashEpoch:       -1,
-
+			},
-				slashed := "N"
-				if d.OnChainDealState.SlashEpoch != -1 {
-					slashed = fmt.Sprintf("Y (epoch %d)", d.OnChainDealState.SlashEpoch)
-				}
-
-				price := types.FIL(types.BigMul(d.LocalDeal.PricePerEpoch, types.NewInt(d.LocalDeal.Duration)))
-				fmt.Fprintf(w, "%s\t%s\t%d\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%d\t%s\n", d.LocalDeal.CreationTime.Format(time.Stamp), d.LocalDeal.ProposalCid, d.LocalDeal.DealID, d.LocalDeal.Provider, dealStateString(color, d.LocalDeal.State), onChain, slashed, d.LocalDeal.PieceCID, types.SizeStr(types.NewInt(d.LocalDeal.Size)), price, d.LocalDeal.Duration, d.LocalDeal.Message)
-			}
-			return w.Flush()
 		}
+	}
 
-		w := tablewriter.New(tablewriter.Col("DealCid"),
+	onChain, err := full.StateMarketStorageDeal(ctx, v.DealID, head.Key())
-			tablewriter.Col("DealId"),
+	if err != nil {
-			tablewriter.Col("Provider"),
+		return deal{LocalDeal: v}
-			tablewriter.Col("State"),
+	}
-			tablewriter.Col("On Chain?"),
-			tablewriter.Col("Slashed?"),
-			tablewriter.Col("PieceCID"),
-			tablewriter.Col("Size"),
-			tablewriter.Col("Price"),
-			tablewriter.Col("Duration"),
-			tablewriter.NewLineCol("Message"))
 
+	return deal{
+		LocalDeal:        v,
+		OnChainDealState: onChain.State,
+	}
+}
+
+func outputStorageDeals(ctx context.Context, out io.Writer, full api.FullNode, localDeals []api.DealInfo, verbose bool, color bool) error {
+	sort.Slice(localDeals, func(i, j int) bool {
+		return localDeals[i].CreationTime.Before(localDeals[j].CreationTime)
+	})
+
+	head, err := full.ChainHead(ctx)
+	if err != nil {
+		return err
+	}
+
+	var deals []deal
+	for _, localDeal := range localDeals {
+		deals = append(deals, dealFromDealInfo(ctx, full, head, localDeal))
+	}
+
+	if verbose {
+		w := tabwriter.NewWriter(out, 2, 4, 2, ' ', 0)
+		fmt.Fprintf(w, "Created\tDealCid\tDealId\tProvider\tState\tOn Chain?\tSlashed?\tPieceCID\tSize\tPrice\tDuration\tMessage\n")
 		for _, d := range deals {
-			propcid := ellipsis(d.LocalDeal.ProposalCid.String(), 8)
-
 			onChain := "N"
 			if d.OnChainDealState.SectorStartEpoch != -1 {
 				onChain = fmt.Sprintf("Y (epoch %d)", d.OnChainDealState.SectorStartEpoch)
@@ -1072,27 +1093,57 @@ var clientListDeals = &cli.Command{
 				slashed = fmt.Sprintf("Y (epoch %d)", d.OnChainDealState.SlashEpoch)
 			}
 
-			piece := ellipsis(d.LocalDeal.PieceCID.String(), 8)
-
 			price := types.FIL(types.BigMul(d.LocalDeal.PricePerEpoch, types.NewInt(d.LocalDeal.Duration)))
+			fmt.Fprintf(w, "%s\t%s\t%d\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%d\t%s\n", d.LocalDeal.CreationTime.Format(time.Stamp), d.LocalDeal.ProposalCid, d.LocalDeal.DealID, d.LocalDeal.Provider, dealStateString(color, d.LocalDeal.State), onChain, slashed, d.LocalDeal.PieceCID, types.SizeStr(types.NewInt(d.LocalDeal.Size)), price, d.LocalDeal.Duration, d.LocalDeal.Message)
+		}
+		return w.Flush()
+	}
 
-			w.Write(map[string]interface{}{
+	w := tablewriter.New(tablewriter.Col("DealCid"),
-				"DealCid":   propcid,
+		tablewriter.Col("DealId"),
-				"DealId":    d.LocalDeal.DealID,
+		tablewriter.Col("Provider"),
-				"Provider":  d.LocalDeal.Provider,
+		tablewriter.Col("State"),
-				"State":     dealStateString(color, d.LocalDeal.State),
+		tablewriter.Col("On Chain?"),
-				"On Chain?": onChain,
+		tablewriter.Col("Slashed?"),
-				"Slashed?":  slashed,
+		tablewriter.Col("PieceCID"),
-				"PieceCID":  piece,
+		tablewriter.Col("Size"),
-				"Size":      types.SizeStr(types.NewInt(d.LocalDeal.Size)),
+		tablewriter.Col("Price"),
-				"Price":     price,
+		tablewriter.Col("Duration"),
-				"Duration":  d.LocalDeal.Duration,
+		tablewriter.NewLineCol("Message"))
-				"Message":   d.LocalDeal.Message,
+
-			})
+	for _, d := range deals {
+		propcid := ellipsis(d.LocalDeal.ProposalCid.String(), 8)
+
+		onChain := "N"
+		if d.OnChainDealState.SectorStartEpoch != -1 {
+			onChain = fmt.Sprintf("Y (epoch %d)", d.OnChainDealState.SectorStartEpoch)
 		}
 
-		return w.Flush(os.Stdout)
+		slashed := "N"
-	},
+		if d.OnChainDealState.SlashEpoch != -1 {
+			slashed = fmt.Sprintf("Y (epoch %d)", d.OnChainDealState.SlashEpoch)
+		}
+
+		piece := ellipsis(d.LocalDeal.PieceCID.String(), 8)
+
+		price := types.FIL(types.BigMul(d.LocalDeal.PricePerEpoch, types.NewInt(d.LocalDeal.Duration)))
+
+		w.Write(map[string]interface{}{
+			"DealCid":   propcid,
+			"DealId":    d.LocalDeal.DealID,
+			"Provider":  d.LocalDeal.Provider,
+			"State":     dealStateString(color, d.LocalDeal.State),
+			"On Chain?": onChain,
+			"Slashed?":  slashed,
+			"PieceCID":  piece,
+			"Size":      types.SizeStr(types.NewInt(d.LocalDeal.Size)),
+			"Price":     price,
+			"Duration":  d.LocalDeal.Duration,
+			"Message":   d.LocalDeal.Message,
+		})
+	}
+
+	return w.Flush(out)
 }
 
 func dealStateString(c bool, state storagemarket.StorageDealStatus) string {

documentation/en/api-methods.md

@@ -36,6 +36,7 @@
   * [ClientFindData](#ClientFindData)
   * [ClientGenCar](#ClientGenCar)
   * [ClientGetDealInfo](#ClientGetDealInfo)
+  * [ClientGetDealUpdates](#ClientGetDealUpdates)
   * [ClientHasLocal](#ClientHasLocal)
   * [ClientImport](#ClientImport)
   * [ClientListDataTransfers](#ClientListDataTransfers)
@@ -915,6 +916,42 @@ Response:
 }
 ```
 
+### ClientGetDealUpdates
+ClientGetDealUpdates returns the status of updated deals
+
+
+Perms: read
+
+Inputs: `null`
+
+Response:
+```json
+{
+  "ProposalCid": {
+    "/": "bafy2bzacea3wsdh6y3a36tb3skempjoxqpuyompjbmfeyf34fi3uy6uue42v4"
+  },
+  "State": 42,
+  "Message": "string value",
+  "Provider": "t01234",
+  "DataRef": {
+    "TransferType": "string value",
+    "Root": {
+      "/": "bafy2bzacea3wsdh6y3a36tb3skempjoxqpuyompjbmfeyf34fi3uy6uue42v4"
+    },
+    "PieceCid": null,
+    "PieceSize": 1024
+  },
+  "PieceCID": {
+    "/": "bafy2bzacea3wsdh6y3a36tb3skempjoxqpuyompjbmfeyf34fi3uy6uue42v4"
+  },
+  "Size": 42,
+  "PricePerEpoch": "0",
+  "Duration": 42,
+  "DealID": 5432,
+  "CreationTime": "0001-01-01T00:00:00Z"
+}
+```
+
 ### ClientHasLocal
 ClientHasLocal indicates whether a certain CID is locally stored.
 

node/impl/client/client.go

@@ -223,6 +223,21 @@ func (a *API) ClientGetDealInfo(ctx context.Context, d cid.Cid) (*api.DealInfo,
 	}, nil
 }
 
+func (a *API) ClientGetDealUpdates(ctx context.Context) (<-chan api.DealInfo, error) {
+	updates := make(chan api.DealInfo)
+
+	unsub := a.SMDealClient.SubscribeToEvents(func(_ storagemarket.ClientEvent, deal storagemarket.ClientDeal) {
+		updates <- newDealInfo(deal)
+	})
+
+	go func() {
+		defer unsub()
+		<-ctx.Done()
+	}()
+
+	return updates, nil
+}
+
 func (a *API) ClientHasLocal(ctx context.Context, root cid.Cid) (bool, error) {
 	// TODO: check if we have the ENTIRE dag
 
@@ -816,3 +831,19 @@ func (a *API) ClientDataTransferUpdates(ctx context.Context) (<-chan api.DataTra
 
 	return channels, nil
 }
+
+func newDealInfo(v storagemarket.ClientDeal) api.DealInfo {
+	return api.DealInfo{
+		ProposalCid:   v.ProposalCid,
+		DataRef:       v.DataRef,
+		State:         v.State,
+		Message:       v.Message,
+		Provider:      v.Proposal.Provider,
+		PieceCID:      v.Proposal.PieceCID,
+		Size:          uint64(v.Proposal.PieceSize.Unpadded()),
+		PricePerEpoch: v.Proposal.StoragePricePerEpoch,
+		Duration:      uint64(v.Proposal.Duration()),
+		DealID:        v.DealID,
+		CreationTime:  v.CreationTime.Time(),
+	}
+}