-
Notifications
You must be signed in to change notification settings - Fork 70
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: libp2p retrieval transports endpoint
- Loading branch information
Showing
9 changed files
with
262 additions
and
5 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,61 @@ | ||
| package modules | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
|
|
||
| "github.com/filecoin-project/boost/node/config" | ||
| "github.com/filecoin-project/boost/retrievalmarket/lp2pimpl" | ||
| "github.com/filecoin-project/boost/retrievalmarket/types" | ||
| "github.com/libp2p/go-libp2p-core/host" | ||
| "github.com/multiformats/go-multiaddr" | ||
| "go.uber.org/fx" | ||
| ) | ||
|
|
||
| func NewTransportsListener(cfg *config.Boost) func(h host.Host) (*lp2pimpl.TransportsListener, error) { | ||
| return func(h host.Host) (*lp2pimpl.TransportsListener, error) { | ||
| protos := []types.Protocol{} | ||
|
|
||
| // Get the libp2p address from the Host | ||
| if len(h.Addrs()) > 0 { | ||
| // TODO: should this be a list of addresses instead? | ||
| protos = append(protos, types.Protocol{ | ||
| Name: "libp2p", | ||
| Endpoint: h.Addrs()[0], | ||
| }) | ||
| } | ||
|
|
||
| // If there's an http retrieval address specified, add HTTP to the list | ||
| // of supported protocols | ||
| if cfg.Dealmaking.HTTPRetrievalMultiAddr != "" { | ||
| maddr, err := multiaddr.NewMultiaddr(cfg.Dealmaking.HTTPRetrievalMultiAddr) | ||
| if err != nil { | ||
| msg := "HTTPRetrievalURL must be in multi-address format. " | ||
| msg += "Could not parse '%s' as multiaddr: %w" | ||
| return nil, fmt.Errorf(msg, cfg.Dealmaking.HTTPRetrievalMultiAddr, err) | ||
| } | ||
|
|
||
| protos = append(protos, types.Protocol{ | ||
| Name: "http", | ||
| Endpoint: maddr, | ||
| }) | ||
| } | ||
|
|
||
| return lp2pimpl.NewTransportsListener(h, protos), nil | ||
| } | ||
| } | ||
|
|
||
| func HandleRetrievalTransports(lc fx.Lifecycle, l lp2pimpl.TransportsListener) { | ||
| lc.Append(fx.Hook{ | ||
| OnStart: func(ctx context.Context) error { | ||
| log.Debug("starting retrieval transports listener") | ||
| l.Start() | ||
| return nil | ||
| }, | ||
| OnStop: func(context.Context) error { | ||
| log.Debug("stopping retrieval transports listener") | ||
| l.Stop() | ||
| return nil | ||
| }, | ||
| }) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,120 @@ | ||
| package lp2pimpl | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "time" | ||
|
|
||
| "github.com/filecoin-project/boost/retrievalmarket/types" | ||
| "github.com/filecoin-project/go-fil-markets/shared" | ||
| logging "github.com/ipfs/go-log/v2" | ||
| "github.com/ipld/go-ipld-prime/codec/dagcbor" | ||
| "github.com/libp2p/go-libp2p-core/host" | ||
| "github.com/libp2p/go-libp2p-core/network" | ||
| "github.com/libp2p/go-libp2p-core/peer" | ||
| "github.com/libp2p/go-libp2p-core/protocol" | ||
| ) | ||
|
|
||
| var clog = logging.Logger("boost:lp2p:tspt:client") | ||
| var slog = logging.Logger("boost:lp2p:tspt") | ||
|
|
||
| // TransportsProtocolID is the protocol for querying which retrieval transports | ||
| // the Storage Provider supports (http, libp2p, etc) | ||
| const TransportsProtocolID = protocol.ID("/fil/retrieval/transports/1.0.0") | ||
|
|
||
| // TransportsListener listens for incoming queries over libp2p | ||
| type TransportsListener struct { | ||
| host host.Host | ||
| protocols []types.Protocol | ||
| } | ||
|
|
||
| const streamReadDeadline = 10 * time.Second | ||
| const streamWriteDeadline = 10 * time.Second | ||
|
|
||
| // QueryClientOption is an option for configuring the libp2p storage deal client | ||
| type QueryClientOption func(*TransportsClient) | ||
|
|
||
| // RetryParameters changes the default parameters around connection reopening | ||
| func RetryParameters(minDuration time.Duration, maxDuration time.Duration, attempts float64, backoffFactor float64) QueryClientOption { | ||
| return func(c *TransportsClient) { | ||
| c.retryStream.SetOptions(shared.RetryParameters(minDuration, maxDuration, attempts, backoffFactor)) | ||
| } | ||
| } | ||
|
|
||
| // TransportsClient sends retrieval queries over libp2p | ||
| type TransportsClient struct { | ||
| retryStream *shared.RetryStream | ||
| } | ||
|
|
||
| func NewTransportsClient(h host.Host, options ...QueryClientOption) *TransportsClient { | ||
| c := &TransportsClient{ | ||
| retryStream: shared.NewRetryStream(h), | ||
| } | ||
| for _, option := range options { | ||
| option(c) | ||
| } | ||
| return c | ||
| } | ||
|
|
||
| // SendQuery sends a retrieval query over a libp2p stream to the peer | ||
| func (c *TransportsClient) SendQuery(ctx context.Context, id peer.ID) (*types.QueryResponse, error) { | ||
| clog.Debugw("query", "provider-peer", id) | ||
|
|
||
| // Create a libp2p stream to the provider | ||
| s, err := c.retryStream.OpenStream(ctx, id, []protocol.ID{TransportsProtocolID}) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| defer s.Close() // nolint | ||
|
|
||
| // Set a deadline on reading from the stream so it doesn't hang | ||
| _ = s.SetReadDeadline(time.Now().Add(streamReadDeadline)) | ||
| defer s.SetReadDeadline(time.Time{}) // nolint | ||
|
|
||
| // Read the response from the stream | ||
| queryResponsei, err := types.BindnodeRegistry.TypeFromReader(s, (*types.QueryResponse)(nil), dagcbor.Decode) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("reading query response: %w", err) | ||
| } | ||
| queryResponse := queryResponsei.(*types.QueryResponse) | ||
|
|
||
| clog.Debugw("response", "provider-peer", id) | ||
|
|
||
| return queryResponse, nil | ||
| } | ||
|
|
||
| func NewTransportsListener(h host.Host, protos []types.Protocol) *TransportsListener { | ||
| return &TransportsListener{ | ||
| host: h, | ||
| protocols: protos, | ||
| } | ||
| } | ||
|
|
||
| func (p *TransportsListener) Start() { | ||
| p.host.SetStreamHandler(TransportsProtocolID, p.handleNewQueryStream) | ||
| } | ||
|
|
||
| func (p *TransportsListener) Stop() { | ||
| p.host.RemoveStreamHandler(TransportsProtocolID) | ||
| } | ||
|
|
||
| // Called when the client opens a libp2p stream | ||
| func (l *TransportsListener) handleNewQueryStream(s network.Stream) { | ||
| defer s.Close() | ||
|
|
||
| slog.Debugw("query", "peer", s.Conn().RemotePeer()) | ||
|
|
||
| response := types.QueryResponse{Protocols: l.protocols} | ||
|
|
||
| // Set a deadline on writing to the stream so it doesn't hang | ||
| _ = s.SetWriteDeadline(time.Now().Add(streamWriteDeadline)) | ||
| defer s.SetWriteDeadline(time.Time{}) // nolint | ||
|
|
||
| // Write the response to the client | ||
| err := types.BindnodeRegistry.TypeToWriter(response, s, dagcbor.Encode) | ||
| if err != nil { | ||
| slog.Infow("error writing query response", "peer", s.Conn().RemotePeer(), "err", err) | ||
| return | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,45 @@ | ||
| package types | ||
|
|
||
| import ( | ||
| _ "embed" | ||
| "fmt" | ||
|
|
||
| "github.com/filecoin-project/go-address" | ||
| "github.com/ipld/go-ipld-prime/node/bindnode" | ||
| bindnoderegistry "github.com/ipld/go-ipld-prime/node/bindnode/registry" | ||
| "github.com/multiformats/go-multiaddr" | ||
| ) | ||
|
|
||
| type Protocol struct { | ||
| // The name of the transport protocol eg "libp2p" or "http" | ||
| Name string | ||
| // The address of the endpoint in multiaddr format | ||
| Endpoint multiaddr.Multiaddr | ||
| } | ||
|
|
||
| type QueryResponse struct { | ||
| Protocols []Protocol | ||
| } | ||
|
|
||
| //go:embed transports.ipldsch | ||
| var embedSchema []byte | ||
|
|
||
| // MultiAddrBindnodeOption converts a filecoin Address type to and from a Bytes | ||
| // field in a schema | ||
| var MultiAddrBindnodeOption = bindnode.TypedBytesConverter(&address.Address{}, multiAddrFromBytes, multiAddrToBytes) | ||
|
|
||
| func multiAddrFromBytes(b []byte) (interface{}, error) { | ||
| return multiaddr.NewMultiaddrBytes(b) | ||
| } | ||
|
|
||
| func multiAddrToBytes(iface interface{}) ([]byte, error) { | ||
| var ma multiaddr.Multiaddr | ||
| ma, ok := iface.(multiaddr.Multiaddr) | ||
| if !ok { | ||
| return nil, fmt.Errorf("expected *Multiaddr value") | ||
| } | ||
|
|
||
| return ma.Bytes(), nil | ||
| } | ||
|
|
||
| var BindnodeRegistry = bindnoderegistry.NewRegistry() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
| # Defines the response to a query asking which transport protocols a | ||
| # Storage Provider supports | ||
|
|
||
| type Protocol struct { | ||
| # The name of the transport protocol eg "libp2p" or "http" | ||
| Name string | ||
| # The address of the endpoint in multiaddr format | ||
| Endpoint Multiaddr | ||
| } | ||
|
|
||
| type QueryResponse struct { | ||
| Protocols [Protocol] | ||
| } |