Merge branch 'bloxico/system-test-matrix' of https://github.com/filecoin-project/lotus into merge_lotus
This commit is contained in:
Darko Brdareski
+354
-265
@@ -4,17 +4,22 @@ import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
bstore "github.com/ipfs/go-ipfs-blockstore"
|
||||
format "github.com/ipfs/go-ipld-format"
|
||||
unixfile "github.com/ipfs/go-unixfs/file"
|
||||
"github.com/ipld/go-car"
|
||||
"github.com/ipld/go-car/util"
|
||||
carv2 "github.com/ipld/go-car/v2"
|
||||
carv2bs "github.com/ipld/go-car/v2/blockstore"
|
||||
"github.com/ipld/go-ipld-prime/datamodel"
|
||||
"golang.org/x/xerrors"
|
||||
|
||||
"github.com/filecoin-project/go-padreader"
|
||||
@@ -58,7 +63,6 @@ import (
|
||||
"github.com/filecoin-project/go-state-types/abi"
|
||||
"github.com/filecoin-project/specs-actors/v3/actors/builtin/market"
|
||||
|
||||
marketevents "github.com/filecoin-project/lotus/markets/loggers"
|
||||
"github.com/filecoin-project/lotus/node/config"
|
||||
"github.com/filecoin-project/lotus/node/repo/imports"
|
||||
|
||||
@@ -146,7 +150,7 @@ func (a *API) dealStarter(ctx context.Context, params *api.StartDealParams, isSt
|
||||
if err != nil {
|
||||
return nil, xerrors.Errorf("failed to find blockstore for root CID: %w", err)
|
||||
}
|
||||
if has, err := bs.Has(params.Data.Root); err != nil {
|
||||
if has, err := bs.Has(ctx, params.Data.Root); err != nil {
|
||||
return nil, xerrors.Errorf("failed to query blockstore for root CID: %w", err)
|
||||
} else if !has {
|
||||
return nil, xerrors.Errorf("failed to find root CID in blockstore: %w", err)
|
||||
@@ -516,7 +520,7 @@ func (a *API) ClientImport(ctx context.Context, ref api.FileRef) (res *api.Impor
|
||||
}
|
||||
defer f.Close() //nolint:errcheck
|
||||
|
||||
hd, _, err := car.ReadHeader(bufio.NewReader(f))
|
||||
hd, err := car.ReadHeader(bufio.NewReader(f))
|
||||
if err != nil {
|
||||
return nil, xerrors.Errorf("failed to read CAR header: %w", err)
|
||||
}
|
||||
@@ -760,325 +764,405 @@ func (a *API) ClientCancelRetrievalDeal(ctx context.Context, dealID rm.DealID) e
|
||||
}
|
||||
}
|
||||
|
||||
func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref *api.FileRef) error {
|
||||
events := make(chan marketevents.RetrievalEvent)
|
||||
go a.clientRetrieve(ctx, order, ref, events)
|
||||
|
||||
for {
|
||||
select {
|
||||
case evt, ok := <-events:
|
||||
if !ok { // done successfully
|
||||
return nil
|
||||
}
|
||||
|
||||
if evt.Err != "" {
|
||||
return xerrors.Errorf("retrieval failed: %s", evt.Err)
|
||||
}
|
||||
case <-ctx.Done():
|
||||
return xerrors.Errorf("retrieval timed out")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (a *API) ClientRetrieveWithEvents(ctx context.Context, order api.RetrievalOrder, ref *api.FileRef) (<-chan marketevents.RetrievalEvent, error) {
|
||||
events := make(chan marketevents.RetrievalEvent)
|
||||
go a.clientRetrieve(ctx, order, ref, events)
|
||||
return events, nil
|
||||
}
|
||||
|
||||
type retrievalSubscribeEvent struct {
|
||||
event rm.ClientEvent
|
||||
state rm.ClientDealState
|
||||
}
|
||||
|
||||
func consumeAllEvents(ctx context.Context, dealID rm.DealID, subscribeEvents chan retrievalSubscribeEvent, events chan marketevents.RetrievalEvent) error {
|
||||
for {
|
||||
var subscribeEvent retrievalSubscribeEvent
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return xerrors.New("Retrieval Timed Out")
|
||||
case subscribeEvent = <-subscribeEvents:
|
||||
if subscribeEvent.state.ID != dealID {
|
||||
// we can't check the deal ID ahead of time because:
|
||||
// 1. We need to subscribe before retrieving.
|
||||
// 2. We won't know the deal ID until after retrieving.
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return xerrors.New("Retrieval Timed Out")
|
||||
case events <- marketevents.RetrievalEvent{
|
||||
Event: subscribeEvent.event,
|
||||
Status: subscribeEvent.state.Status,
|
||||
BytesReceived: subscribeEvent.state.TotalReceived,
|
||||
FundsSpent: subscribeEvent.state.FundsSpent,
|
||||
}:
|
||||
}
|
||||
|
||||
state := subscribeEvent.state
|
||||
switch state.Status {
|
||||
case rm.DealStatusCompleted:
|
||||
return nil
|
||||
case rm.DealStatusRejected:
|
||||
return xerrors.Errorf("Retrieval Proposal Rejected: %s", state.Message)
|
||||
case rm.DealStatusCancelled:
|
||||
return xerrors.Errorf("Retrieval was cancelled externally: %s", state.Message)
|
||||
case
|
||||
rm.DealStatusDealNotFound,
|
||||
rm.DealStatusErrored:
|
||||
return xerrors.Errorf("Retrieval Error: %s", state.Message)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref *api.FileRef, events chan marketevents.RetrievalEvent) {
|
||||
defer close(events)
|
||||
|
||||
finish := func(e error) {
|
||||
if e != nil {
|
||||
events <- marketevents.RetrievalEvent{Err: e.Error(), FundsSpent: big.Zero()}
|
||||
}
|
||||
}
|
||||
|
||||
func getDataSelector(dps *api.Selector, matchPath bool) (datamodel.Node, error) {
|
||||
sel := selectorparse.CommonSelector_ExploreAllRecursively
|
||||
if order.DatamodelPathSelector != nil {
|
||||
if dps != nil {
|
||||
|
||||
ssb := builder.NewSelectorSpecBuilder(basicnode.Prototype.Any)
|
||||
if strings.HasPrefix(string(*dps), "{") {
|
||||
var err error
|
||||
sel, err = selectorparse.ParseJSONSelector(string(*dps))
|
||||
if err != nil {
|
||||
return nil, xerrors.Errorf("failed to parse json-selector '%s': %w", *dps, err)
|
||||
}
|
||||
} else {
|
||||
ssb := builder.NewSelectorSpecBuilder(basicnode.Prototype.Any)
|
||||
|
||||
selspec, err := textselector.SelectorSpecFromPath(
|
||||
selspec, err := textselector.SelectorSpecFromPath(
|
||||
textselector.Expression(*dps), matchPath,
|
||||
|
||||
*order.DatamodelPathSelector,
|
||||
ssb.ExploreRecursive(
|
||||
selector.RecursionLimitNone(),
|
||||
ssb.ExploreUnion(ssb.Matcher(), ssb.ExploreAll(ssb.ExploreRecursiveEdge())),
|
||||
),
|
||||
)
|
||||
if err != nil {
|
||||
return nil, xerrors.Errorf("failed to parse text-selector '%s': %w", *dps, err)
|
||||
}
|
||||
|
||||
// URGH - this is a direct copy from https://github.com/filecoin-project/go-fil-markets/blob/v1.12.0/shared/selectors.go#L10-L16
|
||||
// Unable to use it because we need the SelectorSpec, and markets exposes just a reified node
|
||||
ssb.ExploreRecursive(
|
||||
selector.RecursionLimitNone(),
|
||||
ssb.ExploreAll(ssb.ExploreRecursiveEdge()),
|
||||
),
|
||||
)
|
||||
if err != nil {
|
||||
finish(xerrors.Errorf("failed to parse text-selector '%s': %w", *order.DatamodelPathSelector, err))
|
||||
return
|
||||
sel = selspec.Node()
|
||||
log.Infof("partial retrieval of datamodel-path-selector %s/*", *dps)
|
||||
}
|
||||
|
||||
sel = selspec.Node()
|
||||
log.Infof("partial retrieval of datamodel-path-selector %s/*", *order.DatamodelPathSelector)
|
||||
}
|
||||
|
||||
// summary:
|
||||
// 1. if we're retrieving from an import, FromLocalCAR will be set.
|
||||
// Skip the retrieval itself, and use the provided car as a blockstore further down
|
||||
// to extract a CAR or UnixFS export from.
|
||||
// 2. if we're using an IPFS blockstore for retrieval, retrieve into it,
|
||||
// then use the virtual blockstore to extract a CAR or UnixFS export from it.
|
||||
// 3. if we have to retrieve, perform a CARv2 retrieval, then either
|
||||
// extract the CARv1 (with ExtractV1File) or use it as a blockstore further down.
|
||||
return sel, nil
|
||||
}
|
||||
|
||||
// this indicates we're proxying to IPFS.
|
||||
func (a *API) ClientRetrieve(ctx context.Context, params api.RetrievalOrder) (*api.RestrievalRes, error) {
|
||||
sel, err := getDataSelector(params.DataSelector, false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
di, err := a.doRetrieval(ctx, params, sel)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &api.RestrievalRes{
|
||||
DealID: di,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (a *API) doRetrieval(ctx context.Context, order api.RetrievalOrder, sel datamodel.Node) (rm.DealID, error) {
|
||||
if order.MinerPeer == nil || order.MinerPeer.ID == "" {
|
||||
mi, err := a.StateMinerInfo(ctx, order.Miner, types.EmptyTSK)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
order.MinerPeer = &rm.RetrievalPeer{
|
||||
ID: *mi.PeerId,
|
||||
Address: order.Miner,
|
||||
}
|
||||
}
|
||||
|
||||
if order.Total.Int == nil {
|
||||
return 0, xerrors.Errorf("cannot make retrieval deal for null total")
|
||||
}
|
||||
|
||||
if order.Size == 0 {
|
||||
return 0, xerrors.Errorf("cannot make retrieval deal for zero bytes")
|
||||
}
|
||||
|
||||
ppb := types.BigDiv(order.Total, types.NewInt(order.Size))
|
||||
|
||||
params, err := rm.NewParamsV1(ppb, order.PaymentInterval, order.PaymentIntervalIncrease, sel, order.Piece, order.UnsealPrice)
|
||||
if err != nil {
|
||||
return 0, xerrors.Errorf("Error in retrieval params: %s", err)
|
||||
}
|
||||
|
||||
id := a.Retrieval.NextID()
|
||||
id, err = a.Retrieval.Retrieve(
|
||||
ctx,
|
||||
id,
|
||||
order.Root,
|
||||
params,
|
||||
order.Total,
|
||||
*order.MinerPeer,
|
||||
order.Client,
|
||||
order.Miner,
|
||||
)
|
||||
|
||||
if err != nil {
|
||||
return 0, xerrors.Errorf("Retrieve failed: %w", err)
|
||||
}
|
||||
|
||||
return id, nil
|
||||
}
|
||||
|
||||
func (a *API) ClientRetrieveWait(ctx context.Context, deal rm.DealID) error {
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
|
||||
subscribeEvents := make(chan rm.ClientDealState, 1)
|
||||
|
||||
unsubscribe := a.Retrieval.SubscribeToEvents(func(event rm.ClientEvent, state rm.ClientDealState) {
|
||||
// We'll check the deal IDs inside consumeAllEvents.
|
||||
if state.ID != deal {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case subscribeEvents <- state:
|
||||
}
|
||||
})
|
||||
defer unsubscribe()
|
||||
|
||||
{
|
||||
state, err := a.Retrieval.GetDeal(deal)
|
||||
if err != nil {
|
||||
return xerrors.Errorf("getting deal state: %w", err)
|
||||
}
|
||||
select {
|
||||
case subscribeEvents <- state:
|
||||
default: // already have an event queued from the subscription
|
||||
}
|
||||
}
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return xerrors.New("Retrieval Timed Out")
|
||||
case state := <-subscribeEvents:
|
||||
switch state.Status {
|
||||
case rm.DealStatusCompleted:
|
||||
return nil
|
||||
case rm.DealStatusRejected:
|
||||
return xerrors.Errorf("Retrieval Proposal Rejected: %s", state.Message)
|
||||
case rm.DealStatusCancelled:
|
||||
return xerrors.Errorf("Retrieval was cancelled externally: %s", state.Message)
|
||||
case
|
||||
rm.DealStatusDealNotFound,
|
||||
rm.DealStatusErrored:
|
||||
return xerrors.Errorf("Retrieval Error: %s", state.Message)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type ExportDest struct {
|
||||
Writer io.Writer
|
||||
Path string
|
||||
}
|
||||
|
||||
func (ed *ExportDest) doWrite(cb func(io.Writer) error) error {
|
||||
if ed.Writer != nil {
|
||||
return cb(ed.Writer)
|
||||
}
|
||||
|
||||
f, err := os.OpenFile(ed.Path, os.O_CREATE|os.O_WRONLY, 0644)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := cb(f); err != nil {
|
||||
_ = f.Close()
|
||||
return err
|
||||
}
|
||||
|
||||
return f.Close()
|
||||
}
|
||||
|
||||
func (a *API) ClientExport(ctx context.Context, exportRef api.ExportRef, ref api.FileRef) error {
|
||||
return a.ClientExportInto(ctx, exportRef, ref.IsCAR, ExportDest{Path: ref.Path})
|
||||
}
|
||||
|
||||
func (a *API) ClientExportInto(ctx context.Context, exportRef api.ExportRef, car bool, dest ExportDest) error {
|
||||
proxyBss, retrieveIntoIPFS := a.RtvlBlockstoreAccessor.(*retrievaladapter.ProxyBlockstoreAccessor)
|
||||
|
||||
carBss, retrieveIntoCAR := a.RtvlBlockstoreAccessor.(*retrievaladapter.CARBlockstoreAccessor)
|
||||
carPath := exportRef.FromLocalCAR
|
||||
|
||||
carPath := order.FromLocalCAR
|
||||
|
||||
// we actually need to retrieve from the network
|
||||
if carPath == "" {
|
||||
|
||||
if !retrieveIntoIPFS && !retrieveIntoCAR {
|
||||
// we don't recognize the blockstore accessor.
|
||||
finish(xerrors.Errorf("unsupported retrieval blockstore accessor"))
|
||||
return
|
||||
}
|
||||
|
||||
if order.MinerPeer == nil || order.MinerPeer.ID == "" {
|
||||
mi, err := a.StateMinerInfo(ctx, order.Miner, types.EmptyTSK)
|
||||
if err != nil {
|
||||
finish(err)
|
||||
return
|
||||
}
|
||||
|
||||
order.MinerPeer = &rm.RetrievalPeer{
|
||||
ID: *mi.PeerId,
|
||||
Address: order.Miner,
|
||||
}
|
||||
}
|
||||
|
||||
if order.Total.Int == nil {
|
||||
finish(xerrors.Errorf("cannot make retrieval deal for null total"))
|
||||
return
|
||||
}
|
||||
|
||||
if order.Size == 0 {
|
||||
finish(xerrors.Errorf("cannot make retrieval deal for zero bytes"))
|
||||
return
|
||||
}
|
||||
|
||||
ppb := types.BigDiv(order.Total, types.NewInt(order.Size))
|
||||
|
||||
params, err := rm.NewParamsV1(ppb, order.PaymentInterval, order.PaymentIntervalIncrease, sel, order.Piece, order.UnsealPrice)
|
||||
if err != nil {
|
||||
finish(xerrors.Errorf("Error in retrieval params: %s", err))
|
||||
return
|
||||
}
|
||||
|
||||
// Subscribe to events before retrieving to avoid losing events.
|
||||
subscribeEvents := make(chan retrievalSubscribeEvent, 1)
|
||||
subscribeCtx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
unsubscribe := a.Retrieval.SubscribeToEvents(func(event rm.ClientEvent, state rm.ClientDealState) {
|
||||
// We'll check the deal IDs inside consumeAllEvents.
|
||||
if state.PayloadCID.Equals(order.Root) {
|
||||
select {
|
||||
case <-subscribeCtx.Done():
|
||||
case subscribeEvents <- retrievalSubscribeEvent{event, state}:
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
id := a.Retrieval.NextID()
|
||||
id, err = a.Retrieval.Retrieve(
|
||||
ctx,
|
||||
id,
|
||||
order.Root,
|
||||
params,
|
||||
order.Total,
|
||||
*order.MinerPeer,
|
||||
order.Client,
|
||||
order.Miner,
|
||||
)
|
||||
|
||||
if err != nil {
|
||||
unsubscribe()
|
||||
finish(xerrors.Errorf("Retrieve failed: %w", err))
|
||||
return
|
||||
}
|
||||
|
||||
err = consumeAllEvents(ctx, id, subscribeEvents, events)
|
||||
|
||||
unsubscribe()
|
||||
if err != nil {
|
||||
finish(xerrors.Errorf("Retrieve: %w", err))
|
||||
return
|
||||
return xerrors.Errorf("unsupported retrieval blockstore accessor")
|
||||
}
|
||||
|
||||
if retrieveIntoCAR {
|
||||
carPath = carBss.PathFor(id)
|
||||
carPath = carBss.PathFor(exportRef.DealID)
|
||||
}
|
||||
}
|
||||
|
||||
if ref == nil {
|
||||
// If ref is nil, it only fetches the data into the configured blockstore
|
||||
// (if fetching from network).
|
||||
finish(nil)
|
||||
return
|
||||
}
|
||||
|
||||
// determine where did the retrieval go
|
||||
var retrievalBs bstore.Blockstore
|
||||
if retrieveIntoIPFS {
|
||||
retrievalBs = proxyBss.Blockstore
|
||||
} else {
|
||||
cbs, err := stores.ReadOnlyFilestore(carPath)
|
||||
if err != nil {
|
||||
finish(err)
|
||||
return
|
||||
return err
|
||||
}
|
||||
defer cbs.Close() //nolint:errcheck
|
||||
retrievalBs = cbs
|
||||
}
|
||||
|
||||
dserv := merkledag.NewDAGService(blockservice.New(retrievalBs, offline.Exchange(retrievalBs)))
|
||||
|
||||
// Are we outputting a CAR?
|
||||
if ref.IsCAR {
|
||||
|
||||
if car {
|
||||
// not IPFS and we do full selection - just extract the CARv1 from the CARv2 we stored the retrieval in
|
||||
if !retrieveIntoIPFS && order.DatamodelPathSelector == nil {
|
||||
finish(carv2.ExtractV1File(carPath, ref.Path))
|
||||
return
|
||||
if !retrieveIntoIPFS && len(exportRef.DAGs) == 0 && dest.Writer == nil {
|
||||
return carv2.ExtractV1File(carPath, dest.Path)
|
||||
}
|
||||
|
||||
// generating a CARv1 from the configured blockstore
|
||||
f, err := os.OpenFile(ref.Path, os.O_CREATE|os.O_WRONLY, 0644)
|
||||
if err != nil {
|
||||
finish(err)
|
||||
return
|
||||
}
|
||||
|
||||
err = car.NewSelectiveCar(
|
||||
ctx,
|
||||
retrievalBs,
|
||||
[]car.Dag{{
|
||||
Root: order.Root,
|
||||
Selector: sel,
|
||||
}},
|
||||
car.MaxTraversalLinks(config.MaxTraversalLinks),
|
||||
).Write(f)
|
||||
if err != nil {
|
||||
finish(err)
|
||||
return
|
||||
}
|
||||
|
||||
finish(f.Close())
|
||||
return
|
||||
}
|
||||
|
||||
// we are extracting a UnixFS file.
|
||||
ds := merkledag.NewDAGService(blockservice.New(retrievalBs, offline.Exchange(retrievalBs)))
|
||||
root := order.Root
|
||||
roots, err := parseDagSpec(ctx, exportRef.Root, exportRef.DAGs, dserv, car)
|
||||
if err != nil {
|
||||
return xerrors.Errorf("parsing dag spec: %w", err)
|
||||
}
|
||||
if car {
|
||||
return a.outputCAR(ctx, dserv, retrievalBs, exportRef.Root, roots, dest)
|
||||
}
|
||||
|
||||
// if we used a selector - need to find the sub-root the user actually wanted to retrieve
|
||||
if order.DatamodelPathSelector != nil {
|
||||
if len(roots) != 1 {
|
||||
return xerrors.Errorf("unixfs retrieval requires one root node, got %d", len(roots))
|
||||
}
|
||||
|
||||
var subRootFound bool
|
||||
return a.outputUnixFS(ctx, roots[0].root, dserv, dest)
|
||||
}
|
||||
|
||||
// no err check - we just compiled this before starting, but now we do not wrap a `*`
|
||||
selspec, _ := textselector.SelectorSpecFromPath(*order.DatamodelPathSelector, nil) //nolint:errcheck
|
||||
func (a *API) outputCAR(ctx context.Context, ds format.DAGService, bs bstore.Blockstore, root cid.Cid, dags []dagSpec, dest ExportDest) error {
|
||||
// generating a CARv1 from the configured blockstore
|
||||
roots := make([]cid.Cid, len(dags))
|
||||
for i, dag := range dags {
|
||||
roots[i] = dag.root
|
||||
}
|
||||
|
||||
return dest.doWrite(func(w io.Writer) error {
|
||||
|
||||
if err := car.WriteHeader(&car.CarHeader{
|
||||
Roots: roots,
|
||||
Version: 1,
|
||||
}, w); err != nil {
|
||||
return fmt.Errorf("failed to write car header: %s", err)
|
||||
}
|
||||
|
||||
cs := cid.NewSet()
|
||||
|
||||
for _, dagSpec := range dags {
|
||||
if err := utils.TraverseDag(
|
||||
ctx,
|
||||
ds,
|
||||
root,
|
||||
dagSpec.selector,
|
||||
func(p traversal.Progress, n ipld.Node, r traversal.VisitReason) error {
|
||||
if r == traversal.VisitReason_SelectionMatch {
|
||||
var c cid.Cid
|
||||
if p.LastBlock.Link == nil {
|
||||
c = root
|
||||
} else {
|
||||
cidLnk, castOK := p.LastBlock.Link.(cidlink.Link)
|
||||
if !castOK {
|
||||
return xerrors.Errorf("cidlink cast unexpectedly failed on '%s'", p.LastBlock.Link)
|
||||
}
|
||||
|
||||
c = cidLnk.Cid
|
||||
}
|
||||
|
||||
if cs.Visit(c) {
|
||||
nb, err := bs.Get(ctx, c)
|
||||
if err != nil {
|
||||
return xerrors.Errorf("getting block data: %w", err)
|
||||
}
|
||||
|
||||
err = util.LdWrite(w, c.Bytes(), nb.RawData())
|
||||
if err != nil {
|
||||
return xerrors.Errorf("writing block data: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
return nil
|
||||
},
|
||||
); err != nil {
|
||||
return xerrors.Errorf("error while traversing car dag: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
func (a *API) outputUnixFS(ctx context.Context, root cid.Cid, ds format.DAGService, dest ExportDest) error {
|
||||
nd, err := ds.Get(ctx, root)
|
||||
if err != nil {
|
||||
return xerrors.Errorf("ClientRetrieve: %w", err)
|
||||
}
|
||||
file, err := unixfile.NewUnixfsFile(ctx, ds, nd)
|
||||
if err != nil {
|
||||
return xerrors.Errorf("ClientRetrieve: %w", err)
|
||||
}
|
||||
|
||||
if dest.Writer == nil {
|
||||
return files.WriteTo(file, dest.Path)
|
||||
}
|
||||
|
||||
switch f := file.(type) {
|
||||
case files.File:
|
||||
_, err = io.Copy(dest.Writer, f)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
default:
|
||||
return fmt.Errorf("file type %T is not supported", nd)
|
||||
}
|
||||
}
|
||||
|
||||
type dagSpec struct {
|
||||
root cid.Cid
|
||||
selector ipld.Node
|
||||
}
|
||||
|
||||
func parseDagSpec(ctx context.Context, root cid.Cid, dsp []api.DagSpec, ds format.DAGService, car bool) ([]dagSpec, error) {
|
||||
if len(dsp) == 0 {
|
||||
return []dagSpec{
|
||||
{
|
||||
root: root,
|
||||
selector: nil,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
out := make([]dagSpec, len(dsp))
|
||||
for i, spec := range dsp {
|
||||
|
||||
if spec.DataSelector == nil {
|
||||
return nil, xerrors.Errorf("invalid DagSpec at position %d: `DataSelector` can not be nil", i)
|
||||
}
|
||||
|
||||
// reify selector
|
||||
var err error
|
||||
out[i].selector, err = getDataSelector(spec.DataSelector, car && spec.ExportMerkleProof)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// find the pointed-at root node within the containing ds
|
||||
var rsn ipld.Node
|
||||
|
||||
if strings.HasPrefix(string(*spec.DataSelector), "{") {
|
||||
var err error
|
||||
rsn, err = selectorparse.ParseJSONSelector(string(*spec.DataSelector))
|
||||
if err != nil {
|
||||
return nil, xerrors.Errorf("failed to parse json-selector '%s': %w", *spec.DataSelector, err)
|
||||
}
|
||||
} else {
|
||||
selspec, _ := textselector.SelectorSpecFromPath(textselector.Expression(*spec.DataSelector), car && spec.ExportMerkleProof, nil) //nolint:errcheck
|
||||
rsn = selspec.Node()
|
||||
}
|
||||
|
||||
var newRoot cid.Cid
|
||||
var errHalt = errors.New("halt walk")
|
||||
if err := utils.TraverseDag(
|
||||
ctx,
|
||||
ds,
|
||||
root,
|
||||
selspec.Node(),
|
||||
rsn,
|
||||
func(p traversal.Progress, n ipld.Node, r traversal.VisitReason) error {
|
||||
if r == traversal.VisitReason_SelectionMatch {
|
||||
|
||||
if p.LastBlock.Path.String() != p.Path.String() {
|
||||
if !car && p.LastBlock.Path.String() != p.Path.String() {
|
||||
return xerrors.Errorf("unsupported selection path '%s' does not correspond to a block boundary (a.k.a. CID link)", p.Path.String())
|
||||
}
|
||||
|
||||
if p.LastBlock.Link == nil {
|
||||
// this is likely the root node that we've matched here
|
||||
newRoot = root
|
||||
return errHalt
|
||||
}
|
||||
|
||||
cidLnk, castOK := p.LastBlock.Link.(cidlink.Link)
|
||||
if !castOK {
|
||||
return xerrors.Errorf("cidlink cast unexpectedly failed on '%s'", p.LastBlock.Link.String())
|
||||
return xerrors.Errorf("cidlink cast unexpectedly failed on '%s'", p.LastBlock.Link)
|
||||
}
|
||||
|
||||
root = cidLnk.Cid
|
||||
subRootFound = true
|
||||
newRoot = cidLnk.Cid
|
||||
|
||||
return errHalt
|
||||
}
|
||||
return nil
|
||||
},
|
||||
); err != nil {
|
||||
finish(xerrors.Errorf("error while locating partial retrieval sub-root: %w", err))
|
||||
return
|
||||
); err != nil && err != errHalt {
|
||||
return nil, xerrors.Errorf("error while locating partial retrieval sub-root: %w", err)
|
||||
}
|
||||
|
||||
if !subRootFound {
|
||||
finish(xerrors.Errorf("path selection '%s' does not match a node within %s", *order.DatamodelPathSelector, root))
|
||||
return
|
||||
if newRoot == cid.Undef {
|
||||
return nil, xerrors.Errorf("path selection does not match a node within %s", root)
|
||||
}
|
||||
|
||||
out[i].root = newRoot
|
||||
}
|
||||
|
||||
nd, err := ds.Get(ctx, root)
|
||||
if err != nil {
|
||||
finish(xerrors.Errorf("ClientRetrieve: %w", err))
|
||||
return
|
||||
}
|
||||
file, err := unixfile.NewUnixfsFile(ctx, ds, nd)
|
||||
if err != nil {
|
||||
finish(xerrors.Errorf("ClientRetrieve: %w", err))
|
||||
return
|
||||
}
|
||||
|
||||
finish(files.WriteTo(file, ref.Path))
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (a *API) ClientListRetrievals(ctx context.Context) ([]api.RetrievalInfo, error) {
|
||||
@@ -1110,8 +1194,13 @@ func (a *API) ClientListRetrievals(ctx context.Context) ([]api.RetrievalInfo, er
|
||||
func (a *API) ClientGetRetrievalUpdates(ctx context.Context) (<-chan api.RetrievalInfo, error) {
|
||||
updates := make(chan api.RetrievalInfo)
|
||||
|
||||
unsub := a.Retrieval.SubscribeToEvents(func(_ rm.ClientEvent, deal rm.ClientDealState) {
|
||||
updates <- a.newRetrievalInfo(ctx, deal)
|
||||
unsub := a.Retrieval.SubscribeToEvents(func(evt rm.ClientEvent, deal rm.ClientDealState) {
|
||||
update := a.newRetrievalInfo(ctx, deal)
|
||||
update.Event = &evt
|
||||
select {
|
||||
case updates <- update:
|
||||
case <-ctx.Done():
|
||||
}
|
||||
})
|
||||
|
||||
go func() {
|
||||
@@ -1196,7 +1285,7 @@ func (a *API) ClientCalcCommP(ctx context.Context, inpath string) (*api.CommPRet
|
||||
}
|
||||
|
||||
// check that the data is a car file; if it's not, retrieval won't work
|
||||
_, _, err = car.ReadHeader(bufio.NewReader(rdr))
|
||||
_, err = car.ReadHeader(bufio.NewReader(rdr))
|
||||
if err != nil {
|
||||
return nil, xerrors.Errorf("not a car file: %w", err)
|
||||
}
|
||||
|
||||
@@ -64,7 +64,7 @@ func TestImportLocal(t *testing.T) {
|
||||
require.NoError(t, err)
|
||||
require.True(t, local)
|
||||
|
||||
order := api.RetrievalOrder{
|
||||
order := api.ExportRef{
|
||||
Root: root,
|
||||
FromLocalCAR: it.CARPath,
|
||||
}
|
||||
@@ -72,8 +72,7 @@ func TestImportLocal(t *testing.T) {
|
||||
// retrieve as UnixFS.
|
||||
out1 := filepath.Join(dir, "retrieval1.data") // as unixfs
|
||||
out2 := filepath.Join(dir, "retrieval2.data") // as car
|
||||
//stm: @CLIENT_RETRIEVAL_RETRIEVE_001
|
||||
err = a.ClientRetrieve(ctx, order, &api.FileRef{
|
||||
err = a.ClientExport(ctx, order, api.FileRef{
|
||||
Path: out1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
@@ -82,7 +81,7 @@ func TestImportLocal(t *testing.T) {
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, b, outBytes)
|
||||
|
||||
err = a.ClientRetrieve(ctx, order, &api.FileRef{
|
||||
err = a.ClientExport(ctx, order, api.FileRef{
|
||||
Path: out2,
|
||||
IsCAR: true,
|
||||
})
|
||||
@@ -112,7 +111,7 @@ func TestImportLocal(t *testing.T) {
|
||||
// recreate the unixfs dag, and see if it matches the original file byte by byte
|
||||
// import the car into a memory blockstore, then export the unixfs file.
|
||||
bs := blockstore.NewBlockstore(datastore.NewMapDatastore())
|
||||
_, err = car.LoadCar(bs, exported.DataReader())
|
||||
_, err = car.LoadCar(ctx, bs, exported.DataReader())
|
||||
require.NoError(t, err)
|
||||
|
||||
dag := merkledag.NewDAGService(blockservice.New(bs, offline.Exchange(bs)))
|
||||
|
||||
Reference in New Issue
Block a user