feat: implement transport address

We need a way to specify the list of peer IDs to contact in the application
configuration, along with transport type + address to use.

In the past, we have the static peer lists in each transport, which map a peer
ID to an address. But we need to be able to dynamically dial peers not listed.

libp2p has the multiaddr which is a compact way to specify this:

 - /ip4/45.32.195.135/udp/4001/quic-v1/p2p/12D3KooWKX2W1HwKSxrRRWnHLtpATmevMWRoms3GxmgWYxEZMWav
 - Contains the IP address, the port, protocol (quic-v1), and peer ID.
 - https://github.com/multiformats/go-multiaddr/blob/master/protocols.go
 - Has a hardcoded list of multiaddr types (protocols).

Reasons to not use the multiaddr:

 - Our transports do not map 1-1 with the multiaddr transports
 - We don't have a central place with all protocols registered

Use something called a tptaddr instead:

 - {transport-type-id}|{addr}
 - ignore everything after the first |
 - strip the {transport-type-id}, lookup the transport w/ that type id, and pass
   addr to DialPeer.
 - add TransportTypeId to the transport interface
 - simplified transport type id:
   - udp-quic: dial using the udp quic controller (bifrost/transport/udp)
   - ws: websocket controller (supports both ws and wss)
     - ws|wss://my.url.com/

Add controller which maps peer IDs to transports:

 - Handle EstablishLinkWithPeer directives
 - Create a LookupTptAddr directive to lookup the list of tptaddr for a peer.
 - For each tptaddr with the peer create a DialTptAddr
   - Requests transports establish a link specifically using that tptaddr.
   - The transports can parse the tptaddr and check if it matches.
 - Expect that the results of DialTptAddr will also be pushed to
   EstablishLinkWithPeer automatically by the transport. See
   transport/controller/establish-link.go which will add a link waiter for any
   link that matches that peer id.
 - Once all directives become idle mark the resolver as idle.

End to end process:

 1. Someone creates EstablishLinkWithPeer
 2. The tptaddr controller creates LookupTptAddr with the peer id
 3. The tpt address storage controller returns the list of transport addresses
 4. The tptaddr controller creates DialTptAddr directives for each address
 5. The transports that resolve those directives also resolve EstablishLinkWithPeer.

Signed-off-by: Christian Stewart <[email protected]>

Author: paralin

agent/controller/config.go

@@ -5,7 +5,6 @@ import (
 	"github.com/aperturerobotics/bifrost/util/confparse"
 	"github.com/aperturerobotics/controllerbus/config"
 	"github.com/libp2p/go-libp2p/core/crypto"
-	"google.golang.org/protobuf/proto"
 )
 
 // ConfigID is the identifier for the config type.
@@ -23,7 +22,7 @@ func (c *Config) EqualsConfig(c2 config.Config) bool {
 		return false
 	}
 
-	return proto.Equal(c, oc)
+	return c.EqualVT(oc)
 }
 
 // Validate validates the configuration.

core/core_bus.go

@@ -12,6 +12,8 @@ import (
 	stream_echo "github.com/aperturerobotics/bifrost/stream/echo"
 	stream_forwarding "github.com/aperturerobotics/bifrost/stream/forwarding"
 	stream_listening "github.com/aperturerobotics/bifrost/stream/listening"
+	tptaddr_controller "github.com/aperturerobotics/bifrost/tptaddr/controller"
+	tptaddr_static "github.com/aperturerobotics/bifrost/tptaddr/static"
 	iproctpt "github.com/aperturerobotics/bifrost/transport/inproc"
 	udptpt "github.com/aperturerobotics/bifrost/transport/udp"
 	wtpt "github.com/aperturerobotics/bifrost/transport/websocket"
@@ -65,4 +67,8 @@ func AddFactories(b bus.Bus, sr *static.Resolver) {
 	// entity graph
 	sr.AddFactory(egc.NewFactory(b))
 	sr.AddFactory(bifrosteg.NewFactory(b))
+
+	// tptaddr
+	sr.AddFactory(tptaddr_controller.NewFactory(b))
+	sr.AddFactory(tptaddr_static.NewFactory(b))
 }

link/establish-link-ex.go

@@ -0,0 +1,33 @@
+package link
+
+import (
+	"context"
+
+	"github.com/aperturerobotics/bifrost/peer"
+	"github.com/aperturerobotics/controllerbus/bus"
+)
+
+// EstablishLinkWithPeerEx executes a EstablishLinkWithPeer directive.
+// Returns a release function.
+func EstablishLinkWithPeerEx(
+	ctx context.Context,
+	b bus.Bus,
+	localPeerID, remotePeerID peer.ID,
+	returnIfIdle bool,
+) (Link, func(), error) {
+	estl, _, ref, err := bus.ExecWaitValue[EstablishLinkWithPeerValue](
+		ctx,
+		b,
+		NewEstablishLinkWithPeer(
+			localPeerID, remotePeerID,
+		),
+		returnIfIdle,
+		nil,
+		nil,
+	)
+	if err != nil {
+		return nil, func() {}, err
+	}
+
+	return estl, ref.Release, nil
+}

link/establish-link.go

@@ -1,17 +1,19 @@
 package link
 
 import (
-	"errors"
 	"time"
 
 	"github.com/aperturerobotics/bifrost/peer"
 	"github.com/aperturerobotics/controllerbus/directive"
+	"github.com/pkg/errors"
 )
 
 // holdOpenDur is the default hold open duration
 var holdOpenDur = time.Second * 10
 
 // EstablishLinkWithPeer is a directive to establish a link with a peer.
+//
+// Value: Link
 type EstablishLinkWithPeer interface {
 	// Directive indicates EstablishLinkWithPeer is a directive.
 	directive.Directive
@@ -24,6 +26,9 @@ type EstablishLinkWithPeer interface {
 	EstablishLinkTargetPeerId() peer.ID
 }
 
+// EstablishLinkWithPeerValue is the type emitted when resolving EstablishLinkWithPeer.
+type EstablishLinkWithPeerValue = Link
+
 // establishLinkWithPeer implements EstablishLinkWithPeer with a peer ID constraint.
 type establishLinkWithPeer struct {
 	src, dest peer.ID
@@ -48,7 +53,7 @@ func (d *establishLinkWithPeer) EstablishLinkSourcePeerId() peer.ID {
 // This is a cursory validation to see if the values "look correct."
 func (d *establishLinkWithPeer) Validate() error {
 	if len(d.dest) == 0 {
-		return errors.New("peer id of destination required")
+		return errors.Wrap(peer.ErrEmptyPeerID, "destination")
 	}
 
 	return nil

link/handle_mounted_stream.go link/handle-mounted-stream.go

@@ -5,7 +5,6 @@ import (
 	"github.com/aperturerobotics/bifrost/protocol"
 	"github.com/aperturerobotics/bifrost/util/confparse"
 	"github.com/aperturerobotics/controllerbus/config"
-	"google.golang.org/protobuf/proto"
 )
 
 // ConfigID is the string used to identify this config object.
@@ -55,7 +54,7 @@ func (c *Config) EqualsConfig(other config.Config) bool {
 		return false
 	}
 
-	return proto.Equal(ot, c)
+	return ot.EqualVT(c)
 }
 
 var _ config.Config = ((*Config)(nil))

link/mounted_stream.go link/mounted-stream.go

@@ -60,7 +60,7 @@ func (c *Controller) GetControllerInfo() *controller.Info {
 	)
 }
 
-// Execute executes the forwarding controller.
+// Execute executes the echo controller.
 // Returning nil ends execution.
 // Returning an error triggers a retry with backoff.
 func (c *Controller) Execute(ctx context.Context) error {

link/open_stream_ex.go link/open-stream-ex.go

@@ -13,13 +13,13 @@ const ControllerID = "bifrost/stream/echo"
 // Version is the controller version.
 var Version = semver.MustParse("0.0.1")
 
-// Factory constructs a forwarding controller
+// Factory constructs a controller
 type Factory struct {
 	// bus is the controller bus
 	bus bus.Bus
 }
 
-// NewFactory builds a forwarding factory.
+// NewFactory builds a factory.
 func NewFactory(bus bus.Bus) *Factory {
 	return &Factory{bus: bus}
 }

link/open_stream_with_link_ex.go link/open-stream-with-link-ex.go

@@ -7,7 +7,6 @@ import (
 	"github.com/aperturerobotics/controllerbus/config"
 	ma "github.com/multiformats/go-multiaddr"
 	"github.com/pkg/errors"
-	"google.golang.org/protobuf/proto"
 )
 
 // ConfigID is the string used to identify this config object.
@@ -73,7 +72,7 @@ func (c *Config) EqualsConfig(other config.Config) bool {
 		return false
 	}
 
-	return proto.Equal(ot, c)
+	return ot.EqualVT(c)
 }
 
 var _ config.Config = ((*Config)(nil))

link/open_stream_with_link.go link/open-stream-with-link.go

@@ -26,6 +26,8 @@ type Testbed struct {
 	Bus bus.Bus
 	// PrivKey is the private key.
 	PrivKey crypto.PrivKey
+	// PeerID is the peer id for private key.
+	PeerID peer.ID
 	// Release releases the testbed.
 	Release func()
 }
@@ -73,6 +75,13 @@ func NewTestbed(ctx context.Context, le *logrus.Entry, opts TestbedOpts) (*Testb
 			return nil, err
 		}
 	}
+	if t.PrivKey != nil {
+		pid, err := peer.IDFromPrivateKey(t.PrivKey)
+		if err != nil {
+			return nil, err
+		}
+		t.PeerID = pid
+	}
 
 	if !opts.NoPeer {
 		// start peer controller

link/open_stream.go link/open-stream.go

@@ -0,0 +1,30 @@
+package tptaddr_controller
+
+import (
+	"github.com/aperturerobotics/controllerbus/config"
+)
+
+// ConfigID is the string used to identify this config object.
+const ConfigID = ControllerID
+
+// Validate validates the configuration.
+func (c *Config) Validate() error {
+	return nil
+}
+
+// GetConfigID returns the unique string for this configuration type.
+func (c *Config) GetConfigID() string {
+	return ConfigID
+}
+
+// EqualsConfig checks if the config is equal to another.
+func (c *Config) EqualsConfig(other config.Config) bool {
+	ot, ok := other.(*Config)
+	if !ok {
+		return false
+	}
+
+	return ot.EqualVT(c)
+}
+
+var _ config.Config = ((*Config)(nil))