datatransfer

README

go-data-transfer

A go module to perform data transfers over ipfs/go-graphsync

Description

This module encapsulates protocols for exchanging piece data between storage clients and miners, both when consummating a storage deal and when retrieving the piece later.

Table of Contents

• Background
• Usage
  • Initialize a data transfer module
  • Register a validator
  • Open a Push or Pull Request
  • Subscribe to Events
• Contribute

Background

Please see the design documentation for this module for a high-level overview and and explanation of the terms and concepts.

Usage

Requires go 1.13

Install the module in your package or app with go get "github.com/filecoin-project/go-data-transfer/datatransfer"

Initialize a data transfer module

1. Set up imports. You need, minimally, the following imports:

   package mypackage
   
   import (
       gsimpl "github.com/ipfs/go-graphsync/impl"
       datatransfer "github.com/filecoin-project/go-data-transfer/impl"
       gstransport "github.com/filecoin-project/go-data-transfer/transport/graphsync"
       "github.com/libp2p/go-libp2p-core/host"
   )
   
   

2. Provide or create a libp2p host.Host

3. You will need a transport protocol. The current default transport is graphsync. go-graphsync GraphExchange

4. Create a data transfer by building a transport interface and then initializing a new data transfer instance

   func NewGraphsyncDataTransfer(h host.Host, gs graphsync.GraphExchange) {
       tp := gstransport.NewTransport(h.ID(), gs)
       dt := impl.NewDataTransfer(h, tp)
   }
   

5. If needed, build out your voucher struct and its validator.

   A push or pull request must include a voucher. The voucher's type must have been registered with the node receiving the request before it's sent, otherwise the request will be rejected.

   datatransfer.Voucher and datatransfer.Validator are the interfaces used for validation of graphsync datatransfer messages. Voucher types plus a Validator for them must be registered with the peer to whom requests will be sent.

Example Toy Voucher and Validator

type myVoucher struct {
	data string
}

func (v *myVoucher) ToBytes() ([]byte, error) {
	return []byte(v.data), nil
}

func (v *myVoucher) FromBytes(data []byte) error {
	v.data = string(data)
	return nil
}

func (v *myVoucher) Type() string {
	return "FakeDTType"
}

type myValidator struct {
	ctx                 context.Context
	ValidationsReceived chan receivedValidation
}

func (vl *myValidator) ValidatePush(
	sender peer.ID,
	voucher datatransfer.Voucher,
	baseCid cid.Cid,
	selector ipld.Node) error {
    
    v := voucher.(*myVoucher)
    if v.data == "" || v.data != "validpush" {
        return errors.New("invalid")
    }   

	return nil
}

func (vl *myValidator) ValidatePull(
	receiver peer.ID,
	voucher datatransfer.Voucher,
	baseCid cid.Cid,
	selector ipld.Node) error {

    v := voucher.(*myVoucher)
    if v.data == "" || v.data != "validpull" {
        return errors.New("invalid")
    }   

	return nil
}

Please see go-data-transfer/blob/master/types.go for more detail.

Register a validator

Before sending push or pull requests, you must register a datatransfer.Voucher by its reflect.Type and dataTransfer.RequestValidator for vouchers that must be sent with the request. Using the trivial examples above:

    func NewGraphsyncDatatransfer(h host.Host, gs graphsync.GraphExchange) {
        tp := gstransport.NewTransport(h.ID(), gs)
        dt := impl.NewDataTransfer(h, tp)

        vouch := &myVoucher{}
        mv := &myValidator{} 
        dt.RegisterVoucherType(reflect.TypeOf(vouch), mv)
    }

For more detail, please see the unit tests.

Open a Push or Pull Request

For a push or pull request, provide a context, a datatransfer.Voucher, a host recipient peer.ID, a baseCID cid.CID and a selector ipld.Node. These calls return a datatransfer.ChannelID and any error:

    channelID, err := dtm.OpenPullDataChannel(ctx, recipient, voucher, baseCid, selector)
    // OR
    channelID, err := dtm.OpenPushDataChannel(ctx, recipient, voucher, baseCid, selector)

Subscribe to Events

The module allows the consumer to be notified when a graphsync Request is sent or a datatransfer push or pull request response is received:

    func ToySubscriberFunc (event Event, channelState ChannelState) {
        if event.Code == datatransfer.Error {
            // log error, flail about helplessly
            return
        }
        // 
        if channelState.Recipient() == our.PeerID && channelState.Received() > 0 {
            // log some stuff, update some state somewhere, send data to a channel, etc.
        }
    }

    dtm := SetupDataTransferManager(ctx, h, gs, baseCid, snode)
    unsubFunc := dtm.SubscribeToEvents(ToySubscriberFunc)

    // . . . later, when you don't need to know about events any more:
    unsubFunc()

Contributing

PRs are welcome! Please first read the design docs and look over the current code. PRs against master require approval of at least two maintainers. For the rest, please see our CONTRIBUTING guide.

License

This repository is dual-licensed under Apache 2.0 and MIT terms.

Copyright 2019. Protocol Labs, Inc.

Documentation

Index

• Constants
• Variables
• type Channel
• type ChannelID
  • func (t *ChannelID) MarshalCBOR(w io.Writer) error
  • func (c ChannelID) OtherParty(thisPeer peer.ID) peer.ID
  • func (c ChannelID) String() string
  • func (t *ChannelID) UnmarshalCBOR(r io.Reader) error
• type ChannelState
• type Event
• type EventCode
• type EventsHandler
• type Manager
• type Message
• type PauseableTransport
• type Registerable
• type Request
• type RequestValidator
• type Response
• type Revalidator
• type Status
• type Subscriber
• type TransferID
• type Transport
• type TransportConfigurer
• type TypeIdentifier
• type Unsubscribe
• type Voucher
• type VoucherResult

Constants

const EmptyTypeIdentifier = TypeIdentifier("")

EmptyTypeIdentifier means there is no voucher present

const ErrChannelNotFound = errorType("channel not found")

ErrChannelNotFound means the channel this command was issued for does not exist

const ErrHandlerAlreadySet = errorType("already set event handler")

ErrHandlerAlreadySet means an event handler was already set for this instance of hooks

const ErrHandlerNotSet = errorType("event handler has not been set")

ErrHandlerNotSet means you cannot issue commands to this interface because the handler has not been set

const ErrIncomplete = errorType("incomplete response")

ErrIncomplete indicates a channel did not finish transferring data successfully

const ErrPause = errorType("pause channel")

ErrPause is a special error that the DataReceived / DataSent hooks can use to pause the channel

const ErrRejected = errorType("response rejected")

ErrRejected indicates a request was not accepted

const ErrResume = errorType("resume channel")

ErrResume is a special error that the RequestReceived / ResponseReceived hooks can use to resume the channel

const ErrUnsupported = errorType("unsupported")

ErrUnsupported indicates an operation is not supported by the transport protocol

Variables

var Events = map[EventCode]string{
	Open:                        "Open",
	Accept:                      "Accept",
	Progress:                    "Progress",
	Cancel:                      "Cancel",
	Error:                       "Error",
	CleanupComplete:             "CleanupComplete",
	NewVoucher:                  "NewVoucher",
	NewVoucherResult:            "NewVoucherResult",
	PauseInitiator:              "PauseInitiator",
	ResumeInitiator:             "ResumeInitiator",
	PauseResponder:              "PauseResponder",
	ResumeResponder:             "ResumeResponder",
	FinishTransfer:              "FinishTransfer",
	ResponderBeginsFinalization: "ResponderBeginsFinalization",
	ResponderCompletes:          "ResponderCompletes",
	BeginFinalizing:             "BeginFinalizing",
	Complete:                    "Complete",
}

Events are human readable names for data transfer events

var Statuses = map[Status]string{

	Requested:                           "Requested",
	Ongoing:                             "Ongoing",
	TransferFinished:                    "TransferFinished",
	ResponderCompleted:                  "ResponderCompleted",
	Finalizing:                          "Finalizing",
	Completing:                          "Completing",
	Completed:                           "Completed",
	Failing:                             "Failing",
	Failed:                              "Failed",
	Cancelling:                          "Cancelling",
	Cancelled:                           "Cancelled",
	InitiatorPaused:                     "InitiatorPaused",
	ResponderPaused:                     "ResponderPaused",
	BothPaused:                          "BothPaused",
	ResponderFinalizing:                 "ResponderFinalizing",
	ResponderFinalizingTransferFinished: "ResponderFinalizingTransferFinished",
	ChannelNotFoundError:                "ChannelNotFoundError",
}

Statuses are human readable names for data transfer states

Types

type Channel

type Channel interface {
	// TransferID returns the transfer id for this channel
	TransferID() TransferID

	// BaseCID returns the CID that is at the root of this data transfer
	BaseCID() cid.Cid

	// Selector returns the IPLD selector for this data transfer (represented as
	// an IPLD node)
	Selector() ipld.Node

	// Voucher returns the voucher for this data transfer
	Voucher() Voucher

	// Sender returns the peer id for the node that is sending data
	Sender() peer.ID

	// Recipient returns the peer id for the node that is receiving data
	Recipient() peer.ID

	// TotalSize returns the total size for the data being transferred
	TotalSize() uint64

	// IsPull returns whether this is a pull request
	IsPull() bool

	// ChannelID returns the ChannelID for this request
	ChannelID() ChannelID

	// OtherParty returns the opposite party in the channel to the passed in party
	OtherParty(thisParty peer.ID) peer.ID
}

Channel represents all the parameters for a single data transfer

type ChannelID

type ChannelID struct {
	Initiator peer.ID
	Responder peer.ID
	ID        TransferID
}

ChannelID is a unique identifier for a channel, distinct by both the other party's peer ID + the transfer ID

func (*ChannelID) MarshalCBOR

func (t *ChannelID) MarshalCBOR(w io.Writer) error

func (ChannelID) OtherParty

func (c ChannelID) OtherParty(thisPeer peer.ID) peer.ID

OtherParty returns the peer on the other side of the request, depending on whether this peer is the initiator or responder

func (ChannelID) String

func (c ChannelID) String() string

func (*ChannelID) UnmarshalCBOR

func (t *ChannelID) UnmarshalCBOR(r io.Reader) error

type ChannelState

type ChannelState interface {
	Channel

	// Status is the current status of this channel
	Status() Status

	// Sent returns the number of bytes sent
	Sent() uint64

	// Received returns the number of bytes received
	Received() uint64

	// Message offers additional information about the current status
	Message() string

	// Vouchers returns all vouchers sent on this channel
	Vouchers() []Voucher

	// VoucherResults are results of vouchers sent on the channel
	VoucherResults() []VoucherResult

	// LastVoucher returns the last voucher sent on the channel
	LastVoucher() Voucher

	// LastVoucherResult returns the last voucher result sent on the channel
	LastVoucherResult() VoucherResult
}

ChannelState is channel parameters plus it's current state

type Event

type Event struct {
	Code      EventCode // What type of event it is
	Message   string    // Any clarifying information about the event
	Timestamp time.Time // when the event happened
}

Event is a struct containing information about a data transfer event

type EventCode

type EventCode int

EventCode is a name for an event that occurs on a data transfer channel

const (
	// Open is an event occurs when a channel is first opened
	Open EventCode = iota

	// Accept is an event that emits when the data transfer is first accepted
	Accept

	// Progress is an event that gets emitted every time more data is transferred
	Progress

	// Cancel indicates one side has cancelled the transfer
	Cancel

	// Error is an event that emits when an error occurs in a data transfer
	Error

	// CleanupComplete emits when a request is cleaned up
	CleanupComplete

	// NewVoucher means we have a new voucher on this channel
	NewVoucher

	// NewVoucherResult means we have a new voucher result on this channel
	NewVoucherResult

	// PauseInitiator emits when the data sender pauses transfer
	PauseInitiator

	// ResumeInitiator emits when the data sender resumes transfer
	ResumeInitiator

	// PauseResponder emits when the data receiver pauses transfer
	PauseResponder

	// ResumeResponder emits when the data receiver resumes transfer
	ResumeResponder

	// FinishTransfer emits when the initiator has completed sending/receiving data
	FinishTransfer

	// ResponderCompletes emits when the initiator receives a message that the responder is finished
	ResponderCompletes

	// ResponderBeginsFinalization emits when the initiator receives a message that the responder is finilizing
	ResponderBeginsFinalization

	// BeginFinalizing emits when the responder completes its operations but awaits a response from the
	// initiator
	BeginFinalizing

	// Complete is emitted when a data transfer is complete
	Complete
)

type EventsHandler

type EventsHandler interface {
	// OnChannelOpened is called when we ask the other peer to send us data on the
	// given channel ID
	// return values are:
	// - nil = this channel is recognized
	// - error = ignore incoming data for this channel
	OnChannelOpened(chid ChannelID) error
	// OnResponseReceived is called when we receive a response to a request
	// - nil = continue receiving data
	// - error = cancel this request
	OnResponseReceived(chid ChannelID, msg Response) error
	// OnDataReceive is called when we receive data for the given channel ID
	// return values are:
	// - nil = proceed with sending data
	// - error = cancel this request
	// - err == ErrPause - pause this request
	OnDataReceived(chid ChannelID, link ipld.Link, size uint64) error
	// OnDataSent is called when we send data for the given channel ID
	// return values are:
	// message = data transfer message along with data
	// err = error
	// - nil = proceed with sending data
	// - error = cancel this request
	// - err == ErrPause - pause this request
	OnDataSent(chid ChannelID, link ipld.Link, size uint64) (Message, error)
	// OnRequestReceived is called when we receive a new request to send data
	// for the given channel ID
	// return values are:
	// message = data transfer message along with reply
	// err = error
	// - nil = proceed with sending data
	// - error = cancel this request
	// - err == ErrPause - pause this request (only for new requests)
	// - err == ErrResume - resume this request (only for update requests)
	OnRequestReceived(chid ChannelID, msg Request) (Response, error)
	// OnResponseCompleted is called when we finish sending data for the given channel ID
	// Error returns are logged but otherwise have not effect
	OnChannelCompleted(chid ChannelID, success bool) error
}

EventsHandler are semantic data transfer events that happen as a result of graphsync hooks

type Manager

type Manager interface {

	// Start initializes data transfer processing
	Start(ctx context.Context) error

	// Stop terminates all data transfers and ends processing
	Stop(ctx context.Context) error

	// RegisterVoucherType registers a validator for the given voucher type
	// will error if voucher type does not implement voucher
	// or if there is a voucher type registered with an identical identifier
	RegisterVoucherType(voucherType Voucher, validator RequestValidator) error

	// RegisterRevalidator registers a revalidator for the given voucher type
	// Note: this is the voucher type used to revalidate. It can share a name
	// with the initial validator type and CAN be the same type, or a different type.
	// The revalidator can simply be the sampe as the original request validator,
	// or a different validator that satisfies the revalidator interface.
	RegisterRevalidator(voucherType Voucher, revalidator Revalidator) error

	// RegisterVoucherResultType allows deserialization of a voucher result,
	// so that a listener can read the metadata
	RegisterVoucherResultType(resultType VoucherResult) error

	// RegisterTransportConfigurer registers the given transport configurer to be run on requests with the given voucher
	// type
	RegisterTransportConfigurer(voucherType Voucher, configurer TransportConfigurer) error

	// open a data transfer that will send data to the recipient peer and
	// transfer parts of the piece that match the selector
	OpenPushDataChannel(ctx context.Context, to peer.ID, voucher Voucher, baseCid cid.Cid, selector ipld.Node) (ChannelID, error)

	// open a data transfer that will request data from the sending peer and
	// transfer parts of the piece that match the selector
	OpenPullDataChannel(ctx context.Context, to peer.ID, voucher Voucher, baseCid cid.Cid, selector ipld.Node) (ChannelID, error)

	// send an intermediate voucher as needed when the receiver sends a request for revalidation
	SendVoucher(ctx context.Context, chid ChannelID, voucher Voucher) error

	// close an open channel (effectively a cancel)
	CloseDataTransferChannel(ctx context.Context, chid ChannelID) error

	// pause a data transfer channel (only allowed if transport supports it)
	PauseDataTransferChannel(ctx context.Context, chid ChannelID) error

	// resume a data transfer channel (only allowed if transport supports it)
	ResumeDataTransferChannel(ctx context.Context, chid ChannelID) error

	// get status of a transfer
	TransferChannelStatus(ctx context.Context, x ChannelID) Status

	// get notified when certain types of events happen
	SubscribeToEvents(subscriber Subscriber) Unsubscribe

	// get all in progress transfers
	InProgressChannels(ctx context.Context) (map[ChannelID]ChannelState, error)
}

Manager is the core interface presented by all implementations of of the data transfer sub system

type Message

type Message interface {
	IsRequest() bool
	IsNew() bool
	IsUpdate() bool
	IsPaused() bool
	IsCancel() bool
	TransferID() TransferID
	cborgen.CBORMarshaler
	cborgen.CBORUnmarshaler
	ToNet(w io.Writer) error
}

Message is a message for the data transfer protocol (either request or response) that can serialize to a protobuf

type PauseableTransport

type PauseableTransport interface {
	Transport
	// PauseChannel paused the given channel ID
	PauseChannel(ctx context.Context,
		chid ChannelID,
	) error
	// ResumeChannel resumes the given channel
	ResumeChannel(ctx context.Context,
		msg Message,
		chid ChannelID,
	) error
}

PauseableTransport is a transport that can also pause and resume channels

type Registerable

type Registerable interface {
	encoding.Encodable
	// Type is a unique string identifier for this voucher type
	Type() TypeIdentifier
}

Registerable is a type of object in a registry. It must be encodable and must have a single method that uniquely identifies its type

type Request

type Request interface {
	Message
	IsPull() bool
	IsVoucher() bool
	VoucherType() TypeIdentifier
	Voucher(decoder encoding.Decoder) (encoding.Encodable, error)
	BaseCid() cid.Cid
	Selector() (ipld.Node, error)
}

Request is a response message for the data transfer protocol

type RequestValidator

type RequestValidator interface {
	// ValidatePush validates a push request received from the peer that will send data
	ValidatePush(
		sender peer.ID,
		voucher Voucher,
		baseCid cid.Cid,
		selector ipld.Node) (VoucherResult, error)
	// ValidatePull validates a pull request received from the peer that will receive data
	ValidatePull(
		receiver peer.ID,
		voucher Voucher,
		baseCid cid.Cid,
		selector ipld.Node) (VoucherResult, error)
}

RequestValidator is an interface implemented by the client of the data transfer module to validate requests

type Response

type Response interface {
	Message
	IsVoucherResult() bool
	IsComplete() bool
	Accepted() bool
	VoucherResultType() TypeIdentifier
	VoucherResult(decoder encoding.Decoder) (encoding.Encodable, error)
	EmptyVoucherResult() bool
}

Response is a response message for the data transfer protocol

type Revalidator

type Revalidator interface {
	// Revalidate revalidates a request with a new voucher
	Revalidate(channelID ChannelID, voucher Voucher) (VoucherResult, error)
	// OnPullDataSent is called on the responder side when more bytes are sent
	// for a given pull request. The first value indicates whether the request was
	// recognized by this revalidator and should be considered 'handled'. If true,
	// the remaining two values are interpreted. If 'false' the request is passed on
	// to the next revalidators.
	// It should return a VoucherResult + ErrPause to
	// request revalidation or nil to continue uninterrupted,
	// other errors will terminate the request.
	OnPullDataSent(chid ChannelID, additionalBytesSent uint64) (bool, VoucherResult, error)
	// OnPushDataReceived is called on the responder side when more bytes are received
	// for a given push request. The first value indicates whether the request was
	// recognized by this revalidator and should be considered 'handled'. If true,
	// the remaining two values are interpreted. If 'false' the request is passed on
	// to the next revalidators. It should return a VoucherResult + ErrPause to
	// request revalidation or nil to continue uninterrupted,
	// other errors will terminate the request
	OnPushDataReceived(chid ChannelID, additionalBytesReceived uint64) (bool, VoucherResult, error)
	// OnComplete is called to make a final request for revalidation -- often for the
	// purpose of settlement. The first value indicates whether the request was
	// recognized by this revalidator and should be considered 'handled'. If true,
	// the remaining two values are interpreted. If 'false' the request is passed on
	// to the next revalidators.
	// if VoucherResult is non nil, the request will enter a settlement phase awaiting
	// a final update
	OnComplete(chid ChannelID) (bool, VoucherResult, error)
}

Revalidator is a request validator revalidates in progress requests by requesting request additional vouchers, and resuming when it receives them

type Status

type Status uint64

Status is the status of transfer for a given channel

const (
	// Requested means a data transfer was requested by has not yet been approved
	Requested Status = iota

	// Ongoing means the data transfer is in progress
	Ongoing

	// TransferFinished indicates the initiator is done sending/receiving
	// data but is awaiting confirmation from the responder
	TransferFinished

	// ResponderCompleted indicates the initiator received a message from the
	// responder that it's completed
	ResponderCompleted

	// Finalizing means the responder is awaiting a final message from the initator to
	// consider the transfer done
	Finalizing

	// Completing just means we have some final cleanup for a completed request
	Completing

	// Completed means the data transfer is completed successfully
	Completed

	// Failing just means we have some final cleanup for a failed request
	Failing

	// Failed means the data transfer failed
	Failed

	// Cancelling just means we have some final cleanup for a cancelled request
	Cancelling

	// Cancelled means the data transfer ended prematurely
	Cancelled

	// InitiatorPaused means the data sender has paused the channel (only the sender can unpause this)
	InitiatorPaused

	// ResponderPaused means the data receiver has paused the channel (only the receiver can unpause this)
	ResponderPaused

	// BothPaused means both sender and receiver have paused the channel seperately (both must unpause)
	BothPaused

	// ResponderFinalizing is a unique state where the responder is awaiting a final voucher
	ResponderFinalizing

	// ResponderFinalizingTransferFinished is a unique state where the responder is awaiting a final voucher
	// and we have received all data
	ResponderFinalizingTransferFinished

	// ChannelNotFoundError means the searched for data transfer does not exist
	ChannelNotFoundError
)

type Subscriber

type Subscriber func(event Event, channelState ChannelState)

Subscriber is a callback that is called when events are emitted

type TransferID

type TransferID uint64

TransferID is an identifier for a data transfer, shared between request/responder and unique to the requester

type Transport

type Transport interface {
	// OpenChannel initiates an outgoing request for the other peer to send data
	// to us on this channel
	// Note: from a data transfer symantic standpoint, it doesn't matter if the
	// request is push or pull -- OpenChannel is called by the party that is
	// intending to receive data
	OpenChannel(ctx context.Context,
		dataSender peer.ID,
		channelID ChannelID,
		root ipld.Link,
		stor ipld.Node,
		msg Message) error

	// CloseChannel closes the given channel
	CloseChannel(ctx context.Context, chid ChannelID) error
	// SetEventHandler sets the handler for events on channels
	SetEventHandler(events EventsHandler) error
	// CleanupChannel is called on the otherside of a cancel - removes any associated
	// data for the channel
	CleanupChannel(chid ChannelID)
}

Transport defines the interface for a transport layer for data transfer. Where the data transfer manager will coordinate setting up push and pull requests, validation, etc, the transport layer is responsible for moving data back and forth, and may be medium specific. For example, some transports may have the ability to pause and resume requests, while others may not. Some may support individual data events, while others may only support message events. Some transport layers may opt to use the actual data transfer network protocols directly while others may be able to encode messages in their own data protocol.

Transport is the minimum interface that must be satisfied to serve as a datatransfer transport layer. Transports must be able to open (open is always called by the receiving peer) and close channels, and set at an event handler

type TransportConfigurer

type TransportConfigurer func(chid ChannelID, voucher Voucher, transport Transport)

TransportConfigurer provides a mechanism to provide transport specific configuration for a given voucher type

type TypeIdentifier

type TypeIdentifier string

TypeIdentifier is a unique string identifier for a type of encodable object in a registry

type Unsubscribe

type Unsubscribe func()

Unsubscribe is a function that gets called to unsubscribe from data transfer events

type Voucher

type Voucher Registerable

Voucher is used to validate a data transfer request against the underlying storage or retrieval deal that precipitated it. The only requirement is a voucher can read and write from bytes, and has a string identifier type

type VoucherResult

type VoucherResult Registerable

VoucherResult is used to provide option additional information about a voucher being rejected or accepted