Merge pull request #2714 from filecoin-project/feat/user-dealfiter

Support external deal decision logic

cmd/lotus-storage-miner/market.go

@@ -2,7 +2,6 @@ package main
 
 import (
 	"bufio"
-	"encoding/json"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -353,13 +352,24 @@ var dealsListCmd = &cli.Command{
 			return err
 		}
 
-		data, err := json.MarshalIndent(deals, "", "  ")
+		w := tabwriter.NewWriter(os.Stdout, 2, 4, 2, ' ', 0)
+
+		_, _ = fmt.Fprintf(w, "ProposalCid\tDealId\tState\tClient\tSize\tPrice\tDuration\n")
+
+		for _, deal := range deals {
+			pc, err := deal.Proposal.Cid()
 			if err != nil {
 				return err
 			}
+			propcid := pc.String()
+			propcid = "..." + propcid[len(propcid)-8:]
 
-		fmt.Println(string(data))
-		return nil
+			fil := types.FIL(types.BigMul(deal.Proposal.StoragePricePerEpoch, types.NewInt(uint64(deal.Proposal.Duration()))))
+
+			_, _ = fmt.Fprintf(w, "%s\t%d\t%s\t%s\t%s\t%s\t%s\n", propcid, deal.DealID, storagemarket.DealStates[deal.State], deal.Proposal.Client, units.BytesSize(float64(deal.Proposal.PieceSize)), fil, deal.Proposal.Duration())
+		}
+
+		return w.Flush()
 	},
 }
 

markets/dealfilter/cli.go

@@ -0,0 +1,40 @@
+package dealfilter
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"os/exec"
+
+	"github.com/filecoin-project/go-fil-markets/storagemarket"
+
+	"github.com/filecoin-project/lotus/node/modules/dtypes"
+)
+
+func CliDealFilter(cmd string) dtypes.DealFilter {
+	// TODO: run some checks on the cmd string
+
+	return func(ctx context.Context, deal storagemarket.MinerDeal) (bool, string, error) {
+		j, err := json.MarshalIndent(deal, "", "  ")
+		if err != nil {
+			return false, "", err
+		}
+
+		var out bytes.Buffer
+
+		c := exec.Command("sh", "-c", cmd)
+		c.Stdin = bytes.NewReader(j)
+		c.Stdout = &out
+		c.Stderr = &out
+
+		switch err := c.Run().(type) {
+		case nil:
+			return true, "", nil
+		case *exec.ExitError:
+			return false, out.String(), nil
+		default:
+			return false, "filter cmd run error", err
+		}
+
+	}
+}

node/builder.go

@@ -3,6 +3,7 @@ package node
 import (
 	"context"
 	"errors"
+	"github.com/filecoin-project/lotus/markets/dealfilter"
 	"time"
 
 	logging "github.com/ipfs/go-log"
@@ -309,6 +310,7 @@ func Online() Option {
 			Override(new(dtypes.ProviderRequestValidator), modules.NewProviderRequestValidator),
 			Override(new(dtypes.ProviderPieceStore), modules.NewProviderPieceStore),
 			Override(new(*storedask.StoredAsk), modules.NewStorageAsk),
+			Override(new(dtypes.DealFilter), modules.BasicDealFilter(nil)),
 			Override(new(modules.ProviderDealFunds), modules.NewProviderDealFunds),
 			Override(new(storagemarket.StorageProvider), modules.StorageProvider),
 			Override(new(storagemarket.StorageProviderNode), storageadapter.NewProviderNodeAdapter),
@@ -425,6 +427,10 @@ func ConfigStorageMiner(c interface{}) Option {
 	return Options(
 		ConfigCommon(&cfg.Common),
 
+		If(cfg.Dealmaking.Filter != "",
+			Override(new(dtypes.DealFilter), modules.BasicDealFilter(dealfilter.CliDealFilter(cfg.Dealmaking.Filter))),
+		),
+
 		Override(new(sectorstorage.SealerConfig), cfg.Storage),
 	)
 }

node/config/def.go

@@ -41,6 +41,8 @@ type DealmakingConfig struct {
 	ConsiderOfflineRetrievalDeals bool
 	PieceCidBlocklist             []cid.Cid
 	ExpectedSealDuration          Duration
+
+	Filter string
 }
 
 // API contains configs for API endpoint

node/modules/dtypes/miner.go

@@ -1,10 +1,14 @@
 package dtypes
 
 import (
-	"github.com/filecoin-project/go-address"
-	"github.com/filecoin-project/specs-actors/actors/abi"
-	"github.com/ipfs/go-cid"
+	"context"
 	"time"
+
+	"github.com/ipfs/go-cid"
+
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-fil-markets/storagemarket"
+	"github.com/filecoin-project/specs-actors/actors/abi"
 )
 
 type MinerAddress address.Address
@@ -64,3 +68,5 @@ type SetExpectedSealDurationFunc func(time.Duration) error
 // GetExpectedSealDurationFunc is a function which reads from miner
 // too determine how long sealing is expected to take
 type GetExpectedSealDurationFunc func() (time.Duration, error)
+
+type DealFilter func(ctx context.Context, deal storagemarket.MinerDeal) (bool, string, error)

node/modules/storageminer.go

@@ -364,27 +364,18 @@ func NewProviderDealFunds(ds dtypes.MetadataDS) (ProviderDealFunds, error) {
 	return funds.NewDealFunds(ds, datastore.NewKey("/marketfunds/provider"))
 }
 
-func StorageProvider(minerAddress dtypes.MinerAddress,
-	ffiConfig *ffiwrapper.Config,
-	storedAsk *storedask.StoredAsk,
-	h host.Host, ds dtypes.MetadataDS,
-	mds dtypes.StagingMultiDstore,
-	r repo.LockedRepo,
-	pieceStore dtypes.ProviderPieceStore,
-	dataTransfer dtypes.ProviderDataTransfer,
-	spn storagemarket.StorageProviderNode,
-	funds ProviderDealFunds,
-	onlineOk dtypes.ConsiderOnlineStorageDealsConfigFunc,
+func BasicDealFilter(user dtypes.DealFilter) func(onlineOk dtypes.ConsiderOnlineStorageDealsConfigFunc,
 	offlineOk dtypes.ConsiderOfflineStorageDealsConfigFunc,
 	blocklistFunc dtypes.StorageDealPieceCidBlocklistConfigFunc,
-	expectedSealTimeFunc dtypes.GetExpectedSealDurationFunc) (storagemarket.StorageProvider, error) {
-	net := smnet.NewFromLibp2pHost(h)
-	store, err := piecefilestore.NewLocalFileStore(piecefilestore.OsPath(r.Path()))
-	if err != nil {
-		return nil, err
-	}
+	expectedSealTimeFunc dtypes.GetExpectedSealDurationFunc,
+	spn storagemarket.StorageProviderNode) dtypes.DealFilter {
+	return func(onlineOk dtypes.ConsiderOnlineStorageDealsConfigFunc,
+		offlineOk dtypes.ConsiderOfflineStorageDealsConfigFunc,
+		blocklistFunc dtypes.StorageDealPieceCidBlocklistConfigFunc,
+		expectedSealTimeFunc dtypes.GetExpectedSealDurationFunc,
+		spn storagemarket.StorageProviderNode) dtypes.DealFilter {
 
-	opt := storageimpl.CustomDealDecisionLogic(func(ctx context.Context, deal storagemarket.MinerDeal) (bool, string, error) {
+		return func(ctx context.Context, deal storagemarket.MinerDeal) (bool, string, error) {
 			b, err := onlineOk()
 			if err != nil {
 				return false, "miner error", err
@@ -433,8 +424,34 @@ func StorageProvider(minerAddress dtypes.MinerAddress,
 				return false, fmt.Sprintf("cannot seal a sector before %s", deal.Proposal.StartEpoch), nil
 			}
 
+			if user != nil {
+				return user(ctx, deal)
+			}
+
 			return true, "", nil
-	})
+		}
+	}
+}
+
+func StorageProvider(minerAddress dtypes.MinerAddress,
+	ffiConfig *ffiwrapper.Config,
+	storedAsk *storedask.StoredAsk,
+	h host.Host, ds dtypes.MetadataDS,
+	mds dtypes.StagingMultiDstore,
+	r repo.LockedRepo,
+	pieceStore dtypes.ProviderPieceStore,
+	dataTransfer dtypes.ProviderDataTransfer,
+	spn storagemarket.StorageProviderNode,
+	df dtypes.DealFilter,
+	funds ProviderDealFunds,
+) (storagemarket.StorageProvider, error) {
+	net := smnet.NewFromLibp2pHost(h)
+	store, err := piecefilestore.NewLocalFileStore(piecefilestore.OsPath(r.Path()))
+	if err != nil {
+		return nil, err
+	}
+
+	opt := storageimpl.CustomDealDecisionLogic(storageimpl.DealDeciderFunc(df))
 
 	return storageimpl.NewProvider(net, namespace.Wrap(ds, datastore.NewKey("/deals/provider")), store, mds, pieceStore, dataTransfer, spn, address.Address(minerAddress), ffiConfig.SealProofType, storedAsk, funds, opt)
 }