lotus/chain/deals/client.go

313 lines
7.7 KiB
Go
Raw Normal View History

2019-08-01 17:12:41 +00:00
package deals
import (
"context"
2019-10-31 21:01:44 +00:00
"github.com/filecoin-project/lotus/lib/statestore"
"github.com/filecoin-project/lotus/node/impl/full"
2019-08-01 17:12:41 +00:00
"github.com/ipfs/go-cid"
2019-08-06 22:04:21 +00:00
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/namespace"
2019-08-01 17:12:41 +00:00
logging "github.com/ipfs/go-log"
2019-08-02 14:09:54 +00:00
"github.com/libp2p/go-libp2p-core/host"
2019-08-07 19:48:53 +00:00
inet "github.com/libp2p/go-libp2p-core/network"
2019-08-01 17:12:41 +00:00
"github.com/libp2p/go-libp2p-core/peer"
2019-09-13 21:00:36 +00:00
"golang.org/x/xerrors"
2019-08-01 17:12:41 +00:00
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/chain/actors"
"github.com/filecoin-project/lotus/chain/address"
"github.com/filecoin-project/lotus/chain/stmgr"
2019-10-22 11:29:41 +00:00
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/chain/wallet"
"github.com/filecoin-project/lotus/lib/cborrpc"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/retrieval/discovery"
2019-08-01 17:12:41 +00:00
)
var log = logging.Logger("deals")
2019-08-06 22:04:21 +00:00
type ClientDeal struct {
ProposalCid cid.Cid
Proposal actors.StorageDealProposal
2019-09-10 14:13:24 +00:00
State api.DealState
2019-08-06 22:04:21 +00:00
Miner peer.ID
2019-09-10 12:35:43 +00:00
s inet.Stream
2019-08-01 17:12:41 +00:00
}
type Client struct {
2019-09-13 21:00:36 +00:00
sm *stmgr.StateManager
2019-10-22 10:09:36 +00:00
chain *store.ChainStore
2019-08-26 13:45:36 +00:00
h host.Host
w *wallet.Wallet
dag dtypes.ClientDAG
discovery *discovery.Local
mpool full.MpoolAPI
2019-08-01 17:12:41 +00:00
2019-11-01 11:07:05 +00:00
deals *statestore.StateStore
2019-09-10 12:35:43 +00:00
conns map[cid.Cid]inet.Stream
2019-08-01 17:12:41 +00:00
incoming chan *ClientDeal
2019-09-10 12:35:43 +00:00
updated chan clientDealUpdate
2019-08-01 17:12:41 +00:00
stop chan struct{}
stopped chan struct{}
}
2019-09-10 12:35:43 +00:00
type clientDealUpdate struct {
2019-09-10 14:13:24 +00:00
newState api.DealState
2019-09-10 12:35:43 +00:00
id cid.Cid
err error
}
func NewClient(sm *stmgr.StateManager, chain *store.ChainStore, h host.Host, w *wallet.Wallet, ds dtypes.MetadataDS, dag dtypes.ClientDAG, discovery *discovery.Local, mpool full.MpoolAPI) *Client {
2019-08-01 17:12:41 +00:00
c := &Client{
2019-09-13 21:00:36 +00:00
sm: sm,
2019-10-22 10:09:36 +00:00
chain: chain,
2019-08-26 13:45:36 +00:00
h: h,
w: w,
dag: dag,
discovery: discovery,
mpool: mpool,
2019-08-01 17:12:41 +00:00
2019-11-01 11:07:05 +00:00
deals: statestore.New(namespace.Wrap(ds, datastore.NewKey("/deals/client"))),
2019-09-10 12:35:43 +00:00
conns: map[cid.Cid]inet.Stream{},
2019-08-01 17:12:41 +00:00
incoming: make(chan *ClientDeal, 16),
2019-09-10 12:35:43 +00:00
updated: make(chan clientDealUpdate, 16),
2019-08-01 17:12:41 +00:00
stop: make(chan struct{}),
stopped: make(chan struct{}),
}
return c
}
2019-09-10 12:35:43 +00:00
func (c *Client) Run(ctx context.Context) {
2019-08-01 17:12:41 +00:00
go func() {
defer close(c.stopped)
for {
select {
case deal := <-c.incoming:
2019-09-10 12:35:43 +00:00
c.onIncoming(deal)
case update := <-c.updated:
c.onUpdated(ctx, update)
2019-08-01 17:12:41 +00:00
case <-c.stop:
return
}
}
}()
}
func (c *Client) onIncoming(deal *ClientDeal) {
2019-09-10 12:35:43 +00:00
log.Info("incoming deal")
2019-08-07 19:48:53 +00:00
2019-09-10 12:35:43 +00:00
if _, ok := c.conns[deal.ProposalCid]; ok {
log.Errorf("tracking deal connection: already tracking connection for deal %s", deal.ProposalCid)
return
2019-08-02 14:09:54 +00:00
}
2019-09-10 12:35:43 +00:00
c.conns[deal.ProposalCid] = deal.s
2019-08-07 19:48:53 +00:00
2019-09-10 12:35:43 +00:00
if err := c.deals.Begin(deal.ProposalCid, deal); err != nil {
// We may have re-sent the proposal
log.Errorf("deal tracking failed: %s", err)
c.failDeal(deal.ProposalCid, err)
return
2019-08-02 14:09:54 +00:00
}
2019-08-07 19:48:53 +00:00
2019-09-10 12:35:43 +00:00
go func() {
c.updated <- clientDealUpdate{
2019-09-10 14:13:24 +00:00
newState: api.DealUnknown,
2019-09-10 12:35:43 +00:00
id: deal.ProposalCid,
err: nil,
}
}()
2019-08-07 19:48:53 +00:00
}
2019-09-10 12:35:43 +00:00
func (c *Client) onUpdated(ctx context.Context, update clientDealUpdate) {
log.Infof("Deal %s updated state to %d", update.id, update.newState)
var deal ClientDeal
2019-11-01 11:07:05 +00:00
err := c.deals.Mutate(update.id, func(d *ClientDeal) error {
2019-09-10 12:35:43 +00:00
d.State = update.newState
deal = *d
return nil
})
2019-09-13 19:43:33 +00:00
if update.err != nil {
log.Errorf("deal %s failed: %s", update.id, update.err)
c.failDeal(update.id, update.err)
return
}
2019-08-02 14:09:54 +00:00
if err != nil {
2019-09-10 12:35:43 +00:00
c.failDeal(update.id, err)
return
2019-08-07 19:48:53 +00:00
}
2019-09-10 12:35:43 +00:00
switch update.newState {
2019-09-10 14:13:24 +00:00
case api.DealUnknown: // new
c.handle(ctx, deal, c.new, api.DealAccepted)
case api.DealAccepted:
c.handle(ctx, deal, c.accepted, api.DealStaged)
case api.DealStaged:
c.handle(ctx, deal, c.staged, api.DealSealing)
case api.DealSealing:
c.handle(ctx, deal, c.sealing, api.DealComplete)
2019-08-02 14:09:54 +00:00
}
2019-08-07 19:48:53 +00:00
}
2019-08-15 00:28:52 +00:00
type ClientDealProposal struct {
2019-10-23 18:04:07 +00:00
Data cid.Cid
2019-08-15 00:28:52 +00:00
2019-10-29 10:01:18 +00:00
PricePerEpoch types.BigInt
ProposalExpiration uint64
Duration uint64
2019-08-15 00:28:52 +00:00
ProviderAddress address.Address
Client address.Address
MinerID peer.ID
2019-08-15 00:28:52 +00:00
}
func (c *Client) Start(ctx context.Context, p ClientDealProposal) (cid.Cid, error) {
// check market funds
clientMarketBalance, err := c.sm.MarketBalance(ctx, p.Client, nil)
if err != nil {
return cid.Undef, xerrors.Errorf("getting client market balance failed: %w", err)
}
2019-10-29 10:01:18 +00:00
if clientMarketBalance.Available.LessThan(types.BigMul(p.PricePerEpoch, types.NewInt(p.Duration))) {
// TODO: move to a smarter market funds manager
smsg, err := c.mpool.MpoolPushMessage(ctx, &types.Message{
To: actors.StorageMarketAddress,
From: p.Client,
2019-10-29 10:01:18 +00:00
Value: types.BigMul(p.PricePerEpoch, types.NewInt(p.Duration)),
GasPrice: types.NewInt(0),
GasLimit: types.NewInt(1000000),
Method: actors.SMAMethods.AddBalance,
})
if err != nil {
return cid.Undef, err
}
_, r, err := c.sm.WaitForMessage(ctx, smsg.Cid())
if err != nil {
return cid.Undef, err
}
if r.ExitCode != 0 {
return cid.Undef, xerrors.Errorf("adding funds to storage miner market actor failed: exit %d", r.ExitCode)
}
}
2019-11-06 17:38:42 +00:00
commP, pieceSize, err := c.commP(ctx, p.Data)
2019-10-23 18:04:07 +00:00
2019-11-06 17:38:42 +00:00
dealProposal := &actors.StorageDealProposal{
PieceRef: commP,
PieceSize: uint64(pieceSize),
2019-10-29 10:01:18 +00:00
PieceSerialization: actors.SerializationUnixFSv0,
Client: p.Client,
Provider: p.ProviderAddress,
ProposalExpiration: p.ProposalExpiration,
Duration: p.Duration,
StoragePricePerEpoch: p.PricePerEpoch,
2019-11-06 17:38:42 +00:00
StorageCollateral: types.NewInt(uint64(pieceSize)), // TODO: real calc
2019-08-01 17:12:41 +00:00
}
2019-11-06 17:38:42 +00:00
if err := api.SignWith(ctx, c.w.Sign, p.Client, dealProposal); err != nil {
2019-10-23 10:42:52 +00:00
return cid.Undef, xerrors.Errorf("signing deal proposal failed: %w", err)
2019-08-15 00:28:52 +00:00
}
2019-11-06 17:38:42 +00:00
proposalNd, err := cborrpc.AsIpld(dealProposal)
2019-08-01 17:12:41 +00:00
if err != nil {
2019-10-23 10:42:52 +00:00
return cid.Undef, xerrors.Errorf("getting proposal node failed: %w", err)
2019-08-01 17:12:41 +00:00
}
2019-08-02 16:25:10 +00:00
s, err := c.h.NewStream(ctx, p.MinerID, DealProtocolID)
if err != nil {
2019-10-23 10:42:52 +00:00
return cid.Undef, xerrors.Errorf("connecting to storage provider failed: %w", err)
2019-08-06 22:04:21 +00:00
}
2019-11-06 17:38:42 +00:00
proposal := &Proposal{
DealProposal: dealProposal,
Piece: p.Data,
}
if err := cborrpc.WriteCborRPC(s, proposal); err != nil {
s.Reset()
2019-10-23 10:42:52 +00:00
return cid.Undef, xerrors.Errorf("sending proposal to storage provider failed: %w", err)
2019-08-06 22:04:21 +00:00
}
deal := &ClientDeal{
2019-09-10 12:35:43 +00:00
ProposalCid: proposalNd.Cid(),
2019-11-06 17:38:42 +00:00
Proposal: *dealProposal,
2019-09-10 14:13:24 +00:00
State: api.DealUnknown,
2019-09-10 12:35:43 +00:00
Miner: p.MinerID,
s: s,
}
2019-08-01 17:12:41 +00:00
2019-09-10 12:35:43 +00:00
c.incoming <- deal
2019-08-26 13:45:36 +00:00
return deal.ProposalCid, c.discovery.AddPeer(p.Data, discovery.RetrievalPeer{
2019-11-06 17:38:42 +00:00
Address: dealProposal.Provider,
2019-08-26 13:45:36 +00:00
ID: deal.Miner,
})
2019-08-01 17:12:41 +00:00
}
2019-09-13 21:00:36 +00:00
func (c *Client) QueryAsk(ctx context.Context, p peer.ID, a address.Address) (*types.SignedStorageAsk, error) {
s, err := c.h.NewStream(ctx, p, AskProtocolID)
if err != nil {
return nil, err
}
req := &AskRequest{
Miner: a,
}
if err := cborrpc.WriteCborRPC(s, req); err != nil {
return nil, xerrors.Errorf("failed to send ask request: %w", err)
}
var out AskResponse
if err := cborrpc.ReadCborRPC(s, &out); err != nil {
return nil, xerrors.Errorf("failed to read ask response: %w", err)
}
if out.Ask == nil {
return nil, xerrors.Errorf("got no ask back")
}
if out.Ask.Ask.Miner != a {
return nil, xerrors.Errorf("got back ask for wrong miner")
}
if err := c.checkAskSignature(out.Ask); err != nil {
return nil, xerrors.Errorf("ask was not properly signed")
}
return out.Ask, nil
}
2019-09-10 14:13:24 +00:00
func (c *Client) List() ([]ClientDeal, error) {
2019-11-01 11:07:05 +00:00
var out []ClientDeal
if err := c.deals.List(&out); err != nil {
return nil, err
}
return out, nil
2019-09-10 14:13:24 +00:00
}
2019-11-06 19:44:28 +00:00
func (c *Client) GetDeal(d cid.Cid) (*ClientDeal, error) {
var out ClientDeal
if err := c.deals.Get(d, &out); err != nil {
return nil, err
}
return &out, nil
}
2019-08-01 17:12:41 +00:00
func (c *Client) Stop() {
close(c.stop)
<-c.stopped
}