Refactoring Transport Interface

[hannahhoward]

Goals

The transport interface has several issues:

Use of Libp2p Protocol vs transport

Currently, the primary manager class for go-data-transfer communicates data in two different ways:

• The Transport interface
• The data transfer libp2p protocol

This was based off the original transport for go-data-transfer being GraphSync, and basically anything GraphSync can do, data transfer used Transport for while anything graphsync can't do was sent through the libp2p protocol.

This means the transport interface is very GraphSync specific and incomplete. What we would like is for data transfer's main manager to know nothing about libp2p and networks. It should be an abstract state manager that simply tracks how transfers progress and pauses them at appropriate times. Anything that goes over the wire should go through the transport interface.

That means since GraphSync can't send every message related to data transfer, it should use the go-data-transfers libp2p transport where needed -- but inside the implementation of the GraphSync transport.

Multiple Transports and the libp2p protocol

One potential issue arises if multiple transports intend to use the data transfer libp2p protocol: which transport handles an incoming message on the protocol?

There are two ways to address this:

1. Put data transport information in the message format
2. Each transport gets its own sub-protocol (i.e. "/data-transfer/1.3.0/bitswap")

My current thinking is to use the second. It provides a simply way to do transport negotiation (if you respond to the transport, you support the protocol) and I prefer not to put transport information in the message format (given part of data transfer's "special sauce" is making the message format agnostic)

Supporting different capabilities in transports

Should go-data-transfer require a transport to do every possible thing go-data-transfer likes to do? Or allow a transport to support only basic actions with other operations -- restarting, pausing/resuming, sending additional messages directly, setting data limits, etc -- being optional?

If that's the case, how do we negotiate a voucher format / exchange protocol and a transport? Like if I need to do paid filecoin retrieval, my transport had better support data limits, pause, resume. If I'm just doing free retrieval or storage, probably just restarts will suffice.

The proposed interface below takes a shot at this but I'm not sure it's complete

Removing error returns on event handlers

Error returns for event handlers do one of two things:

• For OnRequestReceived, they can close the channel if present or pause/resume it if you return datatransfer.ErrPause / ErrResume
• For OnResponseReceived, they can close the channel if present
• For everything else, the transport is supposed to "log the error"

I think this is a bit silly. RequestReceived and ResponseReceived should have specific actions you can take, while for all other returns, if an error occurs, data transfer itself should log it.

Cleaning up the transport sequence for events on a transport

One ongoing complaint about data transfer is not knowing where the remote is with accepting your request. The new transport lays out a clear chain of events:

Initiator:

• calls OpenChannel
• OnChannelOpened event when request actually sent to the responder (since GS now queues outgoing request)
• OnResponseReceived when responder accepts
• OnTransferInitiated when data starts sending/receiving

New States =
Requested -> Opened -> Accepted -> Ongoing

Responder

• OnRequestReceived when incoming request received
• OnTransferInitiated when data starts sending/receiving

New Staters =
Opened -> Accepted -> Ongoing

What about OnTransferQueued?

This event really exists cause of a weakness in Graphsync: validation is only performed at the point transfer starts sending/receiving. This is something Graphsync needs to fix. We should validate requests immediately on receipt, then queue them if they are accepted, rather than throwing them in the queue even though they might get rejected. I propose we fix this.

Is OnTransferInitiated neccesary?

Arguably, OnTransferInitiated can be determined from the Queued/Send/Received goes over zero. I'm not sure how I feel about that.

Data Limits To The Transport

This proposal suggests moving data limit handling into the transport. That doesn't imply the protocol needs to handle it directly: only that go-data-transfer itself doesn't handle knowing when a data limit is hit. Moreover, as a follow on to #308 I think we should make DataLimits getting hit a seperate set of events from Pause/Resume -- they really are different states -- one is I'm stuck till I get more data, the other is one side or the other decided to pause.

Unique and data progress to the transport

Similar to data limits, I think we should have the transport figure out when progress is actually made, rather than dispatch events about data sending/receiving that aren't actually new progress.

Proposed Interface

The proposed new transport interface is as follows:

// EventsHandler are semantic data transfer events that happen as a result of graphsync hooks
type EventsHandler interface {
	// OnChannelOpened is called at the point the transport begins processing the
	// request (prior to that it may simply be queued) -- only applies to initiator
	OnChannelOpened(chid ChannelID)

	// OnTransferInitiated is called at the point the transport actually begins sending/receiving data
	OnTransferInitiated(chid ChannelID)

	// OnResponseReceived is called when we receive a response to a request
	// Parameters:
	// - chid - channel id of the channel that received the response
	// - msg - response received
	// - actions - ChannelActions are actions that can be taken on the channel
	OnResponseReceived(chid ChannelID, msg Response, actions RequestActions)

	// OnRequestReceived is called when we receive a new request to send data
	// for the given channel ID
	// Parameters:
	// - chid - channel id of the channel that received the response
	// - msg - request received
	// - actions - ChannelActions are actions that can be taken on the channel
	OnRequestReceived(chid ChannelID, msg Request, actions ResponseActions)

	// OnTransferCompleted is called when we finish transferring data for the given channel ID
	OnTransferCompleted(chid ChannelID, err error)

	// OnRequestCancelled is called when a request we opened (with the given channel Id) to
	// receive data is cancelled by us.
	OnTransferCancelled(chid ChannelID, err error)

	// OnMessageSendError is called when a network error occurs while sending a message
	OnMessageSendError(chid ChannelID, err error)

	// OnSendDataError is called when a network error occurs sending data
	// at the transport layer
	OnSendDataError(chid ChannelID, err error)

	// OnReceiveDataError is called when a network error occurs receiving data
	// at the transport layer
	OnReceiveDataError(chid ChannelID, err error)

	// OnDataLimitReached is called when a channel hits a previously set data limit
	OnDataLimitReached(chid ChannelID, err error)

	// OnContextAugment allows the transport to attach data transfer tracing information
	// to its local context, in order to create a hierarchical trace
	OnContextAugment(chid ChannelID) func(context.Context) context.Context

	// OnDataQueued is called when data is queued for sending for the given channel ID
	OnDataQueued(chid ChannelID, link ipld.Link, size uint64, index int64)

	// OnDataReceived is called when we receive data for the given channel ID
	OnDataReceived(chid ChannelID, link ipld.Link, size uint64, index int64)

	// OnDataSent is called when we send data for the given channel ID
	OnDataSent(chid ChannelID, link ipld.Link, size uint64, index int64)
}

/*
Transport defines the interface for a transport layer for data
transfer. Where the data transfer manager will coordinate setting up push and
pull requests, persistence, 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 dispatch data update 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. Beyond that, additional actions you can take
with a transport is entirely based on the ChannelActions interface, which may have different
traits exposing different actions. What capabilities are present in ChannelActions is determined
by the Capabilities method  */
type Transport interface {
	Capabilities() TransportCapabilities
	// OpenChannel opens a channel on a given transport to move data back and forth
	OpenChannel(
		ctx context.Context,
		channel Channel,
		msg Message,
	) error
	WithChannel(ctx context.Context, chid ChannelID, actions func(ChannelActions) error) error
	// SetEventHandler sets the handler for events on channels
	SetEventsHandler(events EventsHandler) error
	// CleanupChannel removes any associated data on a closed channel
	CleanupChannel(chid ChannelID)
}

// TransportCapabilities describes additional capabilities supported by ChannelActions
type TransportCapabilities struct {
	// CanSendMessages indicates the channel can send messages in ResponseActions and ChannelActions
	CanSendMessages bool
	// CanRestart indicates ChannelActions will support RestartActions
	CanRestart bool
	// CanPauseResume indicates all actions interfaces support PauseActions
	CanPauseResume bool
	// CanLimitTransfer indications all actions will support LimitActions
	CanLimitTransfer bool
}

// ChannelActions are default actions that can be taken on a channel
type ChannelActions interface {
	CloseActions
}

// RequestActions are default actions that can be taken on a channel
type RequestActions interface {
	CloseActions
	SendMessageActions
}

// ResponseActions are default actions that can be taken on a channel
type ResponseActions interface {
	CloseActions
}

// CloseActions is a trait that allows closing a channel
type CloseActions interface {
	// CloseChannel close this channel and effectively closes it to further
	// action
	CloseChannel() error
}

// SendMessageActions is a trait that allows sending messages
// directly
type SendMessageActions interface {
	// SendMessage sends an arbitrary message over a transport
	SendMessage(message Message) error
}

// RestartActions is a trait that allows restarting a channel
type RestartActions interface {
	// RestartChannel restarts a channel
	RestartChannel(channel ChannelState)
}

// PauseActions a trait that allows pausing and resuming a channel
type PauseActions interface {
	// PauseChannel paused the given channel ID
	PauseChannel() error
	// ResumeChannel resumes the given channel
	ResumeChannel() error
}

// LimitActions is a trait that allows send limits on how much data can transfer on a channel
type LimitActions interface {
	// SetDataLimit tells a transport to pause a channel once the given amount of data is transferred
	SetDataLimit(uint64) error
}

Feel free to reach out to @aschmahmann for feedback as where we are headed

Note one additional change is I have removed the "unique" boolean on the progress events. A non-unique progress event should simply pass a block size of zero. (and let data transfer's logic determine no progress has occurred)

I also changed the name of OnChannelCompleted to OnTransferCompleted to distinguish that sometimes after the transfer itself is finished there is still more work to do.