lotus/retrieval/client.go
Jakub Sztandera 1ed62628a7
Update go-log to v2
Signed-off-by: Jakub Sztandera <kubuxu@protocol.ai>
2020-01-08 20:22:31 +01:00

273 lines
6.8 KiB
Go

package retrieval
import (
"context"
"io"
blocks "github.com/ipfs/go-block-format"
"github.com/ipfs/go-cid"
logging "github.com/ipfs/go-log/v2"
"github.com/libp2p/go-libp2p-core/host"
"github.com/libp2p/go-libp2p-core/network"
"github.com/libp2p/go-libp2p-core/peer"
cbg "github.com/whyrusleeping/cbor-gen"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-cbor-util"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/types"
payapi "github.com/filecoin-project/lotus/node/impl/paych"
"github.com/filecoin-project/lotus/paych"
"github.com/filecoin-project/lotus/retrieval/discovery"
)
var log = logging.Logger("retrieval")
type Client struct {
h host.Host
pmgr *paych.Manager
payapi payapi.PaychAPI
}
func NewClient(h host.Host, pmgr *paych.Manager, payapi payapi.PaychAPI) *Client {
return &Client{h: h, pmgr: pmgr, payapi: payapi}
}
func (c *Client) Query(ctx context.Context, p discovery.RetrievalPeer, data cid.Cid) api.QueryOffer {
s, err := c.h.NewStream(ctx, p.ID, QueryProtocolID)
if err != nil {
log.Warn(err)
return api.QueryOffer{Err: err.Error(), Miner: p.Address, MinerPeerID: p.ID}
}
defer s.Close()
err = cborutil.WriteCborRPC(s, &Query{
Piece: data,
})
if err != nil {
log.Warn(err)
return api.QueryOffer{Err: err.Error(), Miner: p.Address, MinerPeerID: p.ID}
}
var resp QueryResponse
if err := resp.UnmarshalCBOR(s); err != nil {
log.Warn(err)
return api.QueryOffer{Err: err.Error(), Miner: p.Address, MinerPeerID: p.ID}
}
return api.QueryOffer{
Root: data,
Size: resp.Size,
MinPrice: resp.MinPrice,
Miner: p.Address, // TODO: check
MinerPeerID: p.ID,
}
}
type clientStream struct {
payapi payapi.PaychAPI
stream network.Stream
peeker cbg.BytePeeker
root cid.Cid
size types.BigInt
offset uint64
paych address.Address
lane uint64
total types.BigInt
transferred types.BigInt
windowSize uint64 // how much we "trust" the peer
verifier BlockVerifier
}
// C > S
//
// Offset MUST be aligned on chunking boundaries, size is rounded up to leaf size
//
// > DealProposal{Mode: Unixfs0, RootCid, Offset, Size, Payment(nil if free)}
// < Resp{Accept}
// < ..(Intermediate Block)
// < ..Blocks
// < ..(Intermediate Block)
// < ..Blocks
// > DealProposal(...)
// < ...
func (c *Client) RetrieveUnixfs(ctx context.Context, root cid.Cid, size uint64, total types.BigInt, miner peer.ID, client, minerAddr address.Address, out io.Writer) error {
s, err := c.h.NewStream(ctx, miner, ProtocolID)
if err != nil {
return xerrors.Errorf("failed to open stream to miner for retrieval query: %w", err)
}
defer s.Close()
initialOffset := uint64(0) // TODO: Check how much data we have locally
// TODO: Support in handler
// TODO: Allow client to specify this
paych, _, err := c.pmgr.GetPaych(ctx, client, minerAddr, total)
if err != nil {
return xerrors.Errorf("getting payment channel: %w", err)
}
lane, err := c.pmgr.AllocateLane(paych)
if err != nil {
return xerrors.Errorf("allocating payment lane: %w", err)
}
cst := clientStream{
payapi: c.payapi,
stream: s,
peeker: cbg.GetPeeker(s),
root: root,
size: types.NewInt(size),
offset: initialOffset,
paych: paych,
lane: lane,
total: total,
transferred: types.NewInt(0),
windowSize: build.UnixfsChunkSize,
verifier: &UnixFs0Verifier{Root: root},
}
for cst.offset != size+initialOffset {
toFetch := cst.windowSize
if toFetch+cst.offset > size {
toFetch = size - cst.offset
}
log.Infof("Retrieve %dB @%d", toFetch, cst.offset)
err := cst.doOneExchange(ctx, toFetch, out)
if err != nil {
return xerrors.Errorf("retrieval exchange: %w", err)
}
cst.offset += toFetch
}
return nil
}
func (cst *clientStream) doOneExchange(ctx context.Context, toFetch uint64, out io.Writer) error {
payAmount := types.BigDiv(types.BigMul(cst.total, types.NewInt(toFetch)), cst.size)
payment, err := cst.setupPayment(ctx, payAmount)
if err != nil {
return xerrors.Errorf("setting up retrieval payment: %w", err)
}
deal := &DealProposal{
Payment: payment,
Ref: cst.root,
Params: RetParams{
Unixfs0: &Unixfs0Offer{
Offset: cst.offset,
Size: toFetch,
},
},
}
if err := cborutil.WriteCborRPC(cst.stream, deal); err != nil {
return xerrors.Errorf("sending incremental retrieval request: %w", err)
}
var resp DealResponse
if err := cborutil.ReadCborRPC(cst.peeker, &resp); err != nil {
return xerrors.Errorf("reading retrieval response: %w", err)
}
if resp.Status != Accepted {
cst.windowSize = build.UnixfsChunkSize
// TODO: apply some 'penalty' to miner 'reputation' (needs to be the same in both cases)
if resp.Status == Error {
return xerrors.Errorf("storage deal error: %s", resp.Message)
}
if resp.Status == Rejected {
return xerrors.Errorf("storage deal rejected: %s", resp.Message)
}
return xerrors.New("storage deal response had no Accepted section")
}
log.Info("Retrieval accepted, fetching blocks")
return cst.fetchBlocks(toFetch, out)
// TODO: maybe increase miner window size after success
}
func (cst *clientStream) fetchBlocks(toFetch uint64, out io.Writer) error {
blocksToFetch := (toFetch + build.UnixfsChunkSize - 1) / build.UnixfsChunkSize
for i := uint64(0); i < blocksToFetch; {
log.Infof("block %d of %d", i+1, blocksToFetch)
var block Block
if err := cborutil.ReadCborRPC(cst.peeker, &block); err != nil {
return xerrors.Errorf("reading fetchBlock response: %w", err)
}
dataBlocks, err := cst.consumeBlockMessage(block, out)
if err != nil {
return xerrors.Errorf("consuming retrieved blocks: %w", err)
}
i += dataBlocks
}
return nil
}
func (cst *clientStream) consumeBlockMessage(block Block, out io.Writer) (uint64, error) {
prefix, err := cid.PrefixFromBytes(block.Prefix)
if err != nil {
return 0, err
}
cid, err := prefix.Sum(block.Data)
if err != nil {
return 0, err
}
blk, err := blocks.NewBlockWithCid(block.Data, cid)
if err != nil {
return 0, err
}
internal, err := cst.verifier.Verify(context.TODO(), blk, out)
if err != nil {
log.Warnf("block verify failed: %s", err)
return 0, err
}
// TODO: Smarter out, maybe add to filestore automagically
// (Also, persist intermediate nodes)
if internal {
return 0, nil
}
return 1, nil
}
func (cst *clientStream) setupPayment(ctx context.Context, toSend types.BigInt) (api.PaymentInfo, error) {
amount := types.BigAdd(cst.transferred, toSend)
sv, err := cst.payapi.PaychVoucherCreate(ctx, cst.paych, amount, cst.lane)
if err != nil {
return api.PaymentInfo{}, err
}
cst.transferred = amount
return api.PaymentInfo{
Channel: cst.paych,
ChannelMessage: nil,
Vouchers: []*types.SignedVoucher{sv},
}, nil
}