lotus/chain/deals/handler_states.go

314 lines
9.6 KiB
Go
Raw Normal View History

package deals
import (
"bytes"
"context"
2019-09-24 21:13:47 +00:00
"github.com/filecoin-project/go-sectorbuilder/sealing_state"
cbor "github.com/ipfs/go-ipld-cbor"
"github.com/ipfs/go-merkledag"
unixfile "github.com/ipfs/go-unixfs/file"
"golang.org/x/xerrors"
2019-08-26 08:02:26 +00:00
2019-09-24 21:13:47 +00:00
"github.com/filecoin-project/go-lotus/api"
"github.com/filecoin-project/go-lotus/build"
2019-09-06 22:39:47 +00:00
"github.com/filecoin-project/go-lotus/chain/actors"
"github.com/filecoin-project/go-lotus/chain/types"
2019-08-26 08:02:26 +00:00
"github.com/filecoin-project/go-lotus/lib/sectorbuilder"
2019-08-26 10:04:57 +00:00
"github.com/filecoin-project/go-lotus/storage/sectorblocks"
)
2019-09-10 12:35:43 +00:00
type minerHandlerFunc func(ctx context.Context, deal MinerDeal) (func(*MinerDeal), error)
2019-09-10 14:13:24 +00:00
func (h *Handler) handle(ctx context.Context, deal MinerDeal, cb minerHandlerFunc, next api.DealState) {
go func() {
2019-08-12 21:48:18 +00:00
mut, err := cb(ctx, deal)
2019-09-16 18:04:15 +00:00
if err == nil && next == api.DealNoUpdate {
return
}
select {
2019-09-10 12:35:43 +00:00
case h.updated <- minerDealUpdate{
newState: next,
id: deal.ProposalCid,
err: err,
2019-08-12 21:48:18 +00:00
mut: mut,
}:
case <-h.stop:
}
}()
}
// ACCEPTED
2019-09-25 12:56:04 +00:00
func (h *Handler) checkVoucher(ctx context.Context, deal MinerDeal, voucher *types.SignedVoucher, lane uint64, maxClose uint64, amount types.BigInt) error {
err := h.full.PaychVoucherCheckValid(ctx, deal.Proposal.Payment.PayChActor, voucher)
if err != nil {
return err
}
if voucher.Extra == nil {
return xerrors.New("voucher.Extra not set")
}
if voucher.Extra.Actor != deal.Proposal.MinerAddress {
return xerrors.Errorf("extra params actor didn't match miner address in proposal: '%s' != '%s'", voucher.Extra.Actor, deal.Proposal.MinerAddress)
}
if voucher.Extra.Method != actors.MAMethods.PaymentVerifyInclusion {
return xerrors.Errorf("expected extra method %d, got %d", actors.MAMethods.PaymentVerifyInclusion, voucher.Extra.Method)
}
var inclChallenge actors.PieceInclVoucherData
if err := cbor.DecodeInto(voucher.Extra.Data, &inclChallenge); err != nil {
return xerrors.Errorf("failed to decode storage voucher data for verification: %w", err)
}
if inclChallenge.PieceSize.Uint64() != deal.Proposal.Size {
return xerrors.Errorf("paych challenge piece size didn't match deal proposal size: %d != %d", inclChallenge.PieceSize.Uint64(), deal.Proposal.Size)
}
if !bytes.Equal(inclChallenge.CommP, deal.Proposal.CommP) {
return xerrors.New("paych challenge commP didn't match deal proposal")
}
if voucher.MinCloseHeight > maxClose {
return xerrors.Errorf("MinCloseHeight too high (%d), max expected: %d", voucher.MinCloseHeight, maxClose)
}
if voucher.TimeLock > maxClose {
return xerrors.Errorf("TimeLock too high (%d), max expected: %d", voucher.TimeLock, maxClose)
}
if len(voucher.Merges) > 0 {
return xerrors.New("didn't expect any merges")
}
if voucher.Amount.LessThan(amount) {
return xerrors.Errorf("not enough funds in the voucher: %s < %s; vl=%d", voucher.Amount, amount, len(deal.Proposal.Payment.Vouchers))
}
if voucher.Lane != lane {
return xerrors.Errorf("expected all vouchers on lane %d, found voucher on lane %d", lane, voucher.Lane)
}
return nil
}
func (h *Handler) consumeVouchers(ctx context.Context, deal MinerDeal) error {
2019-09-06 22:39:47 +00:00
curHead, err := h.full.ChainHead(ctx)
if err != nil {
2019-09-09 19:21:37 +00:00
return err
2019-09-06 22:39:47 +00:00
}
2019-09-24 21:13:47 +00:00
if len(deal.Proposal.Payment.Vouchers) == 0 {
return xerrors.Errorf("no payment vouchers for deal")
}
increments := deal.Proposal.Duration / uint64(len(deal.Proposal.Payment.Vouchers))
startH := deal.Proposal.Payment.Vouchers[0].TimeLock - (deal.Proposal.Duration / increments)
if startH > curHead.Height()+build.DealVoucherSkewLimit {
return xerrors.Errorf("deal starts too far into the future")
}
vspec := VoucherSpec(deal.Proposal.Duration, deal.Proposal.TotalPrice, startH, nil)
lane := deal.Proposal.Payment.Vouchers[0].Lane
2019-09-06 22:39:47 +00:00
for i, voucher := range deal.Proposal.Payment.Vouchers {
2019-09-24 21:13:47 +00:00
maxClose := curHead.Height() + (increments * uint64(i+1)) + build.DealVoucherSkewLimit
2019-09-06 22:39:47 +00:00
2019-09-25 12:56:04 +00:00
if err := h.checkVoucher(ctx, deal, voucher, lane, maxClose, vspec[i].Amount); err != nil {
return xerrors.Errorf("validating payment voucher %d: %w", i, err)
2019-09-24 21:13:47 +00:00
}
}
2019-09-06 22:39:47 +00:00
2019-09-24 21:13:47 +00:00
minPrice := types.BigMul(types.BigMul(h.pricePerByteBlock, types.NewInt(deal.Proposal.Size)), types.NewInt(deal.Proposal.Duration))
if types.BigCmp(minPrice, deal.Proposal.TotalPrice) > 0 {
return xerrors.Errorf("minimum price: %s", minPrice)
}
2019-09-06 22:39:47 +00:00
2019-09-25 12:56:04 +00:00
prevAmt := types.NewInt(0)
2019-09-24 21:13:47 +00:00
2019-09-25 12:56:04 +00:00
for i, voucher := range deal.Proposal.Payment.Vouchers {
delta, err := h.full.PaychVoucherAdd(ctx, deal.Proposal.Payment.PayChActor, voucher, nil, types.BigSub(vspec[i].Amount, prevAmt))
if err != nil {
return xerrors.Errorf("consuming payment voucher %d: %w", i, err)
}
prevAmt = types.BigAdd(prevAmt, delta)
2019-09-06 22:39:47 +00:00
}
2019-09-09 19:21:37 +00:00
return nil
}
func (h *Handler) accept(ctx context.Context, deal MinerDeal) (func(*MinerDeal), error) {
switch deal.Proposal.SerializationMode {
//case SerializationRaw:
//case SerializationIPLD:
case SerializationUnixFs:
default:
return nil, xerrors.Errorf("deal proposal with unsupported serialization: %s", deal.Proposal.SerializationMode)
}
if deal.Proposal.Payment.ChannelMessage != nil {
log.Info("waiting for channel message to appear on chain")
2019-09-13 20:50:07 +00:00
if _, err := h.full.ChainWaitMsg(ctx, *deal.Proposal.Payment.ChannelMessage); err != nil {
return nil, xerrors.Errorf("waiting for paych message: %w", err)
}
}
2019-09-25 12:56:04 +00:00
if err := h.consumeVouchers(ctx, deal); err != nil {
2019-09-09 19:21:37 +00:00
return nil, err
}
log.Info("fetching data for a deal")
2019-09-09 19:21:37 +00:00
err := h.sendSignedResponse(StorageDealResponse{
2019-09-10 14:13:24 +00:00
State: api.DealAccepted,
Message: "",
Proposal: deal.ProposalCid,
})
if err != nil {
2019-08-12 21:48:18 +00:00
return nil, err
}
2019-08-12 21:48:18 +00:00
return nil, merkledag.FetchGraph(ctx, deal.Ref, h.dag)
}
// STAGED
2019-08-12 21:48:18 +00:00
func (h *Handler) staged(ctx context.Context, deal MinerDeal) (func(*MinerDeal), error) {
err := h.sendSignedResponse(StorageDealResponse{
2019-09-10 14:13:24 +00:00
State: api.DealStaged,
Proposal: deal.ProposalCid,
})
if err != nil {
2019-08-07 19:48:53 +00:00
log.Warnf("Sending deal response failed: %s", err)
}
root, err := h.dag.Get(ctx, deal.Ref)
if err != nil {
2019-08-12 21:48:18 +00:00
return nil, xerrors.Errorf("failed to get file root for deal: %s", err)
}
// TODO: abstract this away into ReadSizeCloser + implement different modes
n, err := unixfile.NewUnixfsFile(ctx, h.dag, root)
if err != nil {
2019-08-12 21:48:18 +00:00
return nil, xerrors.Errorf("cannot open unixfs file: %s", err)
}
2019-08-26 10:04:57 +00:00
uf, ok := n.(sectorblocks.UnixfsReader)
if !ok {
// we probably got directory, unsupported for now
2019-08-26 10:04:57 +00:00
return nil, xerrors.Errorf("unsupported unixfs file type")
}
2019-08-26 08:02:26 +00:00
sectorID, err := h.secst.AddUnixfsPiece(deal.Proposal.PieceRef, uf, deal.Proposal.Duration)
if err != nil {
2019-08-12 21:48:18 +00:00
return nil, xerrors.Errorf("AddPiece failed: %s", err)
}
log.Warnf("New Sector: %d", sectorID)
2019-08-12 21:48:18 +00:00
return func(deal *MinerDeal) {
deal.SectorID = sectorID
}, nil
}
// SEALING
2019-08-12 21:48:18 +00:00
func getInclusionProof(ref string, status sectorbuilder.SectorSealingStatus) (PieceInclusionProof, error) {
for i, p := range status.Pieces {
if p.Key == ref {
return PieceInclusionProof{
Position: uint64(i),
ProofElements: p.InclusionProof,
}, nil
}
}
return PieceInclusionProof{}, xerrors.Errorf("pieceInclusionProof for %s in sector %d not found", ref, status.SectorID)
}
2019-09-16 16:40:26 +00:00
func (h *Handler) waitSealed(ctx context.Context, deal MinerDeal) (sectorbuilder.SectorSealingStatus, error) {
status, err := h.secst.WaitSeal(ctx, deal.SectorID)
2019-08-14 20:27:10 +00:00
if err != nil {
return sectorbuilder.SectorSealingStatus{}, err
}
2019-08-12 21:48:18 +00:00
2019-09-23 10:50:28 +00:00
switch status.State {
case sealing_state.Sealed:
case sealing_state.Failed:
2019-08-14 20:27:10 +00:00
return sectorbuilder.SectorSealingStatus{}, xerrors.Errorf("sealing sector %d for deal %s (ref=%s) failed: %s", deal.SectorID, deal.ProposalCid, deal.Ref, status.SealErrorMsg)
2019-09-23 10:50:28 +00:00
case sealing_state.Pending:
2019-08-14 20:27:10 +00:00
return sectorbuilder.SectorSealingStatus{}, xerrors.Errorf("sector status was 'pending' after call to WaitSeal (for sector %d)", deal.SectorID)
2019-09-23 10:50:28 +00:00
case sealing_state.Sealing:
2019-08-14 20:27:10 +00:00
return sectorbuilder.SectorSealingStatus{}, xerrors.Errorf("sector status was 'wait' after call to WaitSeal (for sector %d)", deal.SectorID)
default:
return sectorbuilder.SectorSealingStatus{}, xerrors.Errorf("unknown SealStatusCode: %d", status.SectorID)
2019-08-12 21:48:18 +00:00
}
2019-08-14 20:27:10 +00:00
2019-08-12 21:48:18 +00:00
return status, nil
}
func (h *Handler) sealing(ctx context.Context, deal MinerDeal) (func(*MinerDeal), error) {
2019-09-16 16:40:26 +00:00
status, err := h.waitSealed(ctx, deal)
2019-08-12 21:48:18 +00:00
if err != nil {
return nil, err
}
2019-08-28 21:11:29 +00:00
// TODO: don't hardcode unixfs
2019-08-28 23:01:38 +00:00
ip, err := getInclusionProof(string(sectorblocks.SerializationUnixfs0)+deal.Ref.String(), status)
2019-08-12 21:48:18 +00:00
if err != nil {
return nil, err
}
proof := &actors.InclusionProof{
Sector: deal.SectorID,
Proof: ip.ProofElements,
}
proofB, err := cbor.DumpObject(proof)
if err != nil {
return nil, err
}
// store proofs for channels
for i, v := range deal.Proposal.Payment.Vouchers {
if v.Extra.Method == actors.MAMethods.PaymentVerifyInclusion {
2019-09-16 21:25:23 +00:00
// TODO: Set correct minAmount
if _, err := h.full.PaychVoucherAdd(ctx, deal.Proposal.Payment.PayChActor, v, proofB, types.NewInt(0)); err != nil {
return nil, xerrors.Errorf("storing payment voucher %d proof: %w", i, err)
}
}
}
2019-08-12 21:48:18 +00:00
err = h.sendSignedResponse(StorageDealResponse{
2019-09-10 14:13:24 +00:00
State: api.DealSealing,
2019-08-12 21:48:18 +00:00
Proposal: deal.ProposalCid,
PieceInclusionProof: ip,
2019-09-10 12:35:43 +00:00
CommD: status.CommD[:],
2019-08-12 21:48:18 +00:00
})
if err != nil {
log.Warnf("Sending deal response failed: %s", err)
}
return nil, nil
}
2019-09-16 16:40:26 +00:00
func (h *Handler) complete(ctx context.Context, deal MinerDeal) (func(*MinerDeal), error) {
mcid, err := h.commt.WaitCommit(ctx, deal.Proposal.MinerAddress, deal.SectorID)
if err != nil {
log.Warnf("Waiting for sector commitment message: %s", err)
}
err = h.sendSignedResponse(StorageDealResponse{
2019-09-16 21:26:19 +00:00
State: api.DealComplete,
Proposal: deal.ProposalCid,
2019-09-16 16:40:26 +00:00
SectorCommitMessage: &mcid,
})
if err != nil {
log.Warnf("Sending deal response failed: %s", err)
}
return nil, nil
}