lotus/itests/kit/deals.go

472 lines
14 KiB
Go
Raw Normal View History

2021-06-18 18:45:29 +00:00
package kit
import (
"context"
"errors"
2021-06-23 17:13:29 +00:00
"fmt"
2021-06-13 23:10:37 +00:00
"os"
"strings"
"testing"
"time"
2021-06-23 17:13:29 +00:00
"github.com/ipfs/go-cid"
files "github.com/ipfs/go-ipfs-files"
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"
2021-06-23 17:13:29 +00:00
"github.com/ipld/go-car"
"github.com/stretchr/testify/require"
"golang.org/x/sync/errgroup"
2022-06-14 15:00:51 +00:00
2022-11-07 15:14:56 +00:00
"github.com/filecoin-project/go-address"
2022-06-14 15:00:51 +00:00
"github.com/filecoin-project/go-fil-markets/retrievalmarket"
"github.com/filecoin-project/go-fil-markets/shared_testutil"
"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/storage/pipeline"
)
type DealHarness struct {
t *testing.T
client *TestFullNode
main *TestMiner
market *TestMiner
}
2021-06-14 04:10:34 +00:00
type MakeFullDealParams struct {
integrate DAG store and CARv2 in deal-making (#6671) This commit removes badger from the deal-making processes, and moves to a new architecture with the dagstore as the cental component on the miner-side, and CARv2s on the client-side. Every deal that has been handed off to the sealing subsystem becomes a shard in the dagstore. Shards are mounted via the LotusMount, which teaches the dagstore how to load the related piece when serving retrievals. When the miner starts the Lotus for the first time with this patch, we will perform a one-time migration of all active deals into the dagstore. This is a lightweight process, and it consists simply of registering the shards in the dagstore. Shards are backed by the unsealed copy of the piece. This is currently a CARv1. However, the dagstore keeps CARv2 indices for all pieces, so when it's time to acquire a shard to serve a retrieval, the unsealed CARv1 is joined with its index (safeguarded by the dagstore), to form a read-only blockstore, thus taking the place of the monolithic badger. Data transfers have been adjusted to interface directly with CARv2 files. On inbound transfers (client retrievals, miner storage deals), we stream the received data into a CARv2 ReadWrite blockstore. On outbound transfers (client storage deals, miner retrievals), we serve the data off a CARv2 ReadOnly blockstore. Client-side imports are managed by the refactored *imports.Manager component (when not using IPFS integration). Just like it before, we use the go-filestore library to avoid duplicating the data from the original file in the resulting UnixFS DAG (concretely the leaves). However, the target of those imports are what we call "ref-CARv2s": CARv2 files placed under the `$LOTUS_PATH/imports` directory, containing the intermediate nodes in full, and the leaves as positional references to the original file on disk. Client-side retrievals are placed into CARv2 files in the location: `$LOTUS_PATH/retrievals`. A new set of `Dagstore*` JSON-RPC operations and `lotus-miner dagstore` subcommands have been introduced on the miner-side to inspect and manage the dagstore. Despite moving to a CARv2-backed system, the IPFS integration has been respected, and it continues to be possible to make storage deals with data held in an IPFS node, and to perform retrievals directly into an IPFS node. NOTE: because the "staging" and "client" Badger blockstores are no longer used, existing imports on the client will be rendered useless. On startup, Lotus will enumerate all imports and print WARN statements on the log for each import that needs to be reimported. These log lines contain these messages: - import lacks carv2 path; import will not work; please reimport - import has missing/broken carv2; please reimport At the end, we will print a "sanity check completed" message indicating the count of imports found, and how many were deemed broken. Co-authored-by: Aarsh Shah <aarshkshah1992@gmail.com> Co-authored-by: Dirk McCormick <dirkmdev@gmail.com> Co-authored-by: Raúl Kripalani <raul@protocol.ai> Co-authored-by: Dirk McCormick <dirkmdev@gmail.com>
2021-08-16 22:34:32 +00:00
Rseed int
FastRet bool
StartEpoch abi.ChainEpoch
UseCARFileForStorageDeal bool
// SuspendUntilCryptoeconStable suspends deal-making, until cryptoecon
// parameters are stabilised. This affects projected collateral, and tests
// will fail in network version 13 and higher if deals are started too soon
// after network birth.
//
// The reason is that the formula for collateral calculation takes
// circulating supply into account:
//
// [portion of power this deal will be] * [~1% of tokens].
//
// In the first epochs after genesis, the total circulating supply is
// changing dramatically in percentual terms. Therefore, if the deal is
// proposed too soon, by the time it gets published on chain, the quoted
// provider collateral will no longer be valid.
//
// The observation is that deals fail with:
//
// GasEstimateMessageGas error: estimating gas used: message execution
// failed: exit 16, reason: Provider collateral out of bounds. (RetCode=16)
//
// Enabling this will suspend deal-making until the network has reached a
2021-06-23 18:21:42 +00:00
// height of 300.
SuspendUntilCryptoeconStable bool
2021-06-14 04:10:34 +00:00
}
// NewDealHarness creates a test harness that contains testing utilities for deals.
func NewDealHarness(t *testing.T, client *TestFullNode, main *TestMiner, market *TestMiner) *DealHarness {
return &DealHarness{
t: t,
client: client,
main: main,
market: market,
}
}
// 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, params MakeFullDealParams) (deal *cid.Cid, res *api.ImportRes, path string) {
integrate DAG store and CARv2 in deal-making (#6671) This commit removes badger from the deal-making processes, and moves to a new architecture with the dagstore as the cental component on the miner-side, and CARv2s on the client-side. Every deal that has been handed off to the sealing subsystem becomes a shard in the dagstore. Shards are mounted via the LotusMount, which teaches the dagstore how to load the related piece when serving retrievals. When the miner starts the Lotus for the first time with this patch, we will perform a one-time migration of all active deals into the dagstore. This is a lightweight process, and it consists simply of registering the shards in the dagstore. Shards are backed by the unsealed copy of the piece. This is currently a CARv1. However, the dagstore keeps CARv2 indices for all pieces, so when it's time to acquire a shard to serve a retrieval, the unsealed CARv1 is joined with its index (safeguarded by the dagstore), to form a read-only blockstore, thus taking the place of the monolithic badger. Data transfers have been adjusted to interface directly with CARv2 files. On inbound transfers (client retrievals, miner storage deals), we stream the received data into a CARv2 ReadWrite blockstore. On outbound transfers (client storage deals, miner retrievals), we serve the data off a CARv2 ReadOnly blockstore. Client-side imports are managed by the refactored *imports.Manager component (when not using IPFS integration). Just like it before, we use the go-filestore library to avoid duplicating the data from the original file in the resulting UnixFS DAG (concretely the leaves). However, the target of those imports are what we call "ref-CARv2s": CARv2 files placed under the `$LOTUS_PATH/imports` directory, containing the intermediate nodes in full, and the leaves as positional references to the original file on disk. Client-side retrievals are placed into CARv2 files in the location: `$LOTUS_PATH/retrievals`. A new set of `Dagstore*` JSON-RPC operations and `lotus-miner dagstore` subcommands have been introduced on the miner-side to inspect and manage the dagstore. Despite moving to a CARv2-backed system, the IPFS integration has been respected, and it continues to be possible to make storage deals with data held in an IPFS node, and to perform retrievals directly into an IPFS node. NOTE: because the "staging" and "client" Badger blockstores are no longer used, existing imports on the client will be rendered useless. On startup, Lotus will enumerate all imports and print WARN statements on the log for each import that needs to be reimported. These log lines contain these messages: - import lacks carv2 path; import will not work; please reimport - import has missing/broken carv2; please reimport At the end, we will print a "sanity check completed" message indicating the count of imports found, and how many were deemed broken. Co-authored-by: Aarsh Shah <aarshkshah1992@gmail.com> Co-authored-by: Dirk McCormick <dirkmdev@gmail.com> Co-authored-by: Raúl Kripalani <raul@protocol.ai> Co-authored-by: Dirk McCormick <dirkmdev@gmail.com>
2021-08-16 22:34:32 +00:00
if params.UseCARFileForStorageDeal {
res, _, path = dh.client.ClientImportCARFile(ctx, params.Rseed, 200)
} else {
res, path = dh.client.CreateImportFile(ctx, params.Rseed, 0)
}
dh.t.Logf("FILE CID: %s", res.Root)
if params.SuspendUntilCryptoeconStable {
dh.t.Logf("deal-making suspending until cryptecon parameters have stabilised")
2021-06-23 18:21:42 +00:00
ts := dh.client.WaitTillChain(ctx, HeightAtLeast(300))
dh.t.Logf("deal-making continuing; current height is %d", ts.Height())
}
dp := dh.DefaultStartDealParams()
dp.Data.Root = res.Root
dp.DealStartEpoch = params.StartEpoch
dp.FastRetrieval = params.FastRet
deal = dh.StartDeal(ctx, dp)
// TODO: this sleep is only necessary because deals don't immediately get logged in the dealstore, we should fix this
time.Sleep(time.Second)
fmt.Printf("WAIT DEAL SEALEDS START\n")
dh.WaitDealSealed(ctx, deal, false, false, nil)
fmt.Printf("WAIT DEAL SEALEDS END\n")
return deal, res, path
}
func (dh *DealHarness) DefaultStartDealParams() api.StartDealParams {
dp := api.StartDealParams{
Data: &storagemarket.DataRef{TransferType: storagemarket.TTGraphsync},
EpochPrice: types.NewInt(1000000),
MinBlocksDuration: uint64(build.MinDealDuration),
}
var err error
dp.Miner, err = dh.main.ActorAddress(context.Background())
require.NoError(dh.t, err)
dp.Wallet, err = dh.client.WalletDefaultAddress(context.Background())
require.NoError(dh.t, err)
return dp
}
// StartDeal starts a storage deal between the client and the miner.
func (dh *DealHarness) StartDeal(ctx context.Context, dealParams api.StartDealParams) *cid.Cid {
dealProposalCid, err := dh.client.ClientStartDeal(ctx, &dealParams)
require.NoError(dh.t, err)
return dealProposalCid
}
// 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.market.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()
}
}
fmt.Printf("WAIT DEAL SEALED LOOP BROKEN\n")
}
2021-10-08 10:52:56 +00:00
// WaitDealSealedQuiet waits until the deal is sealed, without logging anything.
func (dh *DealHarness) WaitDealSealedQuiet(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:
break loop
}
_, err = dh.market.MarketListIncompleteDeals(ctx)
require.NoError(dh.t, err)
time.Sleep(time.Second / 2)
if cb != nil {
cb()
}
}
}
func (dh *DealHarness) ExpectDealFailure(ctx context.Context, deal *cid.Cid, errs string) error {
for {
di, err := dh.client.ClientGetDealInfo(ctx, *deal)
require.NoError(dh.t, err)
switch di.State {
case storagemarket.StorageDealAwaitingPreCommit, storagemarket.StorageDealSealing:
return fmt.Errorf("deal is sealing, and we expected an error: %s", errs)
case storagemarket.StorageDealProposalRejected:
if strings.Contains(di.Message, errs) {
return nil
}
return fmt.Errorf("unexpected error: %s ; expected: %s", di.Message, errs)
case storagemarket.StorageDealFailing:
if strings.Contains(di.Message, errs) {
return nil
}
return fmt.Errorf("unexpected error: %s ; expected: %s", di.Message, errs)
case storagemarket.StorageDealError:
if strings.Contains(di.Message, errs) {
return nil
}
return fmt.Errorf("unexpected error: %s ; expected: %s", di.Message, errs)
case storagemarket.StorageDealActive:
return errors.New("expected to get an error, but didn't get one")
}
mds, err := dh.market.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)
}
}
2021-06-13 23:10:37 +00:00
// WaitDealPublished 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.market.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.main.SectorsListNonGenesis(ctx)
require.NoError(dh.t, err)
for _, snum := range snums {
si, err := dh.main.SectorsStatus(ctx, snum, false)
require.NoError(dh.t, err)
dh.t.Logf("Sector state <%d>-[%d]:, %s", snum, si.SealProof, si.State)
if si.State == api.SectorState(sealing.WaitDeals) {
require.NoError(dh.t, dh.main.SectorStartSealing(ctx, snum))
}
dh.main.FlushSealingBatches(ctx)
}
}
2022-01-12 12:55:45 +00:00
func (dh *DealHarness) PerformRetrieval(ctx context.Context, deal *cid.Cid, root cid.Cid, carExport bool, offers ...api.QueryOffer) (path string) {
2022-11-07 15:14:56 +00:00
return dh.PerformRetrievalWithOrder(ctx, deal, root, carExport, func(offer api.QueryOffer, a address.Address) api.RetrievalOrder {
return offer.Order(a)
}, offers...)
}
func (dh *DealHarness) PerformRetrievalWithOrder(ctx context.Context, deal *cid.Cid, root cid.Cid, carExport bool, makeOrder func(api.QueryOffer, address.Address) api.RetrievalOrder, offers ...api.QueryOffer) (path string) {
2022-01-12 12:55:45 +00:00
var offer api.QueryOffer
if len(offers) == 0 {
// perform retrieval.
info, err := dh.client.ClientGetDealInfo(ctx, *deal)
require.NoError(dh.t, err)
2022-01-12 12:55:45 +00:00
offers, err := dh.client.ClientFindData(ctx, root, &info.PieceCID)
require.NoError(dh.t, err)
require.NotEmpty(dh.t, offers, "no offers")
offer = offers[0]
} else {
offer = offers[0]
}
2022-06-29 17:07:29 +00:00
carFile := dh.t.TempDir() + string(os.PathSeparator) + "ret-car-" + root.String()
caddr, err := dh.client.WalletDefaultAddress(ctx)
require.NoError(dh.t, err)
2021-11-10 14:45:46 +00:00
updatesCtx, cancel := context.WithCancel(ctx)
updates, err := dh.client.ClientGetRetrievalUpdates(updatesCtx)
2021-11-16 11:04:03 +00:00
require.NoError(dh.t, err)
2022-11-07 15:14:56 +00:00
order := makeOrder(offer, caddr)
retrievalRes, err := dh.client.ClientRetrieve(ctx, order)
require.NoError(dh.t, err)
2021-11-10 14:45:46 +00:00
consumeEvents:
for {
var evt api.RetrievalInfo
select {
case <-updatesCtx.Done():
dh.t.Fatal("Retrieval Timed Out")
case evt = <-updates:
if evt.ID != retrievalRes.DealID {
continue
}
}
switch evt.Status {
case retrievalmarket.DealStatusCompleted:
break consumeEvents
case retrievalmarket.DealStatusRejected:
dh.t.Fatalf("Retrieval Proposal Rejected: %s", evt.Message)
case
retrievalmarket.DealStatusDealNotFound,
retrievalmarket.DealStatusErrored:
dh.t.Fatalf("Retrieval Error: %s", evt.Message)
}
}
2021-11-10 14:45:46 +00:00
cancel()
2022-11-07 15:14:56 +00:00
if order.RemoteStore != nil {
// if we're retrieving into a remote store, skip export
return ""
}
2021-11-10 14:45:46 +00:00
require.NoError(dh.t, dh.client.ClientExport(ctx,
api.ExportRef{
Root: root,
DealID: retrievalRes.DealID,
},
api.FileRef{
2022-06-29 17:07:29 +00:00
Path: carFile,
2021-11-10 14:45:46 +00:00
IsCAR: carExport,
}))
2022-06-29 17:07:29 +00:00
ret := carFile
2021-06-13 23:10:37 +00:00
return ret
}
2022-06-29 17:07:29 +00:00
func (dh *DealHarness) ExtractFileFromCAR(ctx context.Context, file *os.File) string {
bserv := dstest.Bserv()
2021-12-14 14:06:59 +00:00
ch, err := car.LoadCar(ctx, bserv.Blockstore(), file)
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)
2022-06-29 17:07:29 +00:00
tmpfile := dh.t.TempDir() + string(os.PathSeparator) + "file-in-car" + b.Cid().String()
2022-06-29 17:07:29 +00:00
err = files.WriteTo(fil, tmpfile)
require.NoError(dh.t, err)
2021-06-13 23:10:37 +00:00
return tmpfile
}
2021-06-23 17:13:29 +00:00
type RunConcurrentDealsOpts struct {
integrate DAG store and CARv2 in deal-making (#6671) This commit removes badger from the deal-making processes, and moves to a new architecture with the dagstore as the cental component on the miner-side, and CARv2s on the client-side. Every deal that has been handed off to the sealing subsystem becomes a shard in the dagstore. Shards are mounted via the LotusMount, which teaches the dagstore how to load the related piece when serving retrievals. When the miner starts the Lotus for the first time with this patch, we will perform a one-time migration of all active deals into the dagstore. This is a lightweight process, and it consists simply of registering the shards in the dagstore. Shards are backed by the unsealed copy of the piece. This is currently a CARv1. However, the dagstore keeps CARv2 indices for all pieces, so when it's time to acquire a shard to serve a retrieval, the unsealed CARv1 is joined with its index (safeguarded by the dagstore), to form a read-only blockstore, thus taking the place of the monolithic badger. Data transfers have been adjusted to interface directly with CARv2 files. On inbound transfers (client retrievals, miner storage deals), we stream the received data into a CARv2 ReadWrite blockstore. On outbound transfers (client storage deals, miner retrievals), we serve the data off a CARv2 ReadOnly blockstore. Client-side imports are managed by the refactored *imports.Manager component (when not using IPFS integration). Just like it before, we use the go-filestore library to avoid duplicating the data from the original file in the resulting UnixFS DAG (concretely the leaves). However, the target of those imports are what we call "ref-CARv2s": CARv2 files placed under the `$LOTUS_PATH/imports` directory, containing the intermediate nodes in full, and the leaves as positional references to the original file on disk. Client-side retrievals are placed into CARv2 files in the location: `$LOTUS_PATH/retrievals`. A new set of `Dagstore*` JSON-RPC operations and `lotus-miner dagstore` subcommands have been introduced on the miner-side to inspect and manage the dagstore. Despite moving to a CARv2-backed system, the IPFS integration has been respected, and it continues to be possible to make storage deals with data held in an IPFS node, and to perform retrievals directly into an IPFS node. NOTE: because the "staging" and "client" Badger blockstores are no longer used, existing imports on the client will be rendered useless. On startup, Lotus will enumerate all imports and print WARN statements on the log for each import that needs to be reimported. These log lines contain these messages: - import lacks carv2 path; import will not work; please reimport - import has missing/broken carv2; please reimport At the end, we will print a "sanity check completed" message indicating the count of imports found, and how many were deemed broken. Co-authored-by: Aarsh Shah <aarshkshah1992@gmail.com> Co-authored-by: Dirk McCormick <dirkmdev@gmail.com> Co-authored-by: Raúl Kripalani <raul@protocol.ai> Co-authored-by: Dirk McCormick <dirkmdev@gmail.com>
2021-08-16 22:34:32 +00:00
N int
FastRetrieval bool
CarExport bool
StartEpoch abi.ChainEpoch
UseCARFileForStorageDeal bool
IndexProvider *shared_testutil.MockIndexProvider
2021-06-23 17:13:29 +00:00
}
func (dh *DealHarness) RunConcurrentDeals(opts RunConcurrentDealsOpts) {
2021-09-20 06:04:21 +00:00
ctx := context.Background()
2021-06-23 17:13:29 +00:00
errgrp, _ := errgroup.WithContext(context.Background())
for i := 0; i < opts.N; i++ {
i := i
errgrp.Go(func() (err error) {
defer dh.t.Logf("finished concurrent deal %d/%d", i, opts.N)
2021-06-23 17:13:29 +00:00
defer func() {
// This is necessary because golang can't deal with test
// failures being reported from children goroutines ¯\_(ツ)_/¯
if r := recover(); r != nil {
err = fmt.Errorf("deal failed: %s", r)
}
}()
dh.t.Logf("making storage deal %d/%d", i, opts.N)
2021-06-23 17:13:29 +00:00
deal, res, inPath := dh.MakeOnlineDeal(context.Background(), MakeFullDealParams{
integrate DAG store and CARv2 in deal-making (#6671) This commit removes badger from the deal-making processes, and moves to a new architecture with the dagstore as the cental component on the miner-side, and CARv2s on the client-side. Every deal that has been handed off to the sealing subsystem becomes a shard in the dagstore. Shards are mounted via the LotusMount, which teaches the dagstore how to load the related piece when serving retrievals. When the miner starts the Lotus for the first time with this patch, we will perform a one-time migration of all active deals into the dagstore. This is a lightweight process, and it consists simply of registering the shards in the dagstore. Shards are backed by the unsealed copy of the piece. This is currently a CARv1. However, the dagstore keeps CARv2 indices for all pieces, so when it's time to acquire a shard to serve a retrieval, the unsealed CARv1 is joined with its index (safeguarded by the dagstore), to form a read-only blockstore, thus taking the place of the monolithic badger. Data transfers have been adjusted to interface directly with CARv2 files. On inbound transfers (client retrievals, miner storage deals), we stream the received data into a CARv2 ReadWrite blockstore. On outbound transfers (client storage deals, miner retrievals), we serve the data off a CARv2 ReadOnly blockstore. Client-side imports are managed by the refactored *imports.Manager component (when not using IPFS integration). Just like it before, we use the go-filestore library to avoid duplicating the data from the original file in the resulting UnixFS DAG (concretely the leaves). However, the target of those imports are what we call "ref-CARv2s": CARv2 files placed under the `$LOTUS_PATH/imports` directory, containing the intermediate nodes in full, and the leaves as positional references to the original file on disk. Client-side retrievals are placed into CARv2 files in the location: `$LOTUS_PATH/retrievals`. A new set of `Dagstore*` JSON-RPC operations and `lotus-miner dagstore` subcommands have been introduced on the miner-side to inspect and manage the dagstore. Despite moving to a CARv2-backed system, the IPFS integration has been respected, and it continues to be possible to make storage deals with data held in an IPFS node, and to perform retrievals directly into an IPFS node. NOTE: because the "staging" and "client" Badger blockstores are no longer used, existing imports on the client will be rendered useless. On startup, Lotus will enumerate all imports and print WARN statements on the log for each import that needs to be reimported. These log lines contain these messages: - import lacks carv2 path; import will not work; please reimport - import has missing/broken carv2; please reimport At the end, we will print a "sanity check completed" message indicating the count of imports found, and how many were deemed broken. Co-authored-by: Aarsh Shah <aarshkshah1992@gmail.com> Co-authored-by: Dirk McCormick <dirkmdev@gmail.com> Co-authored-by: Raúl Kripalani <raul@protocol.ai> Co-authored-by: Dirk McCormick <dirkmdev@gmail.com>
2021-08-16 22:34:32 +00:00
Rseed: 5 + i,
FastRet: opts.FastRetrieval,
StartEpoch: opts.StartEpoch,
UseCARFileForStorageDeal: opts.UseCARFileForStorageDeal,
2021-06-23 17:13:29 +00:00
})
2021-11-10 16:28:23 +00:00
// Check that the storage provider announced the deal to indexers
if opts.IndexProvider != nil {
notifs := opts.IndexProvider.GetNotifs()
2021-11-10 16:28:23 +00:00
_, ok := notifs[string(deal.Bytes())]
require.True(dh.t, ok)
}
dh.t.Logf("retrieving deal %d/%d", i, opts.N)
2021-06-23 17:13:29 +00:00
outPath := dh.PerformRetrieval(context.Background(), deal, res.Root, opts.CarExport)
2021-09-20 06:04:21 +00:00
if opts.CarExport {
f, err := os.Open(outPath)
require.NoError(dh.t, err)
actualFile := dh.ExtractFileFromCAR(ctx, f)
require.NoError(dh.t, f.Close())
2022-06-29 17:07:29 +00:00
AssertFilesEqual(dh.t, inPath, actualFile)
2021-09-20 06:04:21 +00:00
} else {
AssertFilesEqual(dh.t, inPath, outPath)
}
2021-06-23 17:13:29 +00:00
return nil
})
}
require.NoError(dh.t, errgrp.Wait())
}