p2p/protocol/identity: smarter tracking observed addrs

- time them out (already was doing that with addrbook)
- keep count to counter symmetric nats

peer/addr_manager.go

@@ -22,7 +22,7 @@ const (
 	RecentlyConnectedAddrTTL = time.Minute * 10
 
 	// OwnObservedAddrTTL is used for our own external addresses observed by peers.
-	OwnObservedAddrTTL = time.Minute * 20
+	OwnObservedAddrTTL = time.Minute * 10
 
 	// PermanentAddrTTL is the ttl for a "permanent address" (e.g. bootstrap nodes)
 	// if we haven't shipped you an update to ipfs in 356 days

protocol/identify/id.go

@@ -47,7 +47,7 @@ type IDService struct {
 
 	// our own observed addresses.
 	// TODO: instead of expiring, remove these when we disconnect
-	addrs peer.AddrManager
+	observedAddrs ObservedAddrSet
 }
 
 func NewIDService(h host.Host) *IDService {
@@ -61,7 +61,7 @@ func NewIDService(h host.Host) *IDService {
 
 // OwnObservedAddrs returns the addresses peers have reported we've dialed from
 func (ids *IDService) OwnObservedAddrs() []ma.Multiaddr {
-	return ids.addrs.Addrs(ids.Host.ID())
+	return ids.observedAddrs.Addrs()
 }
 
 func (ids *IDService) IdentifyConn(c inet.Conn) {
@@ -250,7 +250,7 @@ func (ids *IDService) consumeObservedAddress(observed []byte, c inet.Conn) {
 
 	// ok! we have the observed version of one of our ListenAddresses!
 	log.Debugf("added own observed listen addr: %s --> %s", c.LocalMultiaddr(), maddr)
-	ids.addrs.AddAddr(ids.Host.ID(), maddr, peer.OwnObservedAddrTTL)
+	ids.observedAddrs.Add(maddr)
 }
 
 func addrInAddrs(a ma.Multiaddr, as []ma.Multiaddr) bool {

protocol/identify/obsaddr.go

@@ -0,0 +1,96 @@
+package identify
+
+import (
+	"sync"
+	"time"
+
+	peer "github.com/jbenet/go-ipfs/p2p/peer"
+
+	ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
+)
+
+// ObservedAddr is an entry for an address reported by our peers.
+// We only use addresses that:
+// - have been observed more than once. (counter symmetric nats)
+// - have been observed recently (10min), because our position in the
+//   network, or network port mapppings, may have changed.
+type ObservedAddr struct {
+	Addr      ma.Multiaddr
+	LastSeen  time.Time
+	TimesSeen int
+}
+
+// ObservedAddrSet keeps track of a set of ObservedAddrs
+// the zero-value is ready to be used.
+type ObservedAddrSet struct {
+	sync.Mutex // guards whole datastruct.
+
+	addrs map[string]ObservedAddr
+	ttl   time.Duration
+}
+
+func (oas *ObservedAddrSet) Addrs() []ma.Multiaddr {
+	oas.Lock()
+	defer oas.Unlock()
+
+	// for zero-value.
+	if oas.addrs == nil {
+		return nil
+	}
+
+	now := time.Now()
+	addrs := make([]ma.Multiaddr, 0, len(oas.addrs))
+	for s, a := range oas.addrs {
+		// remove timed out addresses.
+		if now.Sub(a.LastSeen) > oas.ttl {
+			delete(oas.addrs, s)
+			continue
+		}
+
+		// we only use an address if we've seen it more than once
+		// because symmetric nats may cause all our peers to see
+		// different port numbers and thus report always different
+		// addresses (different ports) for us. These wouldn't be
+		// very useful. We make the assumption that if we've
+		// connected to two different peers, and they both have
+		// reported seeing the same address, it is probably useful.
+		if a.TimesSeen > 1 {
+			addrs = append(addrs, a.Addr)
+		}
+	}
+	return addrs
+}
+
+func (oas *ObservedAddrSet) Add(addr ma.Multiaddr) {
+	oas.Lock()
+	defer oas.Unlock()
+
+	// for zero-value.
+	if oas.addrs == nil {
+		oas.addrs = make(map[string]ObservedAddr)
+		oas.ttl = peer.OwnObservedAddrTTL
+	}
+
+	s := addr.String()
+	oas.addrs[s] = ObservedAddr{
+		Addr:      addr,
+		TimesSeen: oas.addrs[s].TimesSeen + 1,
+		LastSeen:  time.Now(),
+	}
+}
+
+func (oas *ObservedAddrSet) SetTTL(ttl time.Duration) {
+	oas.Lock()
+	defer oas.Unlock()
+	oas.ttl = ttl
+}
+
+func (oas *ObservedAddrSet) TTL() time.Duration {
+	oas.Lock()
+	defer oas.Unlock()
+	// for zero-value.
+	if oas.addrs == nil {
+		oas.ttl = peer.OwnObservedAddrTTL
+	}
+	return oas.ttl
+}

protocol/identify/obsaddr_test.go

@@ -0,0 +1,73 @@
+package identify
+
+import (
+	"testing"
+	"time"
+
+	ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
+)
+
+// TestObsAddrSet
+func TestObsAddrSet(t *testing.T) {
+	m := func(s string) ma.Multiaddr {
+		m, err := ma.NewMultiaddr(s)
+		if err != nil {
+			t.Error(err)
+		}
+		return m
+	}
+
+	addrsMarch := func(a, b []ma.Multiaddr) bool {
+		for _, aa := range a {
+			found := false
+			for _, bb := range b {
+				if aa.Equal(bb) {
+					found = true
+					break
+				}
+			}
+			if !found {
+				return false
+			}
+		}
+		return true
+	}
+
+	a1 := m("/ip4/1.2.3.4/tcp/1231")
+	a2 := m("/ip4/1.2.3.4/tcp/1232")
+	a3 := m("/ip4/1.2.3.4/tcp/1233")
+
+	oas := ObservedAddrSet{}
+
+	if !addrsMarch(oas.Addrs(), nil) {
+		t.Error("addrs should be empty")
+	}
+
+	oas.Add(a1)
+	oas.Add(a2)
+	oas.Add(a3)
+
+	// these are all different so we should not yet get them.
+	if !addrsMarch(oas.Addrs(), nil) {
+		t.Error("addrs should _still_ be empty (once)")
+	}
+
+	oas.Add(a1)
+	if !addrsMarch(oas.Addrs(), []ma.Multiaddr{a1}) {
+		t.Error("addrs should only have a1")
+	}
+
+	oas.Add(a2)
+	oas.Add(a1)
+	oas.Add(a1)
+	if !addrsMarch(oas.Addrs(), []ma.Multiaddr{a1, a2}) {
+		t.Error("addrs should only have a1, a2")
+	}
+
+	// change the timeout constant so we can time it out.
+	oas.SetTTL(time.Millisecond * 200)
+	<-time.After(time.Millisecond * 210)
+	if !addrsMarch(oas.Addrs(), []ma.Multiaddr{nil}) {
+		t.Error("addrs should have timed out")
+	}
+}