lotus/itests/kit2/deals.go

246 lines
6.8 KiB
Go

package kit2
import (
"bytes"
"context"
"io/ioutil"
"testing"
"time"
"github.com/ipfs/go-cid"
files "github.com/ipfs/go-ipfs-files"
"github.com/ipld/go-car"
"github.com/stretchr/testify/require"
"github.com/filecoin-project/go-fil-markets/storagemarket"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/types"
sealing "github.com/filecoin-project/lotus/extern/storage-sealing"
ipld "github.com/ipfs/go-ipld-format"
dag "github.com/ipfs/go-merkledag"
dstest "github.com/ipfs/go-merkledag/test"
unixfile "github.com/ipfs/go-unixfs/file"
)
type DealHarness struct {
t *testing.T
client *TestFullNode
miner *TestMiner
}
// NewDealHarness creates a test harness that contains testing utilities for deals.
func NewDealHarness(t *testing.T, client *TestFullNode, miner *TestMiner) *DealHarness {
return &DealHarness{
t: t,
client: client,
miner: miner,
}
}
// MakeOnlineDeal makes an online deal, generating a random file with the
// supplied seed, and setting the specified fast retrieval flag and start epoch
// on the storage deal. It returns when the deal is sealed.
//
// TODO: convert input parameters to struct, and add size as an input param.
func (dh *DealHarness) MakeOnlineDeal(ctx context.Context, rseed int, fastRet bool, startEpoch abi.ChainEpoch) (deal *cid.Cid, res *api.ImportRes, path string) {
res, path = dh.client.CreateImportFile(ctx, rseed, 0)
dh.t.Logf("FILE CID: %s", res.Root)
deal = dh.StartDeal(ctx, res.Root, fastRet, startEpoch)
// TODO: this sleep is only necessary because deals don't immediately get logged in the dealstore, we should fix this
time.Sleep(time.Second)
dh.WaitDealSealed(ctx, deal, false, false, nil)
return deal, res, path
}
// StartDeal starts a storage deal between the client and the miner.
func (dh *DealHarness) StartDeal(ctx context.Context, fcid cid.Cid, fastRet bool, startEpoch abi.ChainEpoch) *cid.Cid {
maddr, err := dh.miner.ActorAddress(ctx)
require.NoError(dh.t, err)
addr, err := dh.client.WalletDefaultAddress(ctx)
require.NoError(dh.t, err)
deal, err := dh.client.ClientStartDeal(ctx, &api.StartDealParams{
Data: &storagemarket.DataRef{
TransferType: storagemarket.TTGraphsync,
Root: fcid,
},
Wallet: addr,
Miner: maddr,
EpochPrice: types.NewInt(1000000),
DealStartEpoch: startEpoch,
MinBlocksDuration: uint64(build.MinDealDuration),
FastRetrieval: fastRet,
})
require.NoError(dh.t, err)
return deal
}
// WaitDealSealed waits until the deal is sealed.
func (dh *DealHarness) WaitDealSealed(ctx context.Context, deal *cid.Cid, noseal, noSealStart bool, cb func()) {
loop:
for {
di, err := dh.client.ClientGetDealInfo(ctx, *deal)
require.NoError(dh.t, err)
switch di.State {
case storagemarket.StorageDealAwaitingPreCommit, storagemarket.StorageDealSealing:
if noseal {
return
}
if !noSealStart {
dh.StartSealingWaiting(ctx)
}
case storagemarket.StorageDealProposalRejected:
dh.t.Fatal("deal rejected")
case storagemarket.StorageDealFailing:
dh.t.Fatal("deal failed")
case storagemarket.StorageDealError:
dh.t.Fatal("deal errored", di.Message)
case storagemarket.StorageDealActive:
dh.t.Log("COMPLETE", di)
break loop
}
mds, err := dh.miner.MarketListIncompleteDeals(ctx)
require.NoError(dh.t, err)
var minerState storagemarket.StorageDealStatus
for _, md := range mds {
if md.DealID == di.DealID {
minerState = md.State
break
}
}
dh.t.Logf("Deal %d state: client:%s provider:%s\n", di.DealID, storagemarket.DealStates[di.State], storagemarket.DealStates[minerState])
time.Sleep(time.Second / 2)
if cb != nil {
cb()
}
}
}
// WaitDealSealed waits until the deal is published.
func (dh *DealHarness) WaitDealPublished(ctx context.Context, deal *cid.Cid) {
subCtx, cancel := context.WithCancel(ctx)
defer cancel()
updates, err := dh.miner.MarketGetDealUpdates(subCtx)
require.NoError(dh.t, err)
for {
select {
case <-ctx.Done():
dh.t.Fatal("context timeout")
case di := <-updates:
if deal.Equals(di.ProposalCid) {
switch di.State {
case storagemarket.StorageDealProposalRejected:
dh.t.Fatal("deal rejected")
case storagemarket.StorageDealFailing:
dh.t.Fatal("deal failed")
case storagemarket.StorageDealError:
dh.t.Fatal("deal errored", di.Message)
case storagemarket.StorageDealFinalizing, storagemarket.StorageDealAwaitingPreCommit, storagemarket.StorageDealSealing, storagemarket.StorageDealActive:
dh.t.Log("COMPLETE", di)
return
}
dh.t.Log("Deal state: ", storagemarket.DealStates[di.State])
}
}
}
}
func (dh *DealHarness) StartSealingWaiting(ctx context.Context) {
snums, err := dh.miner.SectorsList(ctx)
require.NoError(dh.t, err)
for _, snum := range snums {
si, err := dh.miner.SectorsStatus(ctx, snum, false)
require.NoError(dh.t, err)
dh.t.Logf("Sector state: %s", si.State)
if si.State == api.SectorState(sealing.WaitDeals) {
require.NoError(dh.t, dh.miner.SectorStartSealing(ctx, snum))
}
dh.miner.FlushSealingBatches(ctx)
}
}
func (dh *DealHarness) PerformRetrieval(ctx context.Context, deal *cid.Cid, root cid.Cid, carExport bool) (path string) {
// perform retrieval.
info, err := dh.client.ClientGetDealInfo(ctx, *deal)
require.NoError(dh.t, err)
offers, err := dh.client.ClientFindData(ctx, root, &info.PieceCID)
require.NoError(dh.t, err)
require.NotEmpty(dh.t, offers, "no offers")
tmpfile, err := ioutil.TempFile(dh.t.TempDir(), "ret-car")
require.NoError(dh.t, err)
defer tmpfile.Close()
caddr, err := dh.client.WalletDefaultAddress(ctx)
require.NoError(dh.t, err)
ref := &api.FileRef{
Path: tmpfile.Name(),
IsCAR: carExport,
}
updates, err := dh.client.ClientRetrieveWithEvents(ctx, offers[0].Order(caddr), ref)
require.NoError(dh.t, err)
for update := range updates {
require.Emptyf(dh.t, update.Err, "retrieval failed: %s", update.Err)
}
rdata, err := ioutil.ReadFile(tmpfile.Name())
require.NoError(dh.t, err)
if carExport {
rdata = dh.ExtractFileFromCAR(ctx, rdata)
}
return tmpfile.Name()
}
func (dh *DealHarness) ExtractFileFromCAR(ctx context.Context, rdata []byte) []byte {
bserv := dstest.Bserv()
ch, err := car.LoadCar(bserv.Blockstore(), bytes.NewReader(rdata))
require.NoError(dh.t, err)
b, err := bserv.GetBlock(ctx, ch.Roots[0])
require.NoError(dh.t, err)
nd, err := ipld.Decode(b)
require.NoError(dh.t, err)
dserv := dag.NewDAGService(bserv)
fil, err := unixfile.NewUnixfsFile(ctx, dserv, nd)
require.NoError(dh.t, err)
tmpfile, err := ioutil.TempFile(dh.t.TempDir(), "file-in-car")
require.NoError(dh.t, err)
defer tmpfile.Close()
err = files.WriteTo(fil, tmpfile.Name())
require.NoError(dh.t, err)
rdata, err = ioutil.ReadFile(tmpfile.Name())
require.NoError(dh.t, err)
return rdata
}