@ -1,4 +1,4 @@
package relay
package auto relay
import (
"context"
@ -7,6 +7,8 @@ import (
"sync"
"time"
"golang.org/x/sync/errgroup"
"github.com/libp2p/go-libp2p-core/discovery"
"github.com/libp2p/go-libp2p-core/event"
"github.com/libp2p/go-libp2p-core/network"
@ -15,6 +17,8 @@ import (
basic "github.com/libp2p/go-libp2p/p2p/host/basic"
relayv1 "github.com/libp2p/go-libp2p/p2p/protocol/circuitv1/relay"
circuitv2 "github.com/libp2p/go-libp2p/p2p/protocol/circuitv2/client"
circuitv2_proto "github.com/libp2p/go-libp2p/p2p/protocol/circuitv2/proto"
ma "github.com/multiformats/go-multiaddr"
manet "github.com/multiformats/go-multiaddr/net"
@ -22,6 +26,14 @@ import (
const (
RelayRendezvous = "/libp2p/relay"
rsvpRefreshInterval = time . Minute
rsvpExpirationSlack = 2 * time . Minute
autorelayTag = "autorelay"
protoIDv1 = string ( relayv1 . ProtoID )
protoIDv2 = string ( circuitv2_proto . ProtoIDv2Hop )
)
var (
@ -30,7 +42,7 @@ var (
BootDelay = 20 * time . Second
)
// These are the known PL-operated relays
// These are the known PL-operated v1 relays; will be decommissioned in 2022.
var DefaultRelays = [ ] string {
"/ip4/147.75.80.110/tcp/4001/p2p/QmbFgm5zan8P6eWWmeyfncR5feYEMPbht5b1FW1C37aQ7y" ,
"/ip4/147.75.80.110/udp/4001/quic/p2p/QmbFgm5zan8P6eWWmeyfncR5feYEMPbht5b1FW1C37aQ7y" ,
@ -55,7 +67,7 @@ type AutoRelay struct {
disconnect chan struct { }
mx sync . Mutex
relays map [ peer . ID ] struct { }
relays map [ peer . ID ] * circuitv2 . Reservation // rsvp will be nil if it is a v1 relay
status network . Reachability
cachedAddrs [ ] ma . Multiaddr
@ -71,7 +83,7 @@ func NewAutoRelay(bhost *basic.BasicHost, discover discovery.Discoverer, router
router : router ,
addrsF : bhost . AddrsFactory ,
static : static ,
relays : make ( map [ peer . ID ] struct { } ) ,
relays : make ( map [ peer . ID ] * circuitv2 . Reservation ) ,
disconnect : make ( chan struct { } , 1 ) ,
status : network . ReachabilityUnknown ,
}
@ -92,6 +104,9 @@ func (ar *AutoRelay) background(ctx context.Context) {
subReachability , _ := ar . host . EventBus ( ) . Subscribe ( new ( event . EvtLocalReachabilityChanged ) )
defer subReachability . Close ( )
ticker := time . NewTicker ( rsvpRefreshInterval )
defer ticker . Stop ( )
// when true, we need to identify push
push := false
@ -119,8 +134,13 @@ func (ar *AutoRelay) background(ctx context.Context) {
}
ar . status = evt . Reachability
ar . mx . Unlock ( )
case <- ar . disconnect :
push = true
case now := <- ticker . C :
push = ar . refreshReservations ( ctx , now )
case <- ctx . Done ( ) :
return
}
@ -135,6 +155,67 @@ func (ar *AutoRelay) background(ctx context.Context) {
}
}
func ( ar * AutoRelay ) refreshReservations ( ctx context . Context , now time . Time ) bool {
ar . mx . Lock ( )
if ar . status == network . ReachabilityPublic {
// we are public, forget about the relays, unprotect peers
for p := range ar . relays {
ar . host . ConnManager ( ) . Unprotect ( p , autorelayTag )
delete ( ar . relays , p )
}
ar . mx . Unlock ( )
return true
}
if len ( ar . relays ) == 0 {
ar . mx . Unlock ( )
return false
}
// find reservations about to expire and refresh them in parallel
g := new ( errgroup . Group )
for p , rsvp := range ar . relays {
if rsvp == nil {
// this is a circuitv1 relay, there is no reservation
continue
}
if now . Add ( rsvpExpirationSlack ) . Before ( rsvp . Expiration ) {
continue
}
p := p
g . Go ( func ( ) error {
return ar . refreshRelayReservation ( ctx , p )
} )
}
ar . mx . Unlock ( )
err := g . Wait ( )
return err != nil
}
func ( ar * AutoRelay ) refreshRelayReservation ( ctx context . Context , p peer . ID ) error {
rsvp , err := circuitv2 . Reserve ( ctx , ar . host , peer . AddrInfo { ID : p } )
ar . mx . Lock ( )
defer ar . mx . Unlock ( )
if err != nil {
log . Debugf ( "failed to refresh relay slot reservation with %s: %s" , p , err )
delete ( ar . relays , p )
// unprotect the connection
ar . host . ConnManager ( ) . Unprotect ( p , autorelayTag )
} else {
log . Debugf ( "refreshed relay slot reservation with %s" , p )
ar . relays [ p ] = rsvp
}
return err
}
func ( ar * AutoRelay ) findRelays ( ctx context . Context ) bool {
if ar . numRelays ( ) >= DesiredRelays {
return false
@ -204,14 +285,46 @@ func (ar *AutoRelay) tryRelay(ctx context.Context, pi peer.AddrInfo) bool {
return false
}
ok , err := relayv1 . CanHop ( ctx , ar . host , pi . ID )
protos , err := ar . host . Peerstore ( ) . SupportsProtocols ( pi . ID , protoIDv1 , protoIDv2 )
if err != nil {
log . Debugf ( "error querying relay: %s" , err . Error ( ) )
log . Debugf ( "error checking relay protocol support for peer %s: %s" , pi . ID , err )
return false
}
if ! ok {
// not a hop relay
var supportsv1 , supportsv2 bool
for _ , proto := range protos {
switch proto {
case protoIDv1 :
supportsv1 = true
case protoIDv2 :
supportsv2 = true
}
}
var rsvp * circuitv2 . Reservation
switch {
case supportsv2 :
rsvp , err = circuitv2 . Reserve ( ctx , ar . host , pi )
if err != nil {
log . Debugf ( "error reserving slot with %s: %s" , pi . ID , err )
return false
}
case supportsv1 :
ok , err := relayv1 . CanHop ( ctx , ar . host , pi . ID )
if err != nil {
log . Debugf ( "error querying relay %s for v1 hop: %s" , pi . ID , err )
return false
}
if ! ok {
// not a hop relay
return false
}
default :
// supports neither, unusable relay.
return false
}
@ -222,7 +335,11 @@ func (ar *AutoRelay) tryRelay(ctx context.Context, pi peer.AddrInfo) bool {
if ar . host . Network ( ) . Connectedness ( pi . ID ) != network . Connected {
return false
}
ar . relays [ pi . ID ] = struct { } { }
ar . relays [ pi . ID ] = rsvp
// protect the connection
ar . host . ConnManager ( ) . Protect ( pi . ID , autorelayTag )
return true
}
@ -246,8 +363,29 @@ func (ar *AutoRelay) connect(ctx context.Context, pi peer.AddrInfo) bool {
return false
}
// tag the connection as very important
ar . host . ConnManager ( ) . TagPeer ( pi . ID , "relay" , 42 )
// wait for identify to complete in at least one conn so that we can check the supported protocols
conns := ar . host . Network ( ) . ConnsToPeer ( pi . ID )
if len ( conns ) == 0 {
return false
}
ready := make ( chan struct { } , len ( conns ) )
for _ , conn := range conns {
go func ( conn network . Conn ) {
select {
case <- ar . host . IDService ( ) . IdentifyWait ( conn ) :
ready <- struct { } { }
case <- ctx . Done ( ) :
}
} ( conn )
}
select {
case <- ready :
case <- ctx . Done ( ) :
return false
}
return true
}
vyzo
3 years ago
GitHub