move generic communication substrate into its own 'mesh' package

This is mostly a matter of moving some files, changing package names,
changing some imports, and changing some type references.

We duplicate the odd constant & global var, and the intmac/macint
functions, since neither 'router' or 'mesh' nor, really,' common' are
a natural home for them.

mesh/connection.go

@@ -0,0 +1,437 @@
+package mesh
+
+import (
+	"fmt"
+	"net"
+	"strconv"
+	"sync"
+	"time"
+)
+
+type Connection interface {
+	Local() *Peer
+	Remote() *Peer
+	RemoteTCPAddr() string
+	Outbound() bool
+	Established() bool
+	BreakTie(Connection) ConnectionTieBreak
+	Shutdown(error)
+	Log(args ...interface{})
+}
+
+type ConnectionTieBreak int
+
+const (
+	TieBreakWon ConnectionTieBreak = iota
+	TieBreakLost
+	TieBreakTied
+)
+
+var ErrConnectToSelf = fmt.Errorf("Cannot connect to ourself")
+
+type RemoteConnection struct {
+	local         *Peer
+	remote        *Peer
+	remoteTCPAddr string
+	outbound      bool
+	established   bool
+}
+
+type LocalConnection struct {
+	sync.RWMutex
+	RemoteConnection
+	TCPConn      *net.TCPConn
+	version      byte
+	tcpSender    TCPSender
+	SessionKey   *[32]byte
+	heartbeatTCP *time.Ticker
+	Router       *Router
+	uid          uint64
+	actionChan   chan<- ConnectionAction
+	finished     <-chan struct{} // closed to signal that actorLoop has finished
+	OverlayConn  OverlayConnection
+}
+
+type ConnectionAction func() error
+
+func NewRemoteConnection(from, to *Peer, tcpAddr string, outbound bool, established bool) *RemoteConnection {
+	return &RemoteConnection{
+		local:         from,
+		remote:        to,
+		remoteTCPAddr: tcpAddr,
+		outbound:      outbound,
+		established:   established,
+	}
+}
+
+func (conn *RemoteConnection) Local() *Peer                           { return conn.local }
+func (conn *RemoteConnection) Remote() *Peer                          { return conn.remote }
+func (conn *RemoteConnection) RemoteTCPAddr() string                  { return conn.remoteTCPAddr }
+func (conn *RemoteConnection) Outbound() bool                         { return conn.outbound }
+func (conn *RemoteConnection) Established() bool                      { return conn.established }
+func (conn *RemoteConnection) BreakTie(Connection) ConnectionTieBreak { return TieBreakTied }
+func (conn *RemoteConnection) Shutdown(error)                         {}
+
+func (conn *RemoteConnection) Log(args ...interface{}) {
+	log.Println(append(append([]interface{}{}, fmt.Sprintf("->[%s|%s]:", conn.remoteTCPAddr, conn.remote)), args...)...)
+}
+
+func (conn *RemoteConnection) ErrorLog(args ...interface{}) {
+	log.Errorln(append(append([]interface{}{}, fmt.Sprintf("->[%s|%s]:", conn.remoteTCPAddr, conn.remote)), args...)...)
+}
+
+// Does not return anything. If the connection is successful, it will
+// end up in the local peer's connections map.
+func StartLocalConnection(connRemote *RemoteConnection, tcpConn *net.TCPConn, router *Router, acceptNewPeer bool) {
+	if connRemote.local != router.Ourself.Peer {
+		log.Fatal("Attempt to create local connection from a peer which is not ourself")
+	}
+	// NB, we're taking a copy of connRemote here.
+	actionChan := make(chan ConnectionAction, ChannelSize)
+	finished := make(chan struct{})
+	conn := &LocalConnection{
+		RemoteConnection: *connRemote,
+		Router:           router,
+		TCPConn:          tcpConn,
+		uid:              randUint64(),
+		actionChan:       actionChan,
+		finished:         finished}
+	go conn.run(actionChan, finished, acceptNewPeer)
+}
+
+func (conn *LocalConnection) BreakTie(dupConn Connection) ConnectionTieBreak {
+	dupConnLocal := dupConn.(*LocalConnection)
+	// conn.uid is used as the tie breaker here, in the knowledge that
+	// both sides will make the same decision.
+	if conn.uid < dupConnLocal.uid {
+		return TieBreakWon
+	} else if dupConnLocal.uid < conn.uid {
+		return TieBreakLost
+	} else {
+		return TieBreakTied
+	}
+}
+
+func (conn *LocalConnection) Established() bool {
+	return conn.established
+}
+
+// Send directly, not via the Actor.  If it goes via the Actor we can
+// get a deadlock where LocalConnection is blocked talking to
+// LocalPeer and LocalPeer is blocked trying send a ProtocolMsg via
+// LocalConnection, and the channels are full in both directions so
+// nothing can proceed.
+func (conn *LocalConnection) SendProtocolMsg(m ProtocolMsg) {
+	if err := conn.sendProtocolMsg(m); err != nil {
+		conn.Shutdown(err)
+	}
+}
+
+// ACTOR methods
+
+// NB: The conn.* fields are only written by the connection actor
+// process, which is the caller of the ConnectionAction funs. Hence we
+// do not need locks for reading, and only need write locks for fields
+// read by other processes.
+
+// Async
+func (conn *LocalConnection) Shutdown(err error) {
+	// err should always be a real error, even if only io.EOF
+	if err == nil {
+		panic("nil error")
+	}
+
+	// Run on its own goroutine in case the channel is backed up
+	go func() { conn.sendAction(func() error { return err }) }()
+}
+
+// Send an actor request to the actorLoop, but don't block if
+// actorLoop has exited - see http://blog.golang.org/pipelines for
+// pattern
+func (conn *LocalConnection) sendAction(action ConnectionAction) {
+	select {
+	case conn.actionChan <- action:
+	case <-conn.finished:
+	}
+}
+
+// ACTOR server
+
+func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished chan<- struct{}, acceptNewPeer bool) {
+	var err error // important to use this var and not create another one with 'err :='
+	defer func() { conn.shutdown(err) }()
+	defer close(finished)
+
+	conn.TCPConn.SetLinger(0)
+	intro, err := ProtocolIntroParams{
+		MinVersion: conn.Router.ProtocolMinVersion,
+		MaxVersion: ProtocolMaxVersion,
+		Features:   conn.makeFeatures(),
+		Conn:       conn.TCPConn,
+		Password:   conn.Router.Password,
+		Outbound:   conn.outbound,
+	}.DoIntro()
+	if err != nil {
+		return
+	}
+
+	conn.SessionKey = intro.SessionKey
+	conn.tcpSender = intro.Sender
+	conn.version = intro.Version
+
+	remote, err := conn.parseFeatures(intro.Features)
+	if err != nil {
+		return
+	}
+
+	if err = conn.registerRemote(remote, acceptNewPeer); err != nil {
+		return
+	}
+
+	conn.Log("connection ready; using protocol version", conn.version)
+
+	params := OverlayConnectionParams{
+		RemotePeer:         conn.remote,
+		LocalAddr:          conn.TCPConn.LocalAddr().(*net.TCPAddr),
+		RemoteAddr:         conn.TCPConn.RemoteAddr().(*net.TCPAddr),
+		Outbound:           conn.outbound,
+		ConnUID:            conn.uid,
+		SessionKey:         conn.SessionKey,
+		SendControlMessage: conn.sendOverlayControlMessage,
+		Features:           intro.Features,
+	}
+	if conn.OverlayConn, err = conn.Router.Overlay.PrepareConnection(params); err != nil {
+		return
+	}
+
+	// As soon as we do AddConnection, the new connection becomes
+	// visible to the packet routing logic.  So AddConnection must
+	// come after PrepareConnection
+	if err = conn.Router.Ourself.AddConnection(conn); err != nil {
+		return
+	}
+	conn.Router.ConnectionMaker.ConnectionCreated(conn)
+
+	// OverlayConnection confirmation comes after AddConnection,
+	// because only after that completes do we know the connection is
+	// valid: in particular that it is not a duplicate connection to
+	// the same peer. Overlay communication on a duplicate connection
+	// can cause problems such as tripping up overlay crypto at the
+	// other end due to data being decoded by the other connection. It
+	// is also generally wasteful to engage in any interaction with
+	// the remote on a connection that turns out to be invalid.
+	conn.OverlayConn.Confirm()
+
+	// receiveTCP must follow also AddConnection. In the absence
+	// of any indirect connectivity to the remote peer, the first
+	// we hear about it (and any peers reachable from it) is
+	// through topology gossip it sends us on the connection. We
+	// must ensure that the connection has been added to Ourself
+	// prior to processing any such gossip, otherwise we risk
+	// immediately gc'ing part of that newly received portion of
+	// the topology (though not the remote peer itself, since that
+	// will have a positive ref count), leaving behind dangling
+	// references to peers. Hence we must invoke AddConnection,
+	// which is *synchronous*, first.
+	conn.heartbeatTCP = time.NewTicker(TCPHeartbeat)
+	go conn.receiveTCP(intro.Receiver)
+
+	// AddConnection must precede actorLoop. More precisely, it
+	// must precede shutdown, since that invokes DeleteConnection
+	// and is invoked on termination of this entire
+	// function. Essentially this boils down to a prohibition on
+	// running AddConnection in a separate goroutine, at least not
+	// without some synchronisation. Which in turn requires the
+	// launching of the receiveTCP goroutine to precede actorLoop.
+	err = conn.actorLoop(actionChan)
+}
+
+func (conn *LocalConnection) makeFeatures() map[string]string {
+	features := map[string]string{
+		"PeerNameFlavour": PeerNameFlavour,
+		"Name":            conn.local.Name.String(),
+		"NickName":        conn.local.NickName,
+		"ShortID":         fmt.Sprint(conn.local.ShortID),
+		"UID":             fmt.Sprint(conn.local.UID),
+		"ConnID":          fmt.Sprint(conn.uid),
+	}
+	conn.Router.Overlay.AddFeaturesTo(features)
+	return features
+}
+
+type features map[string]string
+
+func (features features) MustHave(keys []string) error {
+	for _, key := range keys {
+		if _, ok := features[key]; !ok {
+			return fmt.Errorf("Field %s is missing", key)
+		}
+	}
+	return nil
+}
+
+func (features features) Get(key string) string {
+	return features[key]
+}
+
+func (conn *LocalConnection) parseFeatures(features features) (*Peer, error) {
+	if err := features.MustHave([]string{"PeerNameFlavour", "Name", "NickName", "UID", "ConnID"}); err != nil {
+		return nil, err
+	}
+
+	remotePeerNameFlavour := features.Get("PeerNameFlavour")
+	if remotePeerNameFlavour != PeerNameFlavour {
+		return nil, fmt.Errorf("Peer name flavour mismatch (ours: '%s', theirs: '%s')", PeerNameFlavour, remotePeerNameFlavour)
+	}
+
+	name, err := PeerNameFromString(features.Get("Name"))
+	if err != nil {
+		return nil, err
+	}
+
+	nickName := features.Get("NickName")
+
+	var shortID uint64
+	var hasShortID bool
+	if shortIDStr, present := features["ShortID"]; present {
+		hasShortID = true
+		shortID, err = strconv.ParseUint(shortIDStr, 10,
+			PeerShortIDBits)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	uid, err := ParsePeerUID(features.Get("UID"))
+	if err != nil {
+		return nil, err
+	}
+
+	remoteConnID, err := strconv.ParseUint(features.Get("ConnID"), 10, 64)
+	if err != nil {
+		return nil, err
+	}
+
+	conn.uid ^= remoteConnID
+	peer := NewPeer(name, nickName, uid, 0, PeerShortID(shortID))
+	peer.HasShortID = hasShortID
+	return peer, nil
+}
+
+func (conn *LocalConnection) registerRemote(remote *Peer, acceptNewPeer bool) error {
+	if acceptNewPeer {
+		conn.remote = conn.Router.Peers.FetchWithDefault(remote)
+	} else {
+		conn.remote = conn.Router.Peers.FetchAndAddRef(remote.Name)
+		if conn.remote == nil {
+			return fmt.Errorf("Found unknown remote name: %s at %s", remote.Name, conn.remoteTCPAddr)
+		}
+	}
+
+	if conn.remote == conn.local {
+		return ErrConnectToSelf
+	}
+
+	return nil
+}
+
+func (conn *LocalConnection) actorLoop(actionChan <-chan ConnectionAction) (err error) {
+	fwdErrorChan := conn.OverlayConn.ErrorChannel()
+	fwdEstablishedChan := conn.OverlayConn.EstablishedChannel()
+
+	for err == nil {
+		select {
+		case action := <-actionChan:
+			err = action()
+
+		case <-conn.heartbeatTCP.C:
+			err = conn.sendSimpleProtocolMsg(ProtocolHeartbeat)
+
+		case <-fwdEstablishedChan:
+			conn.established = true
+			fwdEstablishedChan = nil
+			conn.Router.Ourself.ConnectionEstablished(conn)
+
+		case err = <-fwdErrorChan:
+		}
+	}
+	return
+}
+
+func (conn *LocalConnection) shutdown(err error) {
+	if conn.remote == nil {
+		log.Errorf("->[%s] connection shutting down due to error during handshake: %v", conn.remoteTCPAddr, err)
+	} else {
+		conn.ErrorLog("connection shutting down due to error:", err)
+	}
+
+	if conn.TCPConn != nil {
+		checkWarn(conn.TCPConn.Close())
+	}
+
+	if conn.remote != nil {
+		conn.Router.Peers.Dereference(conn.remote)
+		conn.Router.Ourself.DeleteConnection(conn)
+	}
+
+	if conn.heartbeatTCP != nil {
+		conn.heartbeatTCP.Stop()
+	}
+
+	if conn.OverlayConn != nil {
+		conn.OverlayConn.Stop()
+	}
+
+	conn.Router.ConnectionMaker.ConnectionTerminated(conn, err)
+}
+
+func (conn *LocalConnection) sendOverlayControlMessage(tag byte, msg []byte) error {
+	return conn.sendProtocolMsg(ProtocolMsg{ProtocolTag(tag), msg})
+}
+
+// Helpers
+
+func (conn *LocalConnection) sendSimpleProtocolMsg(tag ProtocolTag) error {
+	return conn.sendProtocolMsg(ProtocolMsg{tag: tag})
+}
+
+func (conn *LocalConnection) sendProtocolMsg(m ProtocolMsg) error {
+	return conn.tcpSender.Send(append([]byte{byte(m.tag)}, m.msg...))
+}
+
+func (conn *LocalConnection) receiveTCP(receiver TCPReceiver) {
+	var err error
+	for {
+		conn.extendReadDeadline()
+
+		var msg []byte
+		if msg, err = receiver.Receive(); err != nil {
+			break
+		}
+		if len(msg) < 1 {
+			conn.Log("ignoring blank msg")
+			continue
+		}
+		if err = conn.handleProtocolMsg(ProtocolTag(msg[0]), msg[1:]); err != nil {
+			break
+		}
+	}
+	conn.Shutdown(err)
+}
+
+func (conn *LocalConnection) handleProtocolMsg(tag ProtocolTag, payload []byte) error {
+	switch tag {
+	case ProtocolHeartbeat:
+	case ProtocolConnectionEstablished, ProtocolFragmentationReceived, ProtocolPMTUVerified, ProtocolOverlayControlMsg:
+		conn.OverlayConn.ControlMessage(byte(tag), payload)
+	case ProtocolGossipUnicast, ProtocolGossipBroadcast, ProtocolGossip:
+		return conn.Router.handleGossip(tag, payload)
+	default:
+		conn.Log("ignoring unknown protocol tag:", tag)
+	}
+	return nil
+}
+
+func (conn *LocalConnection) extendReadDeadline() {
+	conn.TCPConn.SetReadDeadline(time.Now().Add(TCPHeartbeat * 2))
+}

mesh/connection_maker.go

@@ -0,0 +1,312 @@
+package mesh
+
+import (
+	"fmt"
+	"math/rand"
+	"net"
+	"time"
+)
+
+const (
+	InitialInterval = 2 * time.Second
+	MaxInterval     = 6 * time.Minute
+	ResetAfter      = 1 * time.Minute
+)
+
+type peerAddrs map[string]*net.TCPAddr
+
+type ConnectionMaker struct {
+	ourself     *LocalPeer
+	peers       *Peers
+	port        int
+	discovery   bool
+	targets     map[string]*Target
+	connections map[Connection]struct{}
+	directPeers peerAddrs
+	actionChan  chan<- ConnectionMakerAction
+}
+
+type TargetState int
+
+const (
+	TargetWaiting    TargetState = iota // we are waiting to connect there
+	TargetAttempting                    // we are attempting to connect there
+	TargetConnected                     // we are connected to there
+)
+
+// Information about an address where we may find a peer
+type Target struct {
+	state       TargetState
+	lastError   error         // reason for disconnection last time
+	tryAfter    time.Time     // next time to try this address
+	tryInterval time.Duration // retry delay on next failure
+}
+
+type ConnectionMakerAction func() bool
+
+func NewConnectionMaker(ourself *LocalPeer, peers *Peers, port int, discovery bool) *ConnectionMaker {
+	actionChan := make(chan ConnectionMakerAction, ChannelSize)
+	cm := &ConnectionMaker{
+		ourself:     ourself,
+		peers:       peers,
+		port:        port,
+		discovery:   discovery,
+		directPeers: peerAddrs{},
+		targets:     make(map[string]*Target),
+		connections: make(map[Connection]struct{}),
+		actionChan:  actionChan}
+	go cm.queryLoop(actionChan)
+	return cm
+}
+
+func (cm *ConnectionMaker) InitiateConnections(peers []string, replace bool) []error {
+	errors := []error{}
+	addrs := peerAddrs{}
+	for _, peer := range peers {
+		host, port, err := net.SplitHostPort(peer)
+		if err != nil {
+			host = peer
+			port = "0" // we use that as an indication that "no port was supplied"
+		}
+		if addr, err := net.ResolveTCPAddr("tcp4", fmt.Sprintf("%s:%s", host, port)); err != nil {
+			errors = append(errors, err)
+		} else {
+			addrs[peer] = addr
+		}
+	}
+	cm.actionChan <- func() bool {
+		if replace {
+			cm.directPeers = peerAddrs{}
+		}
+		for peer, addr := range addrs {
+			cm.directPeers[peer] = addr
+			// curtail any existing reconnect interval
+			if target, found := cm.targets[addr.String()]; found {
+				target.nextTryNow()
+			}
+		}
+		return true
+	}
+	return errors
+}
+
+func (cm *ConnectionMaker) ForgetConnections(peers []string) {
+	cm.actionChan <- func() bool {
+		for _, peer := range peers {
+			delete(cm.directPeers, peer)
+		}
+		return false
+	}
+}
+
+func (cm *ConnectionMaker) ConnectionAborted(address string, err error) {
+	cm.actionChan <- func() bool {
+		target := cm.targets[address]
+		target.state = TargetWaiting
+		target.lastError = err
+		target.nextTryLater()
+		return true
+	}
+}
+
+func (cm *ConnectionMaker) ConnectionCreated(conn Connection) {
+	cm.actionChan <- func() bool {
+		cm.connections[conn] = void
+		if conn.Outbound() {
+			target := cm.targets[conn.RemoteTCPAddr()]
+			target.state = TargetConnected
+		}
+		return false
+	}
+}
+
+func (cm *ConnectionMaker) ConnectionTerminated(conn Connection, err error) {
+	cm.actionChan <- func() bool {
+		delete(cm.connections, conn)
+		if conn.Outbound() {
+			target := cm.targets[conn.RemoteTCPAddr()]
+			target.state = TargetWaiting
+			target.lastError = err
+			switch {
+			case err == ErrConnectToSelf:
+				target.nextTryNever()
+			case time.Now().After(target.tryAfter.Add(ResetAfter)):
+				target.nextTryNow()
+			default:
+				target.nextTryLater()
+			}
+		}
+		return true
+	}
+}
+
+func (cm *ConnectionMaker) Refresh() {
+	cm.actionChan <- func() bool { return true }
+}
+
+func (cm *ConnectionMaker) queryLoop(actionChan <-chan ConnectionMakerAction) {
+	timer := time.NewTimer(MaxDuration)
+	run := func() { timer.Reset(cm.checkStateAndAttemptConnections()) }
+	for {
+		select {
+		case action := <-actionChan:
+			if action() {
+				run()
+			}
+		case <-timer.C:
+			run()
+		}
+	}
+}
+
+func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration {
+	var (
+		validTarget  = make(map[string]struct{})
+		directTarget = make(map[string]struct{})
+	)
+	ourConnectedPeers, ourConnectedTargets, ourInboundIPs := cm.ourConnections()
+
+	addTarget := func(address string) {
+		if _, connected := ourConnectedTargets[address]; connected {
+			return
+		}
+		validTarget[address] = void
+		if _, found := cm.targets[address]; found {
+			return
+		}
+		target := &Target{state: TargetWaiting}
+		target.nextTryNow()
+		cm.targets[address] = target
+	}
+
+	// Add direct targets that are not connected
+	for _, addr := range cm.directPeers {
+		completeAddr := *addr
+		attempt := true
+		if completeAddr.Port == 0 {
+			completeAddr.Port = cm.port
+			// If a peer was specified w/o a port, then we do not
+			// attempt to connect to it if we have any inbound
+			// connections from that IP.
+			if _, connected := ourInboundIPs[completeAddr.IP.String()]; connected {
+				attempt = false
+			}
+		}
+		address := completeAddr.String()
+		directTarget[address] = void
+		if attempt {
+			addTarget(address)
+		}
+	}
+
+	// Add targets for peers that someone else is connected to, but we
+	// aren't
+	if cm.discovery {
+		cm.addPeerTargets(ourConnectedPeers, addTarget)
+	}
+
+	return cm.connectToTargets(validTarget, directTarget)
+}
+
+func (cm *ConnectionMaker) ourConnections() (PeerNameSet, map[string]struct{}, map[string]struct{}) {
+	var (
+		ourConnectedPeers   = make(PeerNameSet)
+		ourConnectedTargets = make(map[string]struct{})
+		ourInboundIPs       = make(map[string]struct{})
+	)
+	for conn := range cm.connections {
+		address := conn.RemoteTCPAddr()
+		ourConnectedPeers[conn.Remote().Name] = void
+		ourConnectedTargets[address] = void
+		if conn.Outbound() {
+			continue
+		}
+		if ip, _, err := net.SplitHostPort(address); err == nil { // should always succeed
+			ourInboundIPs[ip] = void
+		}
+	}
+	return ourConnectedPeers, ourConnectedTargets, ourInboundIPs
+}
+
+func (cm *ConnectionMaker) addPeerTargets(ourConnectedPeers PeerNameSet, addTarget func(string)) {
+	cm.peers.ForEach(func(peer *Peer) {
+		if peer == cm.ourself.Peer {
+			return
+		}
+		// Modifying peer.connections requires a write lock on Peers,
+		// and since we are holding a read lock (due to the ForEach),
+		// access without locking the peer is safe.
+		for otherPeer, conn := range peer.connections {
+			if otherPeer == cm.ourself.Name {
+				continue
+			}
+			if _, connected := ourConnectedPeers[otherPeer]; connected {
+				continue
+			}
+			address := conn.RemoteTCPAddr()
+			if conn.Outbound() {
+				addTarget(address)
+			} else if ip, _, err := net.SplitHostPort(address); err == nil {
+				// There is no point connecting to the (likely
+				// ephemeral) remote port of an inbound connection
+				// that some peer has. Let's try to connect on the
+				// weave port instead.
+				addTarget(fmt.Sprintf("%s:%d", ip, cm.port))
+			}
+		}
+	})
+}
+
+func (cm *ConnectionMaker) connectToTargets(validTarget map[string]struct{}, directTarget map[string]struct{}) time.Duration {
+	now := time.Now() // make sure we catch items just added
+	after := MaxDuration
+	for address, target := range cm.targets {
+		if target.state != TargetWaiting {
+			continue
+		}
+		if _, valid := validTarget[address]; !valid {
+			delete(cm.targets, address)
+			continue
+		}
+		if target.tryAfter.IsZero() {
+			continue
+		}
+		switch duration := target.tryAfter.Sub(now); {
+		case duration <= 0:
+			target.state = TargetAttempting
+			_, isCmdLineTarget := directTarget[address]
+			go cm.attemptConnection(address, isCmdLineTarget)
+		case duration < after:
+			after = duration
+		}
+	}
+	return after
+}
+
+func (cm *ConnectionMaker) attemptConnection(address string, acceptNewPeer bool) {
+	log.Printf("->[%s] attempting connection", address)
+	if err := cm.ourself.CreateConnection(address, acceptNewPeer); err != nil {
+		log.Errorf("->[%s] error during connection attempt: %v", address, err)
+		cm.ConnectionAborted(address, err)
+	}
+}
+
+func (t *Target) nextTryNever() {
+	t.tryAfter = time.Time{}
+	t.tryInterval = MaxInterval
+}
+
+func (t *Target) nextTryNow() {
+	t.tryAfter = time.Now()
+	t.tryInterval = InitialInterval
+}
+
+// The delay at the nth retry is a random value in the range
+// [i-i/2,i+i/2], where i = InitialInterval * 1.5^(n-1).
+func (t *Target) nextTryLater() {
+	t.tryAfter = time.Now().Add(t.tryInterval/2 + time.Duration(rand.Int63n(int64(t.tryInterval))))
+	t.tryInterval = t.tryInterval * 3 / 2
+	if t.tryInterval > MaxInterval {
+		t.tryInterval = MaxInterval
+	}
+}

mesh/gossip.go

@@ -0,0 +1,171 @@
+package mesh
+
+import (
+	"fmt"
+)
+
+type GossipData interface {
+	Encode() [][]byte
+	Merge(GossipData)
+}
+
+type Gossip interface {
+	// specific message from one peer to another
+	// intermediate peers relay it using unicast topology.
+	GossipUnicast(dstPeerName PeerName, msg []byte) error
+	// send gossip to every peer, relayed using broadcast topology.
+	GossipBroadcast(update GossipData) error
+}
+
+type Gossiper interface {
+	OnGossipUnicast(sender PeerName, msg []byte) error
+	// merge received data into state and return a representation of
+	// the received data, for further propagation
+	OnGossipBroadcast(sender PeerName, update []byte) (GossipData, error)
+	// return state of everything we know; gets called periodically
+	Gossip() GossipData
+	// merge received data into state and return "everything new I've
+	// just learnt", or nil if nothing in the received data was new
+	OnGossip(update []byte) (GossipData, error)
+}
+
+// Accumulates GossipData that needs to be sent to one destination,
+// and sends it when possible.
+type GossipSender struct {
+	send func(GossipData)
+	cell chan GossipData
+	// for testing
+	sent    bool
+	flushch chan chan bool
+}
+
+func NewGossipSender(send func(GossipData)) *GossipSender {
+	cell := make(chan GossipData, 1)
+	flushch := make(chan chan bool)
+	sender := &GossipSender{
+		send:    send,
+		cell:    cell,
+		flushch: flushch,
+	}
+	go sender.run()
+	return sender
+}
+
+func (sender *GossipSender) run() {
+	for {
+		select {
+		case pending := <-sender.cell:
+			if pending == nil { // receive zero value when chan is closed
+				return
+			}
+			sender.send(pending)
+			sender.sent = true
+		case ch := <-sender.flushch:
+			// send anything pending, then reply back whether we sent
+			// anything since previous flush
+			select {
+			case pending := <-sender.cell:
+				sender.send(pending)
+				sender.sent = true
+			default:
+			}
+			ch <- sender.sent
+			sender.sent = false
+		}
+	}
+}
+
+func (sender *GossipSender) Send(data GossipData) {
+	// NB: this must not be invoked concurrently
+	select {
+	case pending := <-sender.cell:
+		pending.Merge(data)
+		sender.cell <- pending
+	default:
+		sender.cell <- data
+	}
+}
+
+func (sender *GossipSender) Stop() {
+	close(sender.cell)
+}
+
+type GossipChannels map[string]*GossipChannel
+
+func (router *Router) NewGossip(channelName string, g Gossiper) Gossip {
+	channel := NewGossipChannel(channelName, router.Ourself, router.Routes, g)
+	router.gossipLock.Lock()
+	defer router.gossipLock.Unlock()
+	if _, found := router.gossipChannels[channelName]; found {
+		checkFatal(fmt.Errorf("[gossip] duplicate channel %s", channelName))
+	}
+	router.gossipChannels[channelName] = channel
+	return channel
+}
+
+func (router *Router) gossipChannel(channelName string) *GossipChannel {
+	router.gossipLock.RLock()
+	channel, found := router.gossipChannels[channelName]
+	router.gossipLock.RUnlock()
+	if found {
+		return channel
+	}
+	router.gossipLock.Lock()
+	defer router.gossipLock.Unlock()
+	if channel, found = router.gossipChannels[channelName]; found {
+		return channel
+	}
+	channel = NewGossipChannel(channelName, router.Ourself, router.Routes, &surrogateGossiper)
+	channel.log("created surrogate channel")
+	router.gossipChannels[channelName] = channel
+	return channel
+}
+
+func (router *Router) gossipChannelSet() map[*GossipChannel]struct{} {
+	channels := make(map[*GossipChannel]struct{})
+	router.gossipLock.RLock()
+	defer router.gossipLock.RUnlock()
+	for _, channel := range router.gossipChannels {
+		channels[channel] = void
+	}
+	return channels
+}
+
+func (router *Router) SendAllGossip() {
+	for channel := range router.gossipChannelSet() {
+		if gossip := channel.gossiper.Gossip(); gossip != nil {
+			channel.Send(router.Ourself.Name, gossip)
+		}
+	}
+}
+
+func (router *Router) SendAllGossipDown(conn Connection) {
+	for channel := range router.gossipChannelSet() {
+		if gossip := channel.gossiper.Gossip(); gossip != nil {
+			channel.SendDown(conn, channel.gossiper.Gossip())
+		}
+	}
+}
+
+// for testing
+
+func (router *Router) sendPendingGossip() bool {
+	sentSomething := false
+	for channel := range router.gossipChannelSet() {
+		channel.Lock()
+		for _, sender := range channel.senders {
+			sentSomething = sender.flush() || sentSomething
+		}
+		for _, sender := range channel.broadcasters {
+			sentSomething = sender.flush() || sentSomething
+		}
+		channel.Unlock()
+	}
+	return sentSomething
+}
+
+func (sender *GossipSender) flush() bool {
+	ch := make(chan bool)
+	sender.flushch <- ch
+	return <-ch
+}

mesh/gossip_channel.go

@@ -0,0 +1,250 @@
+package mesh
+
+import (
+	"bytes"
+	"encoding/gob"
+	"fmt"
+	"sync"
+)
+
+type GossipChannel struct {
+	sync.Mutex
+	name         string
+	ourself      *LocalPeer
+	routes       *Routes
+	gossiper     Gossiper
+	senders      connectionSenders
+	broadcasters peerSenders
+}
+
+type connectionSenders map[Connection]*GossipSender
+type peerSenders map[PeerName]*GossipSender
+
+func NewGossipChannel(channelName string, ourself *LocalPeer, routes *Routes, g Gossiper) *GossipChannel {
+	return &GossipChannel{
+		name:         channelName,
+		ourself:      ourself,
+		routes:       routes,
+		gossiper:     g,
+		senders:      make(connectionSenders),
+		broadcasters: make(peerSenders)}
+}
+
+func (router *Router) handleGossip(tag ProtocolTag, payload []byte) error {
+	decoder := gob.NewDecoder(bytes.NewReader(payload))
+	var channelName string
+	if err := decoder.Decode(&channelName); err != nil {
+		return err
+	}
+	channel := router.gossipChannel(channelName)
+	var srcName PeerName
+	if err := decoder.Decode(&srcName); err != nil {
+		return err
+	}
+	switch tag {
+	case ProtocolGossipUnicast:
+		return channel.deliverUnicast(srcName, payload, decoder)
+	case ProtocolGossipBroadcast:
+		return channel.deliverBroadcast(srcName, payload, decoder)
+	case ProtocolGossip:
+		return channel.deliver(srcName, payload, decoder)
+	}
+	return nil
+}
+
+func (c *GossipChannel) deliverUnicast(srcName PeerName, origPayload []byte, dec *gob.Decoder) error {
+	var destName PeerName
+	if err := dec.Decode(&destName); err != nil {
+		return err
+	}
+	if c.ourself.Name != destName {
+		if err := c.relayUnicast(destName, origPayload); err != nil {
+			// just log errors from relayUnicast; a problem between us and destination
+			// is not enough reason to break the connection from the source
+			c.log(err)
+		}
+		return nil
+	}
+	var payload []byte
+	if err := dec.Decode(&payload); err != nil {
+		return err
+	}
+	return c.gossiper.OnGossipUnicast(srcName, payload)
+}
+
+func (c *GossipChannel) deliverBroadcast(srcName PeerName, _ []byte, dec *gob.Decoder) error {
+	var payload []byte
+	if err := dec.Decode(&payload); err != nil {
+		return err
+	}
+	data, err := c.gossiper.OnGossipBroadcast(srcName, payload)
+	if err != nil || data == nil {
+		return err
+	}
+	return c.relayBroadcast(srcName, data)
+}
+
+func (c *GossipChannel) deliver(srcName PeerName, _ []byte, dec *gob.Decoder) error {
+	var payload []byte
+	if err := dec.Decode(&payload); err != nil {
+		return err
+	}
+	if data, err := c.gossiper.OnGossip(payload); err != nil {
+		return err
+	} else if data != nil {
+		c.Send(srcName, data)
+	}
+	return nil
+}
+
+func (c *GossipChannel) Send(srcName PeerName, data GossipData) {
+	// do this outside the lock below so we avoid lock nesting
+	c.routes.EnsureRecalculated()
+	selectedConnections := make(ConnectionSet)
+	for name := range c.routes.RandomNeighbours(srcName) {
+		if conn, found := c.ourself.ConnectionTo(name); found {
+			selectedConnections[conn] = void
+		}
+	}
+	if len(selectedConnections) == 0 {
+		return
+	}
+	connections := c.ourself.Connections()
+	c.Lock()
+	defer c.Unlock()
+	c.gcSenders(connections)
+	for conn := range selectedConnections {
+		c.sendDown(conn, data)
+	}
+}
+
+func (c *GossipChannel) SendDown(conn Connection, data GossipData) {
+	connections := c.ourself.Connections()
+	c.Lock()
+	defer c.Unlock()
+	c.gcSenders(connections)
+	c.sendDown(conn, data)
+}
+
+// GC - randomly (courtesy of go's map iterator) pick some existing
+// senders and stop&remove them if the associated connection is no
+// longer active.  We stop as soon as we encounter a valid entry; the
+// idea being that when there is little or no garbage then this
+// executes close to O(1)[1], whereas when there is lots of garbage we
+// remove it quickly.
+//
+// [1] TODO Unfortunately, due to the desire to avoid nested locks,
+// instead of simply invoking LocalPeer.ConnectionTo(name), we pass in
+// the result of LocalPeer.Connections(). That is O(n_our_connections)
+// at best.
+func (c *GossipChannel) gcSenders(connections ConnectionSet) {
+	for conn, sender := range c.senders {
+		if _, found := connections[conn]; !found {
+			delete(c.senders, conn)
+			sender.Stop()
+		} else {
+			break
+		}
+	}
+}
+
+// We have seen a couple of failures which suggest a >128GB slice was encountered.
+// 100MB should be enough for anyone.
+const maxFeasibleMessageLen = 100 * 1024 * 1024
+
+func (c *GossipChannel) sendDown(conn Connection, data GossipData) {
+	sender, found := c.senders[conn]
+	if !found {
+		sender = NewGossipSender(func(pending GossipData) {
+			for _, msg := range pending.Encode() {
+				if len(msg) > maxFeasibleMessageLen {
+					panic(fmt.Sprintf("Gossip message too large: len=%d bytes; on channel '%s' from %+v", len(msg), c.name, pending))
+				}
+				protocolMsg := ProtocolMsg{ProtocolGossip, GobEncode(c.name, c.ourself.Name, msg)}
+				conn.(ProtocolSender).SendProtocolMsg(protocolMsg)
+			}
+		})
+		c.senders[conn] = sender
+	}
+	sender.Send(data)
+}
+
+func (c *GossipChannel) GossipUnicast(dstPeerName PeerName, msg []byte) error {
+	return c.relayUnicast(dstPeerName, GobEncode(c.name, c.ourself.Name, dstPeerName, msg))
+}
+
+func (c *GossipChannel) GossipBroadcast(update GossipData) error {
+	return c.relayBroadcast(c.ourself.Name, update)
+}
+
+func (c *GossipChannel) relayUnicast(dstPeerName PeerName, buf []byte) (err error) {
+	if relayPeerName, found := c.routes.UnicastAll(dstPeerName); !found {
+		err = fmt.Errorf("unknown relay destination: %s", dstPeerName)
+	} else if conn, found := c.ourself.ConnectionTo(relayPeerName); !found {
+		err = fmt.Errorf("unable to find connection to relay peer %s", relayPeerName)
+	} else {
+		conn.(ProtocolSender).SendProtocolMsg(ProtocolMsg{ProtocolGossipUnicast, buf})
+	}
+	return err
+}
+
+func (c *GossipChannel) relayBroadcast(srcName PeerName, update GossipData) error {
+	names := c.routes.PeerNames() // do this outside the lock so they don't nest
+	c.Lock()
+	defer c.Unlock()
+	// GC - randomly (courtesy of go's map iterator) pick some
+	// existing broadcasters and stop&remove them if their source peer
+	// is unknown. We stop as soon as we encounter a valid entry; the
+	// idea being that when there is little or no garbage then this
+	// executes close to O(1)[1], whereas when there is lots of
+	// garbage we remove it quickly.
+	//
+	// [1] TODO Unfortunately, due to the desire to avoid nested
+	// locks, instead of simply invoking Peers.Fetch(name) below, we
+	// have that Peers.Names() invocation above. That is O(n_peers) at
+	// best.
+	for name, broadcaster := range c.broadcasters {
+		if _, found := names[name]; !found {
+			delete(c.broadcasters, name)
+			broadcaster.Stop()
+		} else {
+			break
+		}
+	}
+	broadcaster, found := c.broadcasters[srcName]
+	if !found {
+		broadcaster = NewGossipSender(func(pending GossipData) { c.sendBroadcast(srcName, pending) })
+		c.broadcasters[srcName] = broadcaster
+	}
+	broadcaster.Send(update)
+	return nil
+}
+
+func (c *GossipChannel) sendBroadcast(srcName PeerName, update GossipData) {
+	c.routes.EnsureRecalculated()
+	nextHops := c.routes.BroadcastAll(srcName)
+	if len(nextHops) == 0 {
+		return
+	}
+	connections := c.ourself.ConnectionsTo(nextHops)
+	for _, msg := range update.Encode() {
+		protocolMsg := ProtocolMsg{ProtocolGossipBroadcast, GobEncode(c.name, srcName, msg)}
+		// FIXME a single blocked connection can stall us
+		for _, conn := range connections {
+			conn.(ProtocolSender).SendProtocolMsg(protocolMsg)
+		}
+	}
+}
+
+func (c *GossipChannel) log(args ...interface{}) {
+	log.Println(append(append([]interface{}{}, "[gossip "+c.name+"]:"), args...)...)
+}
+
+func GobEncode(items ...interface{}) []byte {
+	buf := new(bytes.Buffer)
+	enc := gob.NewEncoder(buf)
+	for _, i := range items {
+		checkFatal(enc.Encode(i))
+	}
+	return buf.Bytes()
+}

mesh/gossip_test.go

@@ -0,0 +1,259 @@
+package mesh
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	wt "github.com/weaveworks/weave/testing"
+)
+
+// TODO test gossip unicast; atm we only test topology gossip and
+// surrogates, neither of which employ unicast.
+
+type mockChannelConnection struct {
+	RemoteConnection
+	dest *Router
+}
+
+func NewTestRouter(name string) *Router {
+	peerName, _ := PeerNameFromString(name)
+	router := NewRouter(Config{}, peerName, "nick", nil)
+	router.Start()
+	return router
+}
+
+func (conn *mockChannelConnection) SendProtocolMsg(protocolMsg ProtocolMsg) {
+	if err := conn.dest.handleGossip(protocolMsg.tag, protocolMsg.msg); err != nil {
+		panic(err)
+	}
+}
+
+func sendPendingGossip(routers ...*Router) {
+	// Loop until all routers report they didn't send anything
+	for sentSomething := true; sentSomething; {
+		sentSomething = false
+		for _, router := range routers {
+			sentSomething = router.sendPendingGossip() || sentSomething
+		}
+	}
+}
+
+func (router *Router) AddTestChannelConnection(r *Router) {
+	fromPeer := NewPeerFrom(router.Ourself.Peer)
+	toPeer := NewPeerFrom(r.Ourself.Peer)
+
+	r.Peers.FetchWithDefault(fromPeer)    // Has side-effect of incrementing refcount
+	router.Peers.FetchWithDefault(toPeer) //
+
+	conn := &mockChannelConnection{RemoteConnection{router.Ourself.Peer, toPeer, "", false, true}, r}
+	router.Ourself.handleAddConnection(conn)
+	router.Ourself.handleConnectionEstablished(conn)
+}
+
+func (router *Router) DeleteTestChannelConnection(r *Router) {
+	fromName := router.Ourself.Peer.Name
+	toName := r.Ourself.Peer.Name
+
+	fromPeer := r.Peers.Fetch(fromName)
+	toPeer := router.Peers.Fetch(toName)
+
+	r.Peers.Dereference(fromPeer)
+	router.Peers.Dereference(toPeer)
+
+	conn, _ := router.Ourself.ConnectionTo(toName)
+	router.Ourself.handleDeleteConnection(conn)
+}
+
+func TestGossipTopology(t *testing.T) {
+	wt.RunWithTimeout(t, 5*time.Second, func() {
+		implTestGossipTopology(t)
+	})
+}
+
+// Create a Peer representing the receiver router, with connections to
+// the routers supplied as arguments, carrying across all UID and
+// version information.
+func (router *Router) tp(routers ...*Router) *Peer {
+	peer := NewPeerFrom(router.Ourself.Peer)
+	connections := make(map[PeerName]Connection)
+	for _, r := range routers {
+		p := NewPeerFrom(r.Ourself.Peer)
+		connections[r.Ourself.Peer.Name] = newMockConnection(peer, p)
+	}
+	peer.Version = router.Ourself.Peer.Version
+	peer.connections = connections
+	return peer
+}
+
+// Check that the topology of router matches the peers and all of their connections
+func checkTopology(t *testing.T, router *Router, wantedPeers ...*Peer) {
+	router.Peers.RLock()
+	checkTopologyPeers(t, true, router.Peers.allPeers(), wantedPeers...)
+	router.Peers.RUnlock()
+}
+
+func implTestGossipTopology(t *testing.T) {
+	// Create some peers that will talk to each other
+	r1 := NewTestRouter("01:00:00:01:00:00")
+	r2 := NewTestRouter("02:00:00:02:00:00")
+	r3 := NewTestRouter("03:00:00:03:00:00")
+
+	// Check state when they have no connections
+	checkTopology(t, r1, r1.tp())
+	checkTopology(t, r2, r2.tp())
+
+	// Now try adding some connections
+	r1.AddTestChannelConnection(r2)
+	sendPendingGossip(r1, r2)
+	checkTopology(t, r1, r1.tp(r2), r2.tp())
+	checkTopology(t, r2, r1.tp(r2), r2.tp())
+	r2.AddTestChannelConnection(r1)
+	sendPendingGossip(r1, r2)
+	checkTopology(t, r1, r1.tp(r2), r2.tp(r1))
+	checkTopology(t, r2, r1.tp(r2), r2.tp(r1))
+
+	// Currently, the connection from 2 to 3 is one-way only
+	r2.AddTestChannelConnection(r3)
+	sendPendingGossip(r1, r2, r3)
+	checkTopology(t, r1, r1.tp(r2), r2.tp(r1, r3), r3.tp())
+	checkTopology(t, r2, r1.tp(r2), r2.tp(r1, r3), r3.tp())
+	// When r2 gossiped to r3, 1 was unreachable from r3 so it got removed from the
+	// list of peers, but remains referenced in the connection from 1 to 3.
+	checkTopology(t, r3, r2.tp(r1, r3), r3.tp())
+
+	// Add a connection from 3 to 1 and now r1 is reachable.
+	r3.AddTestChannelConnection(r1)
+	sendPendingGossip(r1, r2, r3)
+	checkTopology(t, r1, r1.tp(r2), r2.tp(r1, r3), r3.tp(r1))
+	checkTopology(t, r2, r1.tp(r2), r2.tp(r1, r3), r3.tp(r1))
+	checkTopology(t, r3, r1.tp(), r2.tp(r1, r3), r3.tp(r1))
+
+	r1.AddTestChannelConnection(r3)
+	sendPendingGossip(r1, r2, r3)
+	checkTopology(t, r1, r1.tp(r2, r3), r2.tp(r1, r3), r3.tp(r1))
+	checkTopology(t, r2, r1.tp(r2, r3), r2.tp(r1, r3), r3.tp(r1))
+	checkTopology(t, r3, r1.tp(r2, r3), r2.tp(r1, r3), r3.tp(r1))
+
+	// Drop the connection from 2 to 3
+	r2.DeleteTestChannelConnection(r3)
+	sendPendingGossip(r1, r2, r3)
+	checkTopology(t, r1, r1.tp(r2, r3), r2.tp(r1), r3.tp(r1))
+	checkTopology(t, r2, r1.tp(r2, r3), r2.tp(r1))
+	checkTopology(t, r3, r1.tp(r2, r3), r2.tp(r1), r3.tp(r1))
+
+	// Drop the connection from 1 to 3
+	r1.DeleteTestChannelConnection(r3)
+	sendPendingGossip(r1, r2, r3)
+	checkTopology(t, r1, r1.tp(r2), r2.tp(r1), r3.tp(r1))
+
+	checkTopology(t, r1, r1.tp(r2), r2.tp(r1), r3.tp(r1))
+	checkTopology(t, r2, r1.tp(r2), r2.tp(r1))
+	// r3 still thinks r1 has a connection to it
+	checkTopology(t, r3, r1.tp(r2, r3), r2.tp(r1), r3.tp(r1))
+
+	// On a timer, r3 will gossip to r1
+	r3.SendAllGossip()
+	sendPendingGossip(r1, r2, r3)
+	checkTopology(t, r1, r1.tp(r2), r2.tp(r1), r3.tp(r1))
+}
+
+func TestGossipSurrogate(t *testing.T) {
+	// create the topology r1 <-> r2 <-> r3
+	r1 := NewTestRouter("01:00:00:01:00:00")
+	r2 := NewTestRouter("02:00:00:02:00:00")
+	r3 := NewTestRouter("03:00:00:03:00:00")
+	r1.AddTestChannelConnection(r2)
+	r2.AddTestChannelConnection(r1)
+	r3.AddTestChannelConnection(r2)
+	r2.AddTestChannelConnection(r3)
+	sendPendingGossip(r1, r2, r3)
+	checkTopology(t, r1, r1.tp(r2), r2.tp(r1, r3), r3.tp(r2))
+	checkTopology(t, r2, r1.tp(r2), r2.tp(r1, r3), r3.tp(r2))
+	checkTopology(t, r3, r1.tp(r2), r2.tp(r1, r3), r3.tp(r2))
+
+	// create a gossiper at either end, but not the middle
+	g1 := newTestGossiper()
+	g3 := newTestGossiper()
+	s1 := r1.NewGossip("Test", g1)
+	s3 := r3.NewGossip("Test", g3)
+
+	// broadcast a message from each end, check it reaches the other
+	broadcast(s1, 1)
+	broadcast(s3, 2)
+	sendPendingGossip(r1, r2, r3)
+	g1.checkHas(t, 2)
+	g3.checkHas(t, 1)
+
+	// check that each end gets their message back through periodic
+	// gossip
+	r1.SendAllGossip()
+	r3.SendAllGossip()
+	sendPendingGossip(r1, r2, r3)
+	g1.checkHas(t, 1, 2)
+	g3.checkHas(t, 1, 2)
+}
+
+type testGossiper struct {
+	sync.RWMutex
+	state map[byte]struct{}
+}
+
+func newTestGossiper() *testGossiper {
+	return &testGossiper{state: make(map[byte]struct{})}
+}
+
+func (g *testGossiper) OnGossipUnicast(sender PeerName, msg []byte) error {
+	return nil
+}
+
+func (g *testGossiper) OnGossipBroadcast(_ PeerName, update []byte) (GossipData, error) {
+	g.Lock()
+	defer g.Unlock()
+	for _, v := range update {
+		g.state[v] = void
+	}
+	return NewSurrogateGossipData(update), nil
+}
+
+func (g *testGossiper) Gossip() GossipData {
+	g.RLock()
+	defer g.RUnlock()
+	state := make([]byte, len(g.state))
+	for v := range g.state {
+		state = append(state, v)
+	}
+	return NewSurrogateGossipData(state)
+}
+
+func (g *testGossiper) OnGossip(update []byte) (GossipData, error) {
+	g.Lock()
+	defer g.Unlock()
+	var delta []byte
+	for _, v := range update {
+		if _, found := g.state[v]; !found {
+			delta = append(delta, v)
+			g.state[v] = void
+		}
+	}
+	if len(delta) > 0 {
+		return NewSurrogateGossipData(delta), nil
+	}
+	return nil, nil
+}
+
+func (g *testGossiper) checkHas(t *testing.T, vs ...byte) {
+	g.RLock()
+	defer g.RUnlock()
+	for _, v := range vs {
+		if _, found := g.state[v]; !found {
+			require.FailNow(t, fmt.Sprintf("%d is missing", v))
+		}
+	}
+}
+
+func broadcast(s Gossip, v byte) {
+	s.GossipBroadcast(NewSurrogateGossipData([]byte{v}))
+}

mesh/local_peer.go

@@ -0,0 +1,261 @@
+package mesh
+
+import (
+	"fmt"
+	"net"
+	"sync"
+	"time"
+)
+
+type LocalPeer struct {
+	sync.RWMutex
+	*Peer
+	router     *Router
+	actionChan chan<- LocalPeerAction
+}
+
+type LocalPeerAction func()
+
+func NewLocalPeer(name PeerName, nickName string, router *Router) *LocalPeer {
+	actionChan := make(chan LocalPeerAction, ChannelSize)
+	peer := &LocalPeer{
+		Peer:       NewPeer(name, nickName, randomPeerUID(), 0, randomPeerShortID()),
+		router:     router,
+		actionChan: actionChan,
+	}
+	go peer.actorLoop(actionChan)
+	return peer
+}
+
+func (peer *LocalPeer) Connections() ConnectionSet {
+	connections := make(ConnectionSet)
+	peer.RLock()
+	defer peer.RUnlock()
+	for _, conn := range peer.connections {
+		connections[conn] = void
+	}
+	return connections
+}
+
+func (peer *LocalPeer) ConnectionTo(name PeerName) (Connection, bool) {
+	peer.RLock()
+	defer peer.RUnlock()
+	conn, found := peer.connections[name]
+	return conn, found // yes, you really can't inline that. FFS.
+}
+
+func (peer *LocalPeer) ConnectionsTo(names []PeerName) []Connection {
+	conns := make([]Connection, 0, len(names))
+	peer.RLock()
+	defer peer.RUnlock()
+	for _, name := range names {
+		conn, found := peer.connections[name]
+		// Again, !found could just be due to a race.
+		if found {
+			conns = append(conns, conn)
+		}
+	}
+	return conns
+}
+
+func (peer *LocalPeer) CreateConnection(peerAddr string, acceptNewPeer bool) error {
+	if err := peer.checkConnectionLimit(); err != nil {
+		return err
+	}
+	tcpAddr, err := net.ResolveTCPAddr("tcp4", peerAddr)
+	if err != nil {
+		return err
+	}
+	tcpConn, err := net.DialTCP("tcp4", nil, tcpAddr)
+	if err != nil {
+		return err
+	}
+	connRemote := NewRemoteConnection(peer.Peer, nil, tcpConn.RemoteAddr().String(), true, false)
+	StartLocalConnection(connRemote, tcpConn, peer.router, acceptNewPeer)
+	return nil
+}
+
+// ACTOR client API
+
+// Sync.
+func (peer *LocalPeer) AddConnection(conn *LocalConnection) error {
+	resultChan := make(chan error)
+	peer.actionChan <- func() {
+		resultChan <- peer.handleAddConnection(conn)
+	}
+	return <-resultChan
+}
+
+// Async.
+func (peer *LocalPeer) ConnectionEstablished(conn *LocalConnection) {
+	peer.actionChan <- func() {
+		peer.handleConnectionEstablished(conn)
+	}
+}
+
+// Sync.
+func (peer *LocalPeer) DeleteConnection(conn *LocalConnection) {
+	resultChan := make(chan interface{})
+	peer.actionChan <- func() {
+		peer.handleDeleteConnection(conn)
+		resultChan <- nil
+	}
+	<-resultChan
+}
+
+// ACTOR server
+
+func (peer *LocalPeer) actorLoop(actionChan <-chan LocalPeerAction) {
+	gossipTimer := time.Tick(GossipInterval)
+	for {
+		select {
+		case action := <-actionChan:
+			action()
+		case <-gossipTimer:
+			peer.router.SendAllGossip()
+		}
+	}
+}
+
+func (peer *LocalPeer) handleAddConnection(conn Connection) error {
+	if peer.Peer != conn.Local() {
+		log.Fatal("Attempt made to add connection to peer where peer is not the source of connection")
+	}
+	if conn.Remote() == nil {
+		log.Fatal("Attempt made to add connection to peer with unknown remote peer")
+	}
+	toName := conn.Remote().Name
+	dupErr := fmt.Errorf("Multiple connections to %s added to %s", conn.Remote(), peer.String())
+	// deliberately non symmetrical
+	if dupConn, found := peer.connections[toName]; found {
+		if dupConn == conn {
+			return nil
+		}
+		switch conn.BreakTie(dupConn) {
+		case TieBreakWon:
+			dupConn.Shutdown(dupErr)
+			peer.handleDeleteConnection(dupConn)
+		case TieBreakLost:
+			return dupErr
+		case TieBreakTied:
+			// oh good grief. Sod it, just kill both of them.
+			dupConn.Shutdown(dupErr)
+			peer.handleDeleteConnection(dupConn)
+			return dupErr
+		}
+	}
+	if err := peer.checkConnectionLimit(); err != nil {
+		return err
+	}
+	_, isConnectedPeer := peer.router.Routes.Unicast(toName)
+	peer.addConnection(conn)
+	if isConnectedPeer {
+		conn.Log("connection added")
+	} else {
+		conn.Log("connection added (new peer)")
+		peer.router.SendAllGossipDown(conn)
+	}
+
+	peer.router.Routes.Recalculate()
+	peer.broadcastPeerUpdate(conn.Remote())
+
+	return nil
+}
+
+func (peer *LocalPeer) handleConnectionEstablished(conn Connection) {
+	if peer.Peer != conn.Local() {
+		log.Fatal("Peer informed of active connection where peer is not the source of connection")
+	}
+	if dupConn, found := peer.connections[conn.Remote().Name]; !found || conn != dupConn {
+		conn.Shutdown(fmt.Errorf("Cannot set unknown connection active"))
+		return
+	}
+	peer.connectionEstablished(conn)
+	conn.Log("connection fully established")
+
+	peer.router.Routes.Recalculate()
+	peer.broadcastPeerUpdate()
+}
+
+func (peer *LocalPeer) handleDeleteConnection(conn Connection) {
+	if peer.Peer != conn.Local() {
+		log.Fatal("Attempt made to delete connection from peer where peer is not the source of connection")
+	}
+	if conn.Remote() == nil {
+		log.Fatal("Attempt made to delete connection to peer with unknown remote peer")
+	}
+	toName := conn.Remote().Name
+	if connFound, found := peer.connections[toName]; !found || connFound != conn {
+		return
+	}
+	peer.deleteConnection(conn)
+	conn.Log("connection deleted")
+	// Must do garbage collection first to ensure we don't send out an
+	// update with unreachable peers (can cause looping)
+	peer.router.Peers.GarbageCollect()
+	peer.router.Routes.Recalculate()
+	peer.broadcastPeerUpdate()
+}
+
+// helpers
+
+func (peer *LocalPeer) broadcastPeerUpdate(peers ...*Peer) {
+	// Some tests run without a router.  This should be fixed so
+	// that the relevant part of Router can be easily run in the
+	// context of a test, but that will involve significant
+	// reworking of tests.
+	if peer.router != nil {
+		peer.router.BroadcastTopologyUpdate(append(peers, peer.Peer))
+	}
+}
+
+func (peer *LocalPeer) checkConnectionLimit() error {
+	limit := peer.router.ConnLimit
+	if 0 != limit && peer.connectionCount() >= limit {
+		return fmt.Errorf("Connection limit reached (%v)", limit)
+	}
+	return nil
+}
+
+func (peer *LocalPeer) addConnection(conn Connection) {
+	peer.Lock()
+	defer peer.Unlock()
+	peer.connections[conn.Remote().Name] = conn
+	peer.Version++
+}
+
+func (peer *LocalPeer) deleteConnection(conn Connection) {
+	peer.Lock()
+	defer peer.Unlock()
+	delete(peer.connections, conn.Remote().Name)
+	peer.Version++
+}
+
+func (peer *LocalPeer) connectionEstablished(conn Connection) {
+	peer.Lock()
+	defer peer.Unlock()
+	peer.Version++
+}
+
+func (peer *LocalPeer) connectionCount() int {
+	peer.RLock()
+	defer peer.RUnlock()
+	return len(peer.connections)
+}
+
+func (peer *LocalPeer) setShortID(shortID PeerShortID) {
+	peer.Lock()
+	defer peer.Unlock()
+	peer.ShortID = shortID
+	peer.Version++
+}
+
+func (peer *LocalPeer) setVersionBeyond(version uint64) bool {
+	peer.Lock()
+	defer peer.Unlock()
+	if version >= peer.Version {
+		peer.Version = version + 1
+		return true
+	}
+	return false
+}

mesh/mocks_test.go

@@ -0,0 +1,115 @@
+// No mocks are tested by this file.
+//
+// It supplies some mock implementations to other unit tests, and is
+// named "...test.go" so it is only compiled under `go test`.
+
+package mesh
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// Add to peers a connection from peers.ourself to p
+func (peers *Peers) AddTestConnection(p *Peer) {
+	summary := p.PeerSummary
+	summary.Version = 0
+	toPeer := NewPeerFromSummary(summary)
+	peers.FetchWithDefault(toPeer) // Has side-effect of incrementing refcount
+	conn := newMockConnection(peers.ourself.Peer, toPeer)
+	peers.ourself.addConnection(conn)
+	peers.ourself.connectionEstablished(conn)
+}
+
+// Add to peers a connection from p1 to p2
+func (peers *Peers) AddTestRemoteConnection(p1, p2 *Peer) {
+	fromPeer := NewPeerFrom(p1)
+	fromPeer = peers.FetchWithDefault(fromPeer)
+	toPeer := NewPeerFrom(p2)
+	toPeer = peers.FetchWithDefault(toPeer)
+	peers.ourself.addConnection(&RemoteConnection{fromPeer, toPeer, "", false, false})
+}
+
+func (peers *Peers) DeleteTestConnection(p *Peer) {
+	toName := p.Name
+	toPeer := peers.Fetch(toName)
+	peers.Dereference(toPeer)
+	conn, _ := peers.ourself.ConnectionTo(toName)
+	peers.ourself.deleteConnection(conn)
+}
+
+// mockConnection used in testing is very similar to a
+// RemoteConnection, without the RemoteTCPAddr(). We are making it a
+// separate type in order to distinguish what is created by the test
+// from what is created by the real code.
+func newMockConnection(from, to *Peer) Connection {
+	type mockConnection struct{ RemoteConnection }
+	return &mockConnection{RemoteConnection{from, to, "", false, false}}
+}
+
+func checkEqualConns(t *testing.T, ourName PeerName, got, wanted map[PeerName]Connection) {
+	checkConns := make(PeerNameSet)
+	for _, conn := range wanted {
+		checkConns[conn.Remote().Name] = void
+	}
+	for _, conn := range got {
+		remoteName := conn.Remote().Name
+		if _, found := checkConns[remoteName]; found {
+			delete(checkConns, remoteName)
+		} else {
+			require.FailNow(t, fmt.Sprintf("Unexpected connection from %s to %s", ourName, remoteName))
+		}
+	}
+	if len(checkConns) > 0 {
+		require.FailNow(t, fmt.Sprintf("Expected connections not found: from %s to %v", ourName, checkConns))
+	}
+}
+
+// Get all the peers from a Peers in a slice
+func (peers *Peers) allPeers() []*Peer {
+	var res []*Peer
+	for _, peer := range peers.byName {
+		res = append(res, peer)
+	}
+	return res
+}
+
+func (peers *Peers) allPeersExcept(excludeName PeerName) []*Peer {
+	res := peers.allPeers()
+	for i, peer := range res {
+		if peer.Name == excludeName {
+			return append(res[:i], res[i+1:]...)
+		}
+	}
+	return res
+}
+
+// Check that the peers slice matches the wanted peers
+func checkPeerArray(t *testing.T, peers []*Peer, wantedPeers ...*Peer) {
+	checkTopologyPeers(t, false, peers, wantedPeers...)
+}
+
+// Check that the peers slice matches the wanted peers and optionally
+// all of their connections
+func checkTopologyPeers(t *testing.T, checkConns bool, peers []*Peer, wantedPeers ...*Peer) {
+	check := make(map[PeerName]*Peer)
+	for _, peer := range wantedPeers {
+		check[peer.Name] = peer
+	}
+	for _, peer := range peers {
+		name := peer.Name
+		if wantedPeer, found := check[name]; found {
+			if checkConns {
+				checkEqualConns(t, name, peer.connections, wantedPeer.connections)
+			}
+			delete(check, name)
+		} else {
+			require.FailNow(t, fmt.Sprintf("Unexpected peer: %s", name))
+		}
+	}
+	if len(check) > 0 {
+		require.FailNow(t, fmt.Sprintf("Expected peers not found: %v", check))
+	}
+}

mesh/overlay.go

@@ -0,0 +1,110 @@
+package mesh
+
+import (
+	"net"
+)
+
+type Overlay interface {
+	// Enhance a features map with overlay-related features
+	AddFeaturesTo(map[string]string)
+
+	// Prepare on overlay connection. The connection should remain
+	// passive until it has been Confirm()ed.
+	PrepareConnection(OverlayConnectionParams) (OverlayConnection, error)
+
+	// Obtain diagnostic information specific to the overlay
+	Diagnostics() interface{}
+}
+
+type OverlayConnectionParams struct {
+	RemotePeer *Peer
+
+	// The local address of the corresponding TCP connection. Used to
+	// derive the local IP address for sending. May differ for
+	// different overlay connections.
+	LocalAddr *net.TCPAddr
+
+	// The remote address of the corresponding TCP connection. Used to
+	// determine the address to send to, but only if the TCP
+	// connection is outbound. Otherwise the Overlay needs to discover
+	// it (e.g. from incoming datagrams).
+	RemoteAddr *net.TCPAddr
+
+	// Is the corresponding TCP connection outbound?
+	Outbound bool
+
+	// Unique identifier for this connection
+	ConnUID uint64
+
+	// Session key, if connection is encrypted; nil otherwise
+	SessionKey *[32]byte
+
+	// Function to send a control message to the counterpart
+	// overlay connection.
+	SendControlMessage func(tag byte, msg []byte) error
+
+	// Features passed at connection initiation
+	Features map[string]string
+}
+
+// All of the machinery to manage overlay connectivity to a particular
+// peer
+type OverlayConnection interface {
+	// Confirm that the connection is really wanted, and so the
+	// Overlay should begin heartbeats etc. to verify the operation of
+	// the overlay connection.
+	Confirm()
+
+	// A channel indicating that the overlay connection is
+	// established, i.e. its operation has been confirmed.
+	EstablishedChannel() <-chan struct{}
+
+	// A channel indicating an error from the overlay connection.  The
+	// overlay connection is not expected to be operational after the
+	// first error, so the channel only needs to buffer a single
+	// error.
+	ErrorChannel() <-chan error
+
+	Stop()
+
+	// Handle a message from the peer.  'tag' exists for
+	// compatibility, and should always be
+	// ProtocolOverlayControlMessage for non-sleeve overlays.
+	ControlMessage(tag byte, msg []byte)
+
+	// User facing overlay name
+	DisplayName() string
+}
+
+type NullOverlay struct{}
+
+func (NullOverlay) AddFeaturesTo(map[string]string) {
+}
+
+func (NullOverlay) PrepareConnection(OverlayConnectionParams) (OverlayConnection, error) {
+	return NullOverlay{}, nil
+}
+
+func (NullOverlay) Diagnostics() interface{} {
+	return nil
+}
+func (NullOverlay) Confirm() {
+}
+
+func (NullOverlay) EstablishedChannel() <-chan struct{} {
+	return nil
+}
+
+func (NullOverlay) ErrorChannel() <-chan error {
+	return nil
+}
+
+func (NullOverlay) Stop() {
+}
+
+func (NullOverlay) ControlMessage(byte, []byte) {
+}
+
+func (NullOverlay) DisplayName() string {
+	return "null"
+}

mesh/peer.go

@@ -0,0 +1,173 @@
+package mesh
+
+import (
+	"crypto/rand"
+	"encoding/binary"
+	"fmt"
+	"sort"
+	"strconv"
+)
+
+func randBytes(n int) []byte {
+	buf := make([]byte, n)
+	_, err := rand.Read(buf)
+	checkFatal(err)
+	return buf
+}
+
+func randUint64() (r uint64) {
+	return binary.LittleEndian.Uint64(randBytes(8))
+}
+
+func randUint16() (r uint16) {
+	return binary.LittleEndian.Uint16(randBytes(2))
+}
+
+type PeerUID uint64
+
+func randomPeerUID() PeerUID {
+	return PeerUID(randUint64())
+}
+
+func ParsePeerUID(s string) (PeerUID, error) {
+	uid, err := strconv.ParseUint(s, 10, 64)
+	return PeerUID(uid), err
+}
+
+// Short IDs exist for the sake of fast datapath.  They are 12 bits,
+// randomly assigned, but we detect and recover from collisions.  This
+// does limit us to 4096 peers, but that should be sufficient for a
+// while.
+type PeerShortID uint16
+
+const PeerShortIDBits = 12
+
+func randomPeerShortID() PeerShortID {
+	return PeerShortID(randUint16() & (1<<PeerShortIDBits - 1))
+}
+
+type PeerSummary struct {
+	NameByte   []byte
+	NickName   string
+	UID        PeerUID
+	Version    uint64
+	ShortID    PeerShortID
+	HasShortID bool
+}
+
+type Peer struct {
+	Name PeerName
+	PeerSummary
+	localRefCount uint64 // maintained by Peers
+	connections   map[PeerName]Connection
+}
+
+type ListOfPeers []*Peer
+
+func (lop ListOfPeers) Len() int {
+	return len(lop)
+}
+func (lop ListOfPeers) Swap(i, j int) {
+	lop[i], lop[j] = lop[j], lop[i]
+}
+func (lop ListOfPeers) Less(i, j int) bool {
+	return lop[i].Name < lop[j].Name
+}
+
+type ConnectionSet map[Connection]struct{}
+
+func NewPeerFromSummary(summary PeerSummary) *Peer {
+	return &Peer{
+		Name:        PeerNameFromBin(summary.NameByte),
+		PeerSummary: summary,
+		connections: make(map[PeerName]Connection),
+	}
+}
+
+func NewPeer(name PeerName, nickName string, uid PeerUID, version uint64, shortID PeerShortID) *Peer {
+	return NewPeerFromSummary(PeerSummary{
+		NameByte:   name.Bin(),
+		NickName:   nickName,
+		UID:        uid,
+		Version:    version,
+		ShortID:    shortID,
+		HasShortID: true,
+	})
+}
+
+func NewPeerFrom(peer *Peer) *Peer {
+	return NewPeerFromSummary(peer.PeerSummary)
+}
+
+func (peer *Peer) String() string {
+	return fmt.Sprint(peer.Name, "(", peer.NickName, ")")
+}
+
+// Calculate the routing table from this peer to all peers reachable
+// from it, returning a "next hop" map of PeerNameX -> PeerNameY,
+// which says "in order to send a message to X, the peer should send
+// the message to its neighbour Y".
+//
+// Because currently we do not have weightings on the connections
+// between peers, there is no need to use a minimum spanning tree
+// algorithm. Instead we employ the simpler and cheaper breadth-first
+// widening. The computation is deterministic, which ensures that when
+// it is performed on the same data by different peers, they get the
+// same result. This is important since otherwise we risk message loss
+// or routing cycles.
+//
+// When the 'establishedAndSymmetric' flag is set, only connections
+// that are marked as 'established' and are symmetric (i.e. where both
+// sides indicate they have a connection to the other) are considered.
+//
+// When a non-nil stopAt peer is supplied, the widening stops when it
+// reaches that peer. The boolean return indicates whether that has
+// happened.
+//
+// NB: This function should generally be invoked while holding a read
+// lock on Peers and LocalPeer.
+func (peer *Peer) Routes(stopAt *Peer, establishedAndSymmetric bool) (bool, map[PeerName]PeerName) {
+	routes := make(unicastRoutes)
+	routes[peer.Name] = UnknownPeerName
+	nextWorklist := []*Peer{peer}
+	for len(nextWorklist) > 0 {
+		worklist := nextWorklist
+		sort.Sort(ListOfPeers(worklist))
+		nextWorklist = []*Peer{}
+		for _, curPeer := range worklist {
+			if curPeer == stopAt {
+				return true, routes
+			}
+			curPeer.ForEachConnectedPeer(establishedAndSymmetric, routes,
+				func(remotePeer *Peer) {
+					nextWorklist = append(nextWorklist, remotePeer)
+					remoteName := remotePeer.Name
+					// We now know how to get to remoteName: the same
+					// way we get to curPeer. Except, if curPeer is
+					// the starting peer in which case we know we can
+					// reach remoteName directly.
+					if curPeer == peer {
+						routes[remoteName] = remoteName
+					} else {
+						routes[remoteName] = routes[curPeer.Name]
+					}
+				})
+		}
+	}
+	return false, routes
+}
+
+func (peer *Peer) ForEachConnectedPeer(establishedAndSymmetric bool, exclude map[PeerName]PeerName, f func(*Peer)) {
+	for remoteName, conn := range peer.connections {
+		if establishedAndSymmetric && !conn.Established() {
+			continue
+		}
+		if _, found := exclude[remoteName]; found {
+			continue
+		}
+		remotePeer := conn.Remote()
+		if remoteConn, found := remotePeer.connections[peer.Name]; !establishedAndSymmetric || (found && remoteConn.Established()) {
+			f(remotePeer)
+		}
+	}
+}

mesh/peer_name_hash.go

@@ -0,0 +1,46 @@
+// +build peer_name_hash
+
+// Let peer names be sha256 hashes...of anything (as long as it's
+// unique).
+
+package mesh
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+)
+
+type PeerName string
+
+const (
+	PeerNameFlavour = "hash"
+	NameSize        = sha256.Size >> 1
+	UnknownPeerName = PeerName("")
+)
+
+func PeerNameFromUserInput(userInput string) (PeerName, error) {
+	// fixed-length identity
+	nameByteAry := sha256.Sum256([]byte(userInput))
+	return PeerNameFromBin(nameByteAry[:NameSize]), nil
+}
+
+func PeerNameFromString(nameStr string) (PeerName, error) {
+	if _, err := hex.DecodeString(nameStr); err != nil {
+		return UnknownPeerName, err
+	}
+	return PeerName(nameStr), nil
+}
+
+func PeerNameFromBin(nameByte []byte) PeerName {
+	return PeerName(hex.EncodeToString(nameByte))
+}
+
+func (name PeerName) Bin() []byte {
+	res, err := hex.DecodeString(string(name))
+	checkFatal(err)
+	return res
+}
+
+func (name PeerName) String() string {
+	return string(name)
+}

mesh/peer_name_mac.go

@@ -0,0 +1,75 @@
+// The !peer_name_alternative effectively makes this the default,
+// i.e. to choose an alternative, run "go build -tags
+// 'peer_name_alternative peer_name_hash'"
+
+// +build peer_name_mac !peer_name_alternative
+
+// Let peer names be MACs...
+//
+// MACs need to be unique across our network, or bad things will
+// happen anyway. So they make pretty good candidates for peer
+// names. And doing so is pretty efficient both computationally and
+// network overhead wise.
+//
+// Note that we do not mandate *what* MAC should be used as the peer
+// name. In particular it doesn't actually have to be the MAC of, say,
+// the network interface the peer is sniffing on.
+
+package mesh
+
+import (
+	"net"
+)
+
+// PeerName is used as a map key. Since net.HardwareAddr isn't
+// suitable for that - it's a slice, and slices can't be map keys - we
+// convert that to/from uint64.
+type PeerName uint64
+
+const (
+	PeerNameFlavour = "mac"
+	NameSize        = 6
+	UnknownPeerName = PeerName(0)
+)
+
+func PeerNameFromUserInput(userInput string) (PeerName, error) {
+	res, err := PeerNameFromString(userInput)
+	return res, err
+}
+
+func PeerNameFromString(nameStr string) (PeerName, error) {
+	mac, err := net.ParseMAC(nameStr)
+	if err != nil {
+		return UnknownPeerName, err
+	}
+	return PeerName(macint(mac)), nil
+}
+
+func PeerNameFromBin(nameByte []byte) PeerName {
+	return PeerName(macint(net.HardwareAddr(nameByte)))
+}
+
+func (name PeerName) Bin() []byte {
+	return intmac(uint64(name))
+}
+
+func (name PeerName) String() string {
+	return intmac(uint64(name)).String()
+}
+
+func macint(mac net.HardwareAddr) (r uint64) {
+	for _, b := range mac {
+		r <<= 8
+		r |= uint64(b)
+	}
+	return
+}
+
+func intmac(key uint64) (r net.HardwareAddr) {
+	r = make([]byte, 6)
+	for i := 5; i >= 0; i-- {
+		r[i] = byte(key)
+		key >>= 8
+	}
+	return
+}

mesh/peers.go

@@ -0,0 +1,538 @@
+package mesh
+
+import (
+	"bytes"
+	"encoding/gob"
+	"fmt"
+	"io"
+	"math/rand"
+	"sync"
+)
+
+var void = struct{}{}
+
+type Peers struct {
+	sync.RWMutex
+	ourself   *LocalPeer
+	byName    map[PeerName]*Peer
+	byShortID map[PeerShortID]ShortIDPeers
+	onGC      []func(*Peer)
+
+	// Called when the mapping from short ids to peers changes
+	onInvalidateShortIDs []func()
+}
+
+type ShortIDPeers struct {
+	// If we know about a single peer with the short id, this is
+	// that peer.  If there is a collision, this is the peer with
+	// the lowest Name.
+	peer *Peer
+
+	// In case of a collision, this holds the other peers.
+	others []*Peer
+}
+
+type UnknownPeerError struct {
+	Name PeerName
+}
+
+func (upe UnknownPeerError) Error() string {
+	return fmt.Sprint("Reference to unknown peer ", upe.Name)
+}
+
+type PeerNameSet map[PeerName]struct{}
+
+type ConnectionSummary struct {
+	NameByte      []byte
+	RemoteTCPAddr string
+	Outbound      bool
+	Established   bool
+}
+
+// Pending notifications due to changes to Peers that need to be sent
+// out once the Peers is unlocked.
+type PeersPendingNotifications struct {
+	// Peers that have been GCed
+	removed []*Peer
+
+	// The mapping from shorts ids to peers changed
+	invalidateShortIDs bool
+
+	// The local short ID needs reassigning due to a collision
+	reassignLocalShortID bool
+
+	// The local peer was modified
+	localPeerModified bool
+}
+
+func NewPeers(ourself *LocalPeer) *Peers {
+	peers := &Peers{
+		ourself:   ourself,
+		byName:    make(map[PeerName]*Peer),
+		byShortID: make(map[PeerShortID]ShortIDPeers),
+	}
+	peers.FetchWithDefault(ourself.Peer)
+	return peers
+}
+
+func (peers *Peers) OnGC(callback func(*Peer)) {
+	peers.Lock()
+	defer peers.Unlock()
+
+	// Although the array underlying peers.onGC might be accessed
+	// without holding the lock in unlockAndNotify, we don't
+	// support removing callbacks, so a simple append here is
+	// safe.
+	peers.onGC = append(peers.onGC, callback)
+}
+
+func (peers *Peers) OnInvalidateShortIDs(callback func()) {
+	peers.Lock()
+	defer peers.Unlock()
+
+	// Safe, as in OnGC
+	peers.onInvalidateShortIDs = append(peers.onInvalidateShortIDs, callback)
+}
+
+func (peers *Peers) unlockAndNotify(pending *PeersPendingNotifications) {
+	broadcastLocalPeer := (pending.reassignLocalShortID && peers.reassignLocalShortID(pending)) ||
+		pending.localPeerModified
+	onGC := peers.onGC
+	onInvalidateShortIDs := peers.onInvalidateShortIDs
+	peers.Unlock()
+
+	if pending.removed != nil {
+		for _, callback := range onGC {
+			for _, peer := range pending.removed {
+				callback(peer)
+			}
+		}
+	}
+
+	if pending.invalidateShortIDs {
+		for _, callback := range onInvalidateShortIDs {
+			callback()
+		}
+	}
+
+	if broadcastLocalPeer {
+		peers.ourself.broadcastPeerUpdate()
+	}
+}
+
+func (peers *Peers) addByShortID(peer *Peer, pending *PeersPendingNotifications) {
+	if !peer.HasShortID {
+		return
+	}
+
+	entry, ok := peers.byShortID[peer.ShortID]
+	if !ok {
+		entry = ShortIDPeers{peer: peer}
+	} else if entry.peer == nil {
+		// This short ID is free, but was used in the past.
+		// Because we are reusing it, it's an invalidation
+		// event.
+		entry.peer = peer
+		pending.invalidateShortIDs = true
+	} else if peer.Name < entry.peer.Name {
+		// Short ID collision, this peer becomes the principal
+		// peer for the short ID, bumping the previous one
+		// into others.
+
+		if entry.peer == peers.ourself.Peer {
+			// The bumped peer is peers.ourself, so we
+			// need to look for a new short id
+			pending.reassignLocalShortID = true
+		}
+
+		entry.others = append(entry.others, entry.peer)
+		entry.peer = peer
+		pending.invalidateShortIDs = true
+	} else {
+		// Short ID collision, this peer is secondary
+		entry.others = append(entry.others, peer)
+	}
+
+	peers.byShortID[peer.ShortID] = entry
+}
+
+func (peers *Peers) deleteByShortID(peer *Peer, pending *PeersPendingNotifications) {
+	if !peer.HasShortID {
+		return
+	}
+
+	entry := peers.byShortID[peer.ShortID]
+	var otherIndex int
+
+	if peer != entry.peer {
+		// peer is secondary, find its index in others
+		otherIndex = -1
+
+		for i, other := range entry.others {
+			if peer == other {
+				otherIndex = i
+				break
+			}
+		}
+
+		if otherIndex < 0 {
+			return
+		}
+	} else if len(entry.others) != 0 {
+		// need to find the peer with the lowest name to
+		// become the new principal one
+		otherIndex = 0
+		minName := entry.others[0].Name
+
+		for i := 1; i < len(entry.others); i++ {
+			otherName := entry.others[i].Name
+			if otherName < minName {
+				minName = otherName
+				otherIndex = i
+			}
+		}
+
+		entry.peer = entry.others[otherIndex]
+		pending.invalidateShortIDs = true
+	} else {
+		// This is the last peer with the short id.  We clear
+		// the entry, don't delete it, in order to detect when
+		// it gets re-used.
+		peers.byShortID[peer.ShortID] = ShortIDPeers{}
+		return
+	}
+
+	entry.others[otherIndex] = entry.others[len(entry.others)-1]
+	entry.others = entry.others[:len(entry.others)-1]
+	peers.byShortID[peer.ShortID] = entry
+}
+
+func (peers *Peers) reassignLocalShortID(pending *PeersPendingNotifications) bool {
+	newShortID, ok := peers.chooseShortID()
+	if ok {
+		peers.setLocalShortID(newShortID, pending)
+		return true
+	}
+
+	// Otherwise we'll try again later on in garbageColleect
+	return false
+}
+
+func (peers *Peers) setLocalShortID(newShortID PeerShortID, pending *PeersPendingNotifications) {
+	peers.deleteByShortID(peers.ourself.Peer, pending)
+	peers.ourself.setShortID(newShortID)
+	peers.addByShortID(peers.ourself.Peer, pending)
+}
+
+// Choose an available short id at random
+func (peers *Peers) chooseShortID() (PeerShortID, bool) {
+	rng := rand.New(rand.NewSource(int64(randUint64())))
+
+	// First, just try picking some short ids at random, and
+	// seeing if they are available:
+	for i := 0; i < 10; i++ {
+		shortID := PeerShortID(rng.Intn(1 << PeerShortIDBits))
+		if peers.byShortID[shortID].peer == nil {
+			return shortID, true
+		}
+	}
+
+	// Looks like most short ids are used.  So count the number of
+	// unused ones, and pick one at random.
+	available := int(1 << PeerShortIDBits)
+	for _, entry := range peers.byShortID {
+		if entry.peer != nil {
+			available--
+		}
+	}
+
+	if available == 0 {
+		// All short ids are used.
+		return 0, false
+	}
+
+	n := rng.Intn(available)
+	var i PeerShortID
+	for {
+		if peers.byShortID[i].peer == nil {
+			if n == 0 {
+				return PeerShortID(i), true
+			}
+
+			n--
+		}
+
+		i++
+	}
+}
+
+func (peers *Peers) FetchWithDefault(peer *Peer) *Peer {
+	peers.Lock()
+	var pending PeersPendingNotifications
+	defer peers.unlockAndNotify(&pending)
+
+	if existingPeer, found := peers.byName[peer.Name]; found {
+		existingPeer.localRefCount++
+		return existingPeer
+	}
+
+	peers.byName[peer.Name] = peer
+	peers.addByShortID(peer, &pending)
+	peer.localRefCount++
+	return peer
+}
+
+func (peers *Peers) Fetch(name PeerName) *Peer {
+	peers.RLock()
+	defer peers.RUnlock()
+	return peers.byName[name]
+}
+
+func (peers *Peers) FetchAndAddRef(name PeerName) *Peer {
+	peers.Lock()
+	defer peers.Unlock()
+	peer := peers.byName[name]
+	if peer != nil {
+		peer.localRefCount++
+	}
+	return peer
+}
+
+func (peers *Peers) FetchByShortID(shortID PeerShortID) *Peer {
+	peers.RLock()
+	defer peers.RUnlock()
+	return peers.byShortID[shortID].peer
+}
+
+func (peers *Peers) Dereference(peer *Peer) {
+	peers.Lock()
+	defer peers.Unlock()
+	peer.localRefCount--
+}
+
+func (peers *Peers) ForEach(fun func(*Peer)) {
+	peers.RLock()
+	defer peers.RUnlock()
+	for _, peer := range peers.byName {
+		fun(peer)
+	}
+}
+
+// Merge an incoming update with our own topology.
+//
+// We add peers hitherto unknown to us, and update peers for which the
+// update contains a more recent version than known to us. The return
+// value is a) a representation of the received update, and b) an
+// "improved" update containing just these new/updated elements.
+func (peers *Peers) ApplyUpdate(update []byte) (PeerNameSet, PeerNameSet, error) {
+	peers.Lock()
+	var pending PeersPendingNotifications
+	defer peers.unlockAndNotify(&pending)
+
+	newPeers, decodedUpdate, decodedConns, err := peers.decodeUpdate(update)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	// By this point, we know the update doesn't refer to any peers we
+	// have no knowledge of. We can now apply the update. Start by
+	// adding in any new peers.
+	for name, newPeer := range newPeers {
+		peers.byName[name] = newPeer
+		peers.addByShortID(newPeer, &pending)
+	}
+
+	// Now apply the updates
+	newUpdate := peers.applyUpdate(decodedUpdate, decodedConns, &pending)
+	peers.garbageCollect(&pending)
+	for _, peerRemoved := range pending.removed {
+		delete(newUpdate, peerRemoved.Name)
+	}
+
+	updateNames := make(PeerNameSet)
+	for _, peer := range decodedUpdate {
+		updateNames[peer.Name] = void
+	}
+
+	return updateNames, newUpdate, nil
+}
+
+func (peers *Peers) Names() PeerNameSet {
+	peers.RLock()
+	defer peers.RUnlock()
+
+	names := make(PeerNameSet)
+	for name := range peers.byName {
+		names[name] = void
+	}
+	return names
+}
+
+func (peers *Peers) EncodePeers(names PeerNameSet) []byte {
+	buf := new(bytes.Buffer)
+	enc := gob.NewEncoder(buf)
+	peers.RLock()
+	defer peers.RUnlock()
+	for name := range names {
+		if peer, found := peers.byName[name]; found {
+			if peer == peers.ourself.Peer {
+				peers.ourself.Encode(enc)
+			} else {
+				peer.Encode(enc)
+			}
+		}
+	}
+	return buf.Bytes()
+}
+
+func (peers *Peers) GarbageCollect() {
+	peers.Lock()
+	var pending PeersPendingNotifications
+	defer peers.unlockAndNotify(&pending)
+
+	peers.garbageCollect(&pending)
+}
+
+func (peers *Peers) garbageCollect(pending *PeersPendingNotifications) {
+	peers.ourself.RLock()
+	_, reached := peers.ourself.Routes(nil, false)
+	peers.ourself.RUnlock()
+
+	for name, peer := range peers.byName {
+		if _, found := reached[peer.Name]; !found && peer.localRefCount == 0 {
+			delete(peers.byName, name)
+			peers.deleteByShortID(peer, pending)
+			pending.removed = append(pending.removed, peer)
+		}
+	}
+
+	if peers.byShortID[peers.ourself.ShortID].peer != peers.ourself.Peer {
+		// The local peer doesn't own its short id.  Garbage
+		// collection might have freed some up, so try to
+		// reassign.
+		pending.reassignLocalShortID = true
+	}
+}
+
+func (peers *Peers) decodeUpdate(update []byte) (newPeers map[PeerName]*Peer, decodedUpdate []*Peer, decodedConns [][]ConnectionSummary, err error) {
+	newPeers = make(map[PeerName]*Peer)
+	decodedUpdate = []*Peer{}
+	decodedConns = [][]ConnectionSummary{}
+
+	decoder := gob.NewDecoder(bytes.NewReader(update))
+
+	for {
+		peerSummary, connSummaries, decErr := decodePeer(decoder)
+		if decErr == io.EOF {
+			break
+		} else if decErr != nil {
+			err = decErr
+			return
+		}
+		newPeer := NewPeerFromSummary(peerSummary)
+		decodedUpdate = append(decodedUpdate, newPeer)
+		decodedConns = append(decodedConns, connSummaries)
+		if _, found := peers.byName[newPeer.Name]; !found {
+			newPeers[newPeer.Name] = newPeer
+		}
+	}
+
+	for _, connSummaries := range decodedConns {
+		for _, connSummary := range connSummaries {
+			remoteName := PeerNameFromBin(connSummary.NameByte)
+			if _, found := newPeers[remoteName]; found {
+				continue
+			}
+			if _, found := peers.byName[remoteName]; found {
+				continue
+			}
+			// Update refers to a peer which we have no knowledge
+			// of. Thus we can't apply the update. Abort.
+			err = UnknownPeerError{remoteName}
+			return
+		}
+	}
+	return
+}
+
+func (peers *Peers) applyUpdate(decodedUpdate []*Peer, decodedConns [][]ConnectionSummary, pending *PeersPendingNotifications) PeerNameSet {
+	newUpdate := make(PeerNameSet)
+	for idx, newPeer := range decodedUpdate {
+		connSummaries := decodedConns[idx]
+		name := newPeer.Name
+		// guaranteed to find peer in the peers.byName
+		switch peer := peers.byName[name]; peer {
+		case peers.ourself.Peer:
+			if newPeer.UID != peer.UID {
+				// The update contains information about an old
+				// incarnation of ourselves. We increase our version
+				// number beyond that which we received, so our
+				// information supersedes the old one when it is
+				// received by other peers.
+				pending.localPeerModified = peers.ourself.setVersionBeyond(newPeer.Version)
+			}
+		case newPeer:
+			peer.connections = makeConnsMap(peer, connSummaries, peers.byName)
+			newUpdate[name] = void
+		default: // existing peer
+			if newPeer.Version < peer.Version ||
+				(newPeer.Version == peer.Version && newPeer.UID <= peer.UID) {
+				continue
+			}
+			peer.Version = newPeer.Version
+			peer.UID = newPeer.UID
+			peer.connections = makeConnsMap(peer, connSummaries, peers.byName)
+
+			if newPeer.ShortID != peer.ShortID {
+				peers.deleteByShortID(peer, pending)
+				peer.ShortID = newPeer.ShortID
+				peers.addByShortID(peer, pending)
+			}
+			newUpdate[name] = void
+		}
+	}
+	return newUpdate
+}
+
+func (peer *Peer) Encode(enc *gob.Encoder) {
+	checkFatal(enc.Encode(peer.PeerSummary))
+
+	connSummaries := []ConnectionSummary{}
+	for _, conn := range peer.connections {
+		connSummaries = append(connSummaries, ConnectionSummary{
+			conn.Remote().NameByte,
+			conn.RemoteTCPAddr(),
+			conn.Outbound(),
+			conn.Established(),
+		})
+	}
+
+	checkFatal(enc.Encode(connSummaries))
+}
+
+func (peer *LocalPeer) Encode(enc *gob.Encoder) {
+	peer.RLock()
+	defer peer.RUnlock()
+	peer.Peer.Encode(enc)
+}
+
+func decodePeer(dec *gob.Decoder) (peerSummary PeerSummary, connSummaries []ConnectionSummary, err error) {
+	if err = dec.Decode(&peerSummary); err != nil {
+		return
+	}
+	if err = dec.Decode(&connSummaries); err != nil {
+		return
+	}
+	return
+}
+
+func makeConnsMap(peer *Peer, connSummaries []ConnectionSummary, byName map[PeerName]*Peer) map[PeerName]Connection {
+	conns := make(map[PeerName]Connection)
+	for _, connSummary := range connSummaries {
+		name := PeerNameFromBin(connSummary.NameByte)
+		remotePeer := byName[name]
+		conn := NewRemoteConnection(peer, remotePeer, connSummary.RemoteTCPAddr, connSummary.Outbound, connSummary.Established)
+		conns[name] = conn
+	}
+	return conns
+}

mesh/peers_test.go

@@ -0,0 +1,376 @@
+package mesh
+
+import (
+	"fmt"
+	"math/rand"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TODO we should also test:
+//
+// - applying an incremental update, including the case where that
+//   leads to an UnknownPeerError
+//
+// - the "improved update" calculation
+//
+// - non-gc of peers that are only referenced locally
+
+func newNode(name PeerName) (*Peer, *Peers) {
+	peer := NewLocalPeer(name, "", nil)
+	peers := NewPeers(peer)
+	return peer.Peer, peers
+}
+
+// Check that ApplyUpdate copies the whole topology from peers
+func checkApplyUpdate(t *testing.T, peers *Peers) {
+	dummyName, _ := PeerNameFromString("99:00:00:01:00:00")
+	// We need a new node outside of the network, with a connection
+	// into it.
+	_, testBedPeers := newNode(dummyName)
+	testBedPeers.AddTestConnection(peers.ourself.Peer)
+	testBedPeers.ApplyUpdate(peers.EncodePeers(peers.Names()))
+
+	checkTopologyPeers(t, true, testBedPeers.allPeersExcept(dummyName), peers.allPeers()...)
+}
+
+func TestPeersEncoding(t *testing.T) {
+	const numNodes = 20
+	const numIters = 1000
+	var peer [numNodes]*Peer
+	var ps [numNodes]*Peers
+	for i := 0; i < numNodes; i++ {
+		name, _ := PeerNameFromString(fmt.Sprintf("%02d:00:00:01:00:00", i))
+		peer[i], ps[i] = newNode(name)
+	}
+
+	var conns []struct{ from, to int }
+	for i := 0; i < numIters; i++ {
+		oper := rand.Intn(2)
+		switch oper {
+		case 0:
+			from, to := rand.Intn(numNodes), rand.Intn(numNodes)
+			if from != to {
+				if _, found := peer[from].connections[peer[to].Name]; !found {
+					ps[from].AddTestConnection(peer[to])
+					conns = append(conns, struct{ from, to int }{from, to})
+					checkApplyUpdate(t, ps[from])
+				}
+			}
+		case 1:
+			if len(conns) > 0 {
+				n := rand.Intn(len(conns))
+				c := conns[n]
+				ps[c.from].DeleteTestConnection(peer[c.to])
+				ps[c.from].GarbageCollect()
+				checkApplyUpdate(t, ps[c.from])
+				conns = append(conns[:n], conns[n+1:]...)
+			}
+		}
+	}
+}
+
+func garbageCollect(peers *Peers) []*Peer {
+	var removed []*Peer
+	peers.OnGC(func(peer *Peer) { removed = append(removed, peer) })
+	peers.GarbageCollect()
+	return removed
+}
+
+func TestPeersGarbageCollection(t *testing.T) {
+	const (
+		peer1NameString = "01:00:00:01:00:00"
+		peer2NameString = "02:00:00:02:00:00"
+		peer3NameString = "03:00:00:03:00:00"
+	)
+	var (
+		peer1Name, _ = PeerNameFromString(peer1NameString)
+		peer2Name, _ = PeerNameFromString(peer2NameString)
+		peer3Name, _ = PeerNameFromString(peer3NameString)
+	)
+
+	// Create some peers with some connections to each other
+	p1, ps1 := newNode(peer1Name)
+	p2, ps2 := newNode(peer2Name)
+	p3, ps3 := newNode(peer3Name)
+	ps1.AddTestConnection(p2)
+	ps2.AddTestRemoteConnection(p1, p2)
+	ps2.AddTestConnection(p1)
+	ps2.AddTestConnection(p3)
+	ps3.AddTestConnection(p1)
+	ps1.AddTestConnection(p3)
+	ps2.AddTestRemoteConnection(p1, p3)
+	ps2.AddTestRemoteConnection(p3, p1)
+
+	// Every peer is referenced, so nothing should be dropped
+	require.Empty(t, garbageCollect(ps1), "peers removed")
+	require.Empty(t, garbageCollect(ps2), "peers removed")
+	require.Empty(t, garbageCollect(ps3), "peers removed")
+
+	// Drop the connection from 2 to 3, and 3 isn't garbage-collected
+	// because 1 has a connection to 3
+	ps2.DeleteTestConnection(p3)
+	require.Empty(t, garbageCollect(ps2), "peers removed")
+
+	// Drop the connection from 1 to 3, and 3 will get removed by
+	// garbage-collection
+	ps1.DeleteTestConnection(p3)
+	checkPeerArray(t, garbageCollect(ps1), p3)
+}
+
+func TestShortIDCollisions(t *testing.T) {
+	rng := rand.New(rand.NewSource(time.Now().UnixNano()))
+	_, peers := newNode(PeerName(1 << PeerShortIDBits))
+
+	// Make enough peers that short id collisions are
+	// overwhelmingly likely
+	ps := make([]*Peer, 1<<PeerShortIDBits)
+	for i := 0; i < 1<<PeerShortIDBits; i++ {
+		ps[i] = NewPeer(PeerName(i), "", PeerUID(i), 0,
+			PeerShortID(rng.Intn(1<<PeerShortIDBits)))
+	}
+
+	shuffle := func() {
+		for i := range ps {
+			j := rng.Intn(i + 1)
+			ps[i], ps[j] = ps[j], ps[i]
+		}
+	}
+
+	// Fill peers
+	shuffle()
+	var pending PeersPendingNotifications
+	for _, p := range ps {
+		peers.addByShortID(p, &pending)
+	}
+
+	// Check invariants
+	counts := make([]int, 1<<PeerShortIDBits)
+	saw := func(p *Peer) {
+		if p != peers.ourself.Peer {
+			counts[p.UID]++
+		}
+	}
+
+	for shortID, entry := range peers.byShortID {
+		if entry.peer == nil {
+			// no principal peer for this short id, so
+			// others must be empty
+			require.Empty(t, entry.others)
+			continue
+		}
+
+		require.Equal(t, shortID, entry.peer.ShortID)
+		saw(entry.peer)
+
+		for _, p := range entry.others {
+			saw(p)
+			require.Equal(t, shortID, p.ShortID)
+
+			// the principal peer should have the lowest name
+			require.True(t, p.Name > entry.peer.Name)
+		}
+	}
+
+	// Check that every peer was seen
+	for _, n := range counts {
+		require.Equal(t, 1, n)
+	}
+
+	// Delete all the peers
+	shuffle()
+	for _, p := range ps {
+		peers.deleteByShortID(p, &pending)
+	}
+
+	for _, entry := range peers.byShortID {
+		if entry.peer != peers.ourself.Peer {
+			require.Nil(t, entry.peer)
+		}
+
+		require.Empty(t, entry.others)
+	}
+}
+
+// Test the easy case of short id reassignment, when few short ids are taken
+func TestShortIDReassignmentEasy(t *testing.T) {
+	rng := rand.New(rand.NewSource(time.Now().UnixNano()))
+	_, peers := newNode(PeerName(0))
+
+	for i := 1; i <= 10; i++ {
+		peers.FetchWithDefault(NewPeer(PeerName(i), "", PeerUID(i), 0,
+			PeerShortID(rng.Intn(1<<PeerShortIDBits))))
+	}
+
+	checkShortIDReassignment(t, peers)
+}
+
+// Test the hard case of short id reassignment, when most short ids are taken
+func TestShortIDReassignmentHard(t *testing.T) {
+	rng := rand.New(rand.NewSource(time.Now().UnixNano()))
+	_, peers := newNode(PeerName(1 << PeerShortIDBits))
+
+	// Take all short ids
+	ps := make([]*Peer, 1<<PeerShortIDBits)
+	var pending PeersPendingNotifications
+	for i := 0; i < 1<<PeerShortIDBits; i++ {
+		ps[i] = NewPeer(PeerName(i), "", PeerUID(i), 0,
+			PeerShortID(i))
+		peers.addByShortID(ps[i], &pending)
+	}
+
+	// As all short ids are taken, an attempted reassigment won't
+	// do anything
+	oldShortID := peers.ourself.ShortID
+	require.False(t, peers.reassignLocalShortID(&pending))
+	require.Equal(t, oldShortID, peers.ourself.ShortID)
+
+	// Free up a few ids
+	for i := 0; i < 10; i++ {
+		x := rng.Intn(len(ps))
+		if ps[x] != nil {
+			peers.deleteByShortID(ps[x], &pending)
+			ps[x] = nil
+		}
+	}
+
+	checkShortIDReassignment(t, peers)
+}
+
+func checkShortIDReassignment(t *testing.T, peers *Peers) {
+	oldShortID := peers.ourself.ShortID
+	peers.reassignLocalShortID(&PeersPendingNotifications{})
+	require.NotEqual(t, oldShortID, peers.ourself.ShortID)
+	require.Equal(t, peers.ourself.Peer, peers.byShortID[peers.ourself.ShortID].peer)
+}
+
+func TestShortIDInvalidation(t *testing.T) {
+	_, peers := newNode(PeerName(1 << PeerShortIDBits))
+
+	// need to use a short id that is not the local peer's
+	shortID := peers.ourself.ShortID + 1
+
+	var pending PeersPendingNotifications
+
+	requireInvalidateShortIDs := func(expect bool) {
+		require.Equal(t, expect, pending.invalidateShortIDs)
+		pending.invalidateShortIDs = false
+	}
+
+	// The use of a fresh short id does not cause invalidation
+	a := NewPeer(PeerName(1), "", PeerUID(1), 0, shortID)
+	peers.addByShortID(a, &pending)
+	requireInvalidateShortIDs(false)
+
+	// An addition which does not change the mapping
+	b := NewPeer(PeerName(2), "", PeerUID(2), 0, shortID)
+	peers.addByShortID(b, &pending)
+	requireInvalidateShortIDs(false)
+
+	// An addition which does change the mapping
+	c := NewPeer(PeerName(0), "", PeerUID(0), 0, shortID)
+	peers.addByShortID(c, &pending)
+	requireInvalidateShortIDs(true)
+
+	// A deletion which does not change the mapping
+	peers.deleteByShortID(b, &pending)
+	requireInvalidateShortIDs(false)
+
+	// A deletion which does change the mapping
+	peers.deleteByShortID(c, &pending)
+	requireInvalidateShortIDs(true)
+
+	// Deleting the last peer with a short id does not cause invalidation
+	peers.deleteByShortID(a, &pending)
+	requireInvalidateShortIDs(false)
+
+	// .. but subsequent reuse of that short id does cause invalidation
+	peers.addByShortID(a, &pending)
+	requireInvalidateShortIDs(true)
+}
+
+func TestShortIDPropagation(t *testing.T) {
+	_, peers1 := newNode(PeerName(1))
+	_, peers2 := newNode(PeerName(2))
+
+	peers1.AddTestConnection(peers2.ourself.Peer)
+	peers1.ApplyUpdate(peers2.EncodePeers(peers2.Names()))
+	peers12 := peers1.Fetch(PeerName(2))
+	old := peers12.PeerSummary
+
+	require.True(t,
+		peers2.reassignLocalShortID(&PeersPendingNotifications{}))
+	peers1.ApplyUpdate(peers2.EncodePeers(peers2.Names()))
+	require.NotEqual(t, old.Version, peers12.Version)
+	require.NotEqual(t, old.ShortID, peers12.ShortID)
+}
+
+func TestShortIDCollision(t *testing.T) {
+	// Create 3 peers
+	_, peers1 := newNode(PeerName(1))
+	_, peers2 := newNode(PeerName(2))
+	_, peers3 := newNode(PeerName(3))
+
+	var pending PeersPendingNotifications
+	peers1.setLocalShortID(1, &pending)
+	peers2.setLocalShortID(2, &pending)
+	peers3.setLocalShortID(3, &pending)
+
+	peers2.AddTestConnection(peers1.ourself.Peer)
+	peers3.AddTestConnection(peers2.ourself.Peer)
+
+	// Propogate from 1 to 2 to 3
+	peers2.ApplyUpdate(peers1.EncodePeers(peers1.Names()))
+	peers3.ApplyUpdate(peers2.EncodePeers(peers2.Names()))
+
+	// Force the short id of peer 1 to collide with peer 2.  Peer
+	// 1 has the lowest name, so it gets to keep the short id
+	peers1.setLocalShortID(2, &pending)
+
+	oldShortID := peers2.ourself.ShortID
+	_, updated, _ := peers2.ApplyUpdate(peers1.EncodePeers(peers1.Names()))
+
+	// peer 2 should have noticed the collision and resolved it
+	require.NotEqual(t, oldShortID, peers2.ourself.ShortID)
+
+	// The Peers do not have a Router, so broadcastPeerUpdate does
+	// nothing in the context of this test.  So we fake what it
+	// would do.
+	updated[PeerName(2)] = void
+
+	// the update from peer 2 should include its short id change
+	peers3.ApplyUpdate(peers2.EncodePeers(updated))
+	require.Equal(t, peers2.ourself.ShortID,
+		peers3.Fetch(PeerName(2)).ShortID)
+}
+
+// Test the case where all short ids are taken, but then some peers go
+// away, so the local peer reassigns
+func TestDeferredShortIDReassignment(t *testing.T) {
+	rng := rand.New(rand.NewSource(time.Now().UnixNano()))
+	_, us := newNode(PeerName(1 << PeerShortIDBits))
+
+	// Connect us to other peers occupying all short ids
+	others := make([]*Peers, 1<<PeerShortIDBits)
+	var pending PeersPendingNotifications
+	for i := range others {
+		_, others[i] = newNode(PeerName(i))
+		others[i].setLocalShortID(PeerShortID(i), &pending)
+		us.AddTestConnection(others[i].ourself.Peer)
+	}
+
+	// Check that, as expected, the local peer does not own its
+	// short id
+	require.NotEqual(t, us.ourself.Peer,
+		us.byShortID[us.ourself.ShortID].peer)
+
+	// Disconnect one peer, and we should now be able to claim its
+	// short id
+	other := others[rng.Intn(1<<PeerShortIDBits)]
+	us.DeleteTestConnection(other.ourself.Peer)
+	us.GarbageCollect()
+
+	require.Equal(t, us.ourself.Peer, us.byShortID[us.ourself.ShortID].peer)
+}

mesh/protocol.go

@@ -0,0 +1,348 @@
+package mesh
+
+import (
+	"bytes"
+	"encoding/gob"
+	"encoding/hex"
+	"fmt"
+	"io"
+	"time"
+)
+
+const (
+	Protocol           = "weave"
+	ProtocolMinVersion = 1
+	ProtocolMaxVersion = 2
+)
+
+var (
+	ProtocolBytes = []byte(Protocol)
+
+	HeaderTimeout = 10 * time.Second
+
+	ProtocolV1Features = []string{
+		"ConnID",
+		"Name",
+		"NickName",
+		"PeerNameFlavour",
+		"UID",
+	}
+
+	ErrExpectedCrypto   = fmt.Errorf("Password specified, but peer requested an unencrypted connection")
+	ErrExpectedNoCrypto = fmt.Errorf("No password specificed, but peer requested an encrypted connection")
+)
+
+// We don't need the full net.TCPConn to do the protocol intro.  This
+// interface contains just the parts we do need, to support testing
+type ProtocolIntroConn interface {
+	// io.Reader
+	Read(b []byte) (n int, err error)
+
+	// io.Writer
+	Write(b []byte) (n int, err error)
+
+	// net.Conn's deadline methods
+	SetDeadline(t time.Time) error
+	SetReadDeadline(t time.Time) error
+	SetWriteDeadline(t time.Time) error
+}
+
+type ProtocolIntroParams struct {
+	MinVersion byte
+	MaxVersion byte
+	Features   map[string]string
+	Conn       ProtocolIntroConn
+	Password   []byte
+	Outbound   bool
+}
+
+type ProtocolIntroResults struct {
+	Features   map[string]string
+	Receiver   TCPReceiver
+	Sender     TCPSender
+	SessionKey *[32]byte
+	Version    byte
+}
+
+func (params ProtocolIntroParams) DoIntro() (res ProtocolIntroResults, err error) {
+	if err = params.Conn.SetDeadline(time.Now().Add(HeaderTimeout)); err != nil {
+		return
+	}
+
+	if res.Version, err = params.exchangeProtocolHeader(); err != nil {
+		return
+	}
+
+	var pubKey, privKey *[32]byte
+	if params.Password != nil {
+		if pubKey, privKey, err = GenerateKeyPair(); err != nil {
+			return
+		}
+	}
+
+	if err = params.Conn.SetWriteDeadline(time.Time{}); err != nil {
+		return
+	}
+	if err = params.Conn.SetReadDeadline(time.Now().Add(TCPHeartbeat * 2)); err != nil {
+		return
+	}
+
+	switch res.Version {
+	case 1:
+		err = res.doIntroV1(params, pubKey, privKey)
+	case 2:
+		err = res.doIntroV2(params, pubKey, privKey)
+	default:
+		panic("unhandled protocol version")
+	}
+
+	return
+}
+
+func (params ProtocolIntroParams) exchangeProtocolHeader() (byte, error) {
+	// Write in a separate goroutine to avoid the possibility of
+	// deadlock.  The result channel is of size 1 so that the
+	// goroutine does not linger even if we encounter an error on
+	// the read side.
+	sendHeader := append(ProtocolBytes, params.MinVersion, params.MaxVersion)
+	writeDone := make(chan error, 1)
+	go func() {
+		_, err := params.Conn.Write(sendHeader)
+		writeDone <- err
+	}()
+
+	header := make([]byte, len(ProtocolBytes)+2)
+	if n, err := io.ReadFull(params.Conn, header); err != nil && n == 0 {
+		return 0, fmt.Errorf("failed to receive remote protocol header: %s", err)
+	} else if err != nil {
+		return 0, fmt.Errorf("received incomplete remote protocol header (%d octets instead of %d): %v; error: %s",
+			n, len(header), header[:n], err)
+	}
+
+	if !bytes.Equal(ProtocolBytes, header[:len(ProtocolBytes)]) {
+		return 0, fmt.Errorf("remote protocol header not recognised: %v", header[:len(ProtocolBytes)])
+	}
+
+	theirMinVersion := header[len(ProtocolBytes)]
+	minVersion := theirMinVersion
+	if params.MinVersion > minVersion {
+		minVersion = params.MinVersion
+	}
+
+	theirMaxVersion := header[len(ProtocolBytes)+1]
+	maxVersion := theirMaxVersion
+	if maxVersion > params.MaxVersion {
+		maxVersion = params.MaxVersion
+	}
+
+	if minVersion > maxVersion {
+		return 0, fmt.Errorf("remote version range [%d,%d] is incompatible with ours [%d,%d]",
+			theirMinVersion, theirMaxVersion,
+			params.MinVersion, params.MaxVersion)
+	}
+
+	if err := <-writeDone; err != nil {
+		return 0, err
+	}
+
+	return maxVersion, nil
+}
+
+// The V1 procotol consists of the protocol identification/version
+// header, followed by a stream of gobified values.  The first value
+// is the encoded features map (never encrypted).  The subsequent
+// values are the messages on the connection (encrypted for an
+// encrypted connection).  For an encrypted connection, the public key
+// is passed in the "PublicKey" feature as a string of hex digits.
+func (res *ProtocolIntroResults) doIntroV1(params ProtocolIntroParams, pubKey, privKey *[32]byte) error {
+	features := filterV1Features(params.Features)
+	if pubKey != nil {
+		features["PublicKey"] = hex.EncodeToString(pubKey[:])
+	}
+
+	enc := gob.NewEncoder(params.Conn)
+	dec := gob.NewDecoder(params.Conn)
+
+	// Encode in a separate goroutine to avoid the possibility of
+	// deadlock.  The result channel is of size 1 so that the
+	// goroutine does not linger even if we encounter an error on
+	// the read side.
+	encodeDone := make(chan error, 1)
+	go func() {
+		encodeDone <- enc.Encode(features)
+	}()
+
+	if err := dec.Decode(&res.Features); err != nil {
+		return err
+	}
+
+	if err := <-encodeDone; err != nil {
+		return err
+	}
+
+	res.Sender = NewGobTCPSender(enc)
+	res.Receiver = NewGobTCPReceiver(dec)
+
+	if pubKey == nil {
+		if _, present := res.Features["PublicKey"]; present {
+			return ErrExpectedNoCrypto
+		}
+	} else {
+		remotePubKeyStr, ok := res.Features["PublicKey"]
+		if !ok {
+			return ErrExpectedCrypto
+		}
+
+		remotePubKey, err := hex.DecodeString(remotePubKeyStr)
+		if err != nil {
+			return err
+		}
+
+		res.setupCrypto(params, remotePubKey, privKey)
+	}
+
+	res.Features = filterV1Features(res.Features)
+	return nil
+}
+
+// In the V1 protocol, the intro fields are sent unencrypted.  So we
+// restrict them to an established subset of fields that are assumed
+// to be safe.
+func filterV1Features(intro map[string]string) map[string]string {
+	safe := make(map[string]string)
+	for _, k := range ProtocolV1Features {
+		if val, ok := intro[k]; ok {
+			safe[k] = val
+		}
+	}
+
+	return safe
+}
+
+// The V2 procotol consists of the protocol identification/version
+// header, followed by:
+//
+// - A single "encryption flag" byte: 0 for no encryption, 1 for
+// encryption.
+//
+// - When the connection is encrypted, 32 bytes follow containing the
+// public key.
+//
+// - Then a stream of length-prefixed messages, which are encrypted
+// for an encrypted connection.
+//
+// The first message contains the encoded features map (so in contrast
+// to V1, it will be encrypted on an encrypted connection).
+func (res *ProtocolIntroResults) doIntroV2(params ProtocolIntroParams, pubKey, privKey *[32]byte) error {
+	// Public key exchange
+	var wbuf []byte
+	if pubKey == nil {
+		wbuf = []byte{0}
+	} else {
+		wbuf = make([]byte, 1+len(*pubKey))
+		wbuf[0] = 1
+		copy(wbuf[1:], (*pubKey)[:])
+	}
+
+	// Write in a separate goroutine to avoid the possibility of
+	// deadlock.  The result channel is of size 1 so that the
+	// goroutine does not linger even if we encounter an error on
+	// the read side.
+	writeDone := make(chan error, 1)
+	go func() {
+		_, err := params.Conn.Write(wbuf)
+		writeDone <- err
+	}()
+
+	rbuf := make([]byte, 1)
+	if _, err := io.ReadFull(params.Conn, rbuf); err != nil {
+		return err
+	}
+
+	switch rbuf[0] {
+	case 0:
+		if pubKey != nil {
+			return ErrExpectedCrypto
+		}
+
+		res.Sender = NewLengthPrefixTCPSender(params.Conn)
+		res.Receiver = NewLengthPrefixTCPReceiver(params.Conn)
+
+	case 1:
+		if pubKey == nil {
+			return ErrExpectedNoCrypto
+		}
+
+		rbuf = make([]byte, len(pubKey))
+		if _, err := io.ReadFull(params.Conn, rbuf); err != nil {
+			return err
+		}
+
+		res.Sender = NewLengthPrefixTCPSender(params.Conn)
+		res.Receiver = NewLengthPrefixTCPReceiver(params.Conn)
+		res.setupCrypto(params, rbuf, privKey)
+
+	default:
+		return fmt.Errorf("Bad encryption flag %d", rbuf[0])
+	}
+
+	if err := <-writeDone; err != nil {
+		return err
+	}
+
+	// Features exchange
+	go func() {
+		buf := new(bytes.Buffer)
+		if err := gob.NewEncoder(buf).Encode(&params.Features); err != nil {
+			writeDone <- err
+			return
+		}
+
+		writeDone <- res.Sender.Send(buf.Bytes())
+	}()
+
+	rbuf, err := res.Receiver.Receive()
+	if err != nil {
+		return err
+	}
+
+	if err := gob.NewDecoder(bytes.NewReader(rbuf)).Decode(&res.Features); err != nil {
+		return err
+	}
+
+	if err := <-writeDone; err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (res *ProtocolIntroResults) setupCrypto(params ProtocolIntroParams, remotePubKey []byte, privKey *[32]byte) {
+	var remotePubKeyArr [32]byte
+	copy(remotePubKeyArr[:], remotePubKey)
+	res.SessionKey = FormSessionKey(&remotePubKeyArr, privKey, params.Password)
+	res.Sender = NewEncryptedTCPSender(res.Sender, res.SessionKey, params.Outbound)
+	res.Receiver = NewEncryptedTCPReceiver(res.Receiver, res.SessionKey, params.Outbound)
+}
+
+type ProtocolTag byte
+
+const (
+	ProtocolHeartbeat = iota
+	ProtocolConnectionEstablished
+	ProtocolFragmentationReceived
+	ProtocolPMTUVerified
+	ProtocolGossip
+	ProtocolGossipUnicast
+	ProtocolGossipBroadcast
+	ProtocolOverlayControlMsg
+)
+
+type ProtocolMsg struct {
+	tag ProtocolTag
+	msg []byte
+}
+
+type ProtocolSender interface {
+	SendProtocolMsg(m ProtocolMsg)
+}

mesh/protocol_crypto.go

@@ -0,0 +1,180 @@
+package mesh
+
+import (
+	"crypto/rand"
+	"crypto/sha256"
+	"encoding/binary"
+	"encoding/gob"
+	"fmt"
+	"io"
+	"sync"
+
+	"golang.org/x/crypto/nacl/box"
+	"golang.org/x/crypto/nacl/secretbox"
+)
+
+const MaxTCPMsgSize = 10 * 1024 * 1024
+
+func GenerateKeyPair() (publicKey, privateKey *[32]byte, err error) {
+	return box.GenerateKey(rand.Reader)
+}
+
+func FormSessionKey(remotePublicKey, localPrivateKey *[32]byte, secretKey []byte) *[32]byte {
+	var sharedKey [32]byte
+	box.Precompute(&sharedKey, remotePublicKey, localPrivateKey)
+	sharedKeySlice := sharedKey[:]
+	sharedKeySlice = append(sharedKeySlice, secretKey...)
+	sessionKey := sha256.Sum256(sharedKeySlice)
+	return &sessionKey
+}
+
+// TCP Senders/Receivers
+
+// The lowest 64 bits of the nonce contain the message sequence
+// number. The top most bit indicates the connection polarity at the
+// sender - '1' for outbound; the next indicates protocol type - '1'
+// for TCP. The remaining 126 bits are zero. The polarity is needed so
+// that the two ends of a connection do not use the same nonces; the
+// protocol type so that the TCP and UDP sender nonces are disjoint.
+// This is a requirement of the NaCl Security Model; see
+// http://nacl.cr.yp.to/box.html.
+type TCPCryptoState struct {
+	sessionKey *[32]byte
+	nonce      [24]byte
+	seqNo      uint64
+}
+
+func NewTCPCryptoState(sessionKey *[32]byte, outbound bool) *TCPCryptoState {
+	s := &TCPCryptoState{sessionKey: sessionKey}
+	if outbound {
+		s.nonce[0] |= (1 << 7)
+	}
+	// Ensure that TCP and UDP sequences are disjoint
+	s.nonce[0] |= (1 << 6)
+	return s
+}
+
+func (s *TCPCryptoState) advance() {
+	s.seqNo++
+	binary.BigEndian.PutUint64(s.nonce[16:24], s.seqNo)
+}
+
+type TCPSender interface {
+	Send([]byte) error
+}
+
+type GobTCPSender struct {
+	encoder *gob.Encoder
+}
+
+type LengthPrefixTCPSender struct {
+	writer io.Writer
+}
+
+type EncryptedTCPSender struct {
+	sync.RWMutex
+	sender TCPSender
+	state  *TCPCryptoState
+}
+
+func NewGobTCPSender(encoder *gob.Encoder) *GobTCPSender {
+	return &GobTCPSender{encoder: encoder}
+}
+
+func (sender *GobTCPSender) Send(msg []byte) error {
+	return sender.encoder.Encode(msg)
+}
+
+func NewLengthPrefixTCPSender(writer io.Writer) *LengthPrefixTCPSender {
+	return &LengthPrefixTCPSender{writer: writer}
+}
+
+func (sender *LengthPrefixTCPSender) Send(msg []byte) error {
+	l := len(msg)
+	if l > MaxTCPMsgSize {
+		return fmt.Errorf("outgoing message exceeds maximum size: %d > %d", l, MaxTCPMsgSize)
+	}
+	// We copy the message so we can send it in a single Write
+	// operation, thus making this thread-safe without locking.
+	prefixedMsg := make([]byte, 4+l)
+	binary.BigEndian.PutUint32(prefixedMsg, uint32(l))
+	copy(prefixedMsg[4:], msg)
+	_, err := sender.writer.Write(prefixedMsg)
+	return err
+}
+
+func NewEncryptedTCPSender(sender TCPSender, sessionKey *[32]byte, outbound bool) *EncryptedTCPSender {
+	return &EncryptedTCPSender{sender: sender, state: NewTCPCryptoState(sessionKey, outbound)}
+}
+
+func (sender *EncryptedTCPSender) Send(msg []byte) error {
+	sender.Lock()
+	defer sender.Unlock()
+	encodedMsg := secretbox.Seal(nil, msg, &sender.state.nonce, sender.state.sessionKey)
+	sender.state.advance()
+	return sender.sender.Send(encodedMsg)
+}
+
+type TCPReceiver interface {
+	Receive() ([]byte, error)
+}
+
+type GobTCPReceiver struct {
+	decoder *gob.Decoder
+}
+
+type LengthPrefixTCPReceiver struct {
+	reader io.Reader
+}
+
+type EncryptedTCPReceiver struct {
+	receiver TCPReceiver
+	state    *TCPCryptoState
+}
+
+func NewGobTCPReceiver(decoder *gob.Decoder) *GobTCPReceiver {
+	return &GobTCPReceiver{decoder: decoder}
+}
+
+func (receiver *GobTCPReceiver) Receive() ([]byte, error) {
+	var msg []byte
+	err := receiver.decoder.Decode(&msg)
+	return msg, err
+}
+
+func NewLengthPrefixTCPReceiver(reader io.Reader) *LengthPrefixTCPReceiver {
+	return &LengthPrefixTCPReceiver{reader: reader}
+}
+
+func (receiver *LengthPrefixTCPReceiver) Receive() ([]byte, error) {
+	lenPrefix := make([]byte, 4)
+	if _, err := io.ReadFull(receiver.reader, lenPrefix); err != nil {
+		return nil, err
+	}
+	l := binary.BigEndian.Uint32(lenPrefix)
+	if l > MaxTCPMsgSize {
+		return nil, fmt.Errorf("incoming message exceeds maximum size: %d > %d", l, MaxTCPMsgSize)
+	}
+	msg := make([]byte, l)
+	_, err := io.ReadFull(receiver.reader, msg)
+	return msg, err
+}
+
+func NewEncryptedTCPReceiver(receiver TCPReceiver, sessionKey *[32]byte, outbound bool) *EncryptedTCPReceiver {
+	return &EncryptedTCPReceiver{receiver: receiver, state: NewTCPCryptoState(sessionKey, !outbound)}
+}
+
+func (receiver *EncryptedTCPReceiver) Receive() ([]byte, error) {
+	msg, err := receiver.receiver.Receive()
+	if err != nil {
+		return nil, err
+	}
+
+	decodedMsg, success := secretbox.Open(nil, msg, &receiver.state.nonce, receiver.state.sessionKey)
+	if !success {
+		return nil, fmt.Errorf("Unable to decrypt TCP msg")
+	}
+
+	receiver.state.advance()
+	return decodedMsg, nil
+}

mesh/protocol_test.go

@@ -0,0 +1,96 @@
+package mesh
+
+import (
+	"io"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+type testConn struct {
+	io.Writer
+	io.Reader
+}
+
+func (testConn) SetDeadline(t time.Time) error {
+	return nil
+}
+
+func (testConn) SetReadDeadline(t time.Time) error {
+	return nil
+}
+
+func (testConn) SetWriteDeadline(t time.Time) error {
+	return nil
+}
+
+func connPair() (ProtocolIntroConn, ProtocolIntroConn) {
+	a := testConn{}
+	b := testConn{}
+	a.Reader, b.Writer = io.Pipe()
+	b.Reader, a.Writer = io.Pipe()
+	return &a, &b
+}
+
+func doIntro(t *testing.T, params ProtocolIntroParams) <-chan ProtocolIntroResults {
+	ch := make(chan ProtocolIntroResults, 1)
+	go func() {
+		res, err := params.DoIntro()
+		require.Nil(t, err)
+		ch <- res
+	}()
+	return ch
+}
+
+func doProtocolIntro(t *testing.T, aver, bver byte, password []byte) byte {
+	aconn, bconn := connPair()
+	aresch := doIntro(t, ProtocolIntroParams{
+		MinVersion: ProtocolMinVersion,
+		MaxVersion: aver,
+		Features:   map[string]string{"Name": "A"},
+		Conn:       aconn,
+		Outbound:   true,
+		Password:   password,
+	})
+	bresch := doIntro(t, ProtocolIntroParams{
+		MinVersion: ProtocolMinVersion,
+		MaxVersion: bver,
+		Features:   map[string]string{"Name": "B"},
+		Conn:       bconn,
+		Outbound:   false,
+		Password:   password,
+	})
+	ares := <-aresch
+	bres := <-bresch
+
+	// Check that features were conveyed
+	require.Equal(t, "B", ares.Features["Name"])
+	require.Equal(t, "A", bres.Features["Name"])
+
+	// Check that Senders and Receivers work
+	go func() {
+		require.Nil(t, ares.Sender.Send([]byte("Hello from A")))
+		require.Nil(t, bres.Sender.Send([]byte("Hello from B")))
+	}()
+
+	data, err := bres.Receiver.Receive()
+	require.Nil(t, err)
+	require.Equal(t, "Hello from A", string(data))
+
+	data, err = ares.Receiver.Receive()
+	require.Nil(t, err)
+	require.Equal(t, "Hello from B", string(data))
+
+	require.Equal(t, ares.Version, bres.Version)
+	return ares.Version
+}
+
+func TestProtocolIntro(t *testing.T) {
+	require.Equal(t, 2, int(doProtocolIntro(t, 2, 2, nil)))
+	require.Equal(t, 2, int(doProtocolIntro(t, 2, 2, []byte("sekr1t"))))
+	require.Equal(t, 1, int(doProtocolIntro(t, 1, 2, nil)))
+	require.Equal(t, 1, int(doProtocolIntro(t, 1, 2, []byte("pa55"))))
+	require.Equal(t, 1, int(doProtocolIntro(t, 2, 1, nil)))
+	require.Equal(t, 1, int(doProtocolIntro(t, 2, 1, []byte("w0rd"))))
+}

mesh/router.go

@@ -0,0 +1,192 @@
+package mesh
+
+import (
+	"fmt"
+	"math"
+	"net"
+	"sync"
+	"time"
+
+	"github.com/weaveworks/weave/common"
+)
+
+const (
+	Port           = 6783
+	ChannelSize    = 16
+	TCPHeartbeat   = 30 * time.Second
+	GossipInterval = 30 * time.Second
+	MaxDuration    = time.Duration(math.MaxInt64)
+
+	acceptMaxTokens  = 100                    // [1]
+	acceptTokenDelay = 100 * time.Millisecond // [2]
+)
+
+// [1] capacity of token bucket for rate limiting accepts
+
+// [2] control rate at which new tokens are added to the bucket
+
+var (
+	log        = common.Log
+	checkFatal = common.CheckFatal
+	checkWarn  = common.CheckWarn
+)
+
+type Config struct {
+	Port               int
+	ProtocolMinVersion byte
+	Password           []byte
+	ConnLimit          int
+	PeerDiscovery      bool
+}
+
+type Router struct {
+	Config
+	Overlay         Overlay
+	Ourself         *LocalPeer
+	Peers           *Peers
+	Routes          *Routes
+	ConnectionMaker *ConnectionMaker
+	gossipLock      sync.RWMutex
+	gossipChannels  GossipChannels
+	TopologyGossip  Gossip
+	acceptLimiter   *TokenBucket
+}
+
+func NewRouter(config Config, name PeerName, nickName string, overlay Overlay) *Router {
+	router := &Router{Config: config, gossipChannels: make(GossipChannels)}
+
+	if overlay == nil {
+		overlay = NullOverlay{}
+	}
+
+	router.Overlay = overlay
+	router.Ourself = NewLocalPeer(name, nickName, router)
+	router.Peers = NewPeers(router.Ourself)
+	router.Peers.OnGC(func(peer *Peer) {
+		log.Println("Removed unreachable peer", peer)
+	})
+	router.Routes = NewRoutes(router.Ourself, router.Peers)
+	router.ConnectionMaker = NewConnectionMaker(router.Ourself, router.Peers, router.Port, router.PeerDiscovery)
+	router.TopologyGossip = router.NewGossip("topology", router)
+	router.acceptLimiter = NewTokenBucket(acceptMaxTokens, acceptTokenDelay)
+
+	return router
+}
+
+// Start listening for TCP connections. This is separate from
+// NewRouter so that gossipers can register before we start forming
+// connections.
+func (router *Router) Start() {
+	router.listenTCP(router.Port)
+}
+
+func (router *Router) Stop() error {
+	// TODO: perform graceful shutdown...
+	return nil
+}
+
+func (router *Router) UsingPassword() bool {
+	return router.Password != nil
+}
+
+func (router *Router) listenTCP(localPort int) {
+	localAddr, err := net.ResolveTCPAddr("tcp4", fmt.Sprint(":", localPort))
+	checkFatal(err)
+	ln, err := net.ListenTCP("tcp4", localAddr)
+	checkFatal(err)
+	go func() {
+		defer ln.Close()
+		for {
+			tcpConn, err := ln.AcceptTCP()
+			if err != nil {
+				log.Errorln(err)
+				continue
+			}
+			router.acceptTCP(tcpConn)
+			router.acceptLimiter.Wait()
+		}
+	}()
+}
+
+func (router *Router) acceptTCP(tcpConn *net.TCPConn) {
+	// someone else is dialing us, so our udp sender is the conn
+	// on router.Port and we wait for them to send us something on UDP to
+	// start.
+	remoteAddrStr := tcpConn.RemoteAddr().String()
+	log.Printf("->[%s] connection accepted", remoteAddrStr)
+	connRemote := NewRemoteConnection(router.Ourself.Peer, nil, remoteAddrStr, false, false)
+	StartLocalConnection(connRemote, tcpConn, router, true)
+}
+
+// Gossiper methods - the Router is the topology Gossiper
+
+type TopologyGossipData struct {
+	peers  *Peers
+	update PeerNameSet
+}
+
+func (router *Router) BroadcastTopologyUpdate(update []*Peer) {
+	names := make(PeerNameSet)
+	for _, p := range update {
+		names[p.Name] = void
+	}
+
+	router.TopologyGossip.GossipBroadcast(&TopologyGossipData{
+		peers:  router.Peers,
+		update: names,
+	})
+}
+
+func (d *TopologyGossipData) Merge(other GossipData) {
+	for name := range other.(*TopologyGossipData).update {
+		d.update[name] = void
+	}
+}
+
+func (d *TopologyGossipData) Encode() [][]byte {
+	return [][]byte{d.peers.EncodePeers(d.update)}
+}
+
+func (router *Router) OnGossipUnicast(sender PeerName, msg []byte) error {
+	return fmt.Errorf("unexpected topology gossip unicast: %v", msg)
+}
+
+func (router *Router) OnGossipBroadcast(_ PeerName, update []byte) (GossipData, error) {
+	origUpdate, _, err := router.applyTopologyUpdate(update)
+	if err != nil || len(origUpdate) == 0 {
+		return nil, err
+	}
+	return &TopologyGossipData{peers: router.Peers, update: origUpdate}, nil
+}
+
+func (router *Router) Gossip() GossipData {
+	return &TopologyGossipData{peers: router.Peers, update: router.Peers.Names()}
+}
+
+func (router *Router) OnGossip(update []byte) (GossipData, error) {
+	_, newUpdate, err := router.applyTopologyUpdate(update)
+	if err != nil || len(newUpdate) == 0 {
+		return nil, err
+	}
+	return &TopologyGossipData{peers: router.Peers, update: newUpdate}, nil
+}
+
+func (router *Router) applyTopologyUpdate(update []byte) (PeerNameSet, PeerNameSet, error) {
+	origUpdate, newUpdate, err := router.Peers.ApplyUpdate(update)
+	if _, ok := err.(UnknownPeerError); err != nil && ok {
+		// That update contained a reference to a peer which wasn't
+		// itself included in the update, and we didn't know about
+		// already. We ignore this; eventually we should receive an
+		// update containing a complete topology.
+		log.Println("Topology gossip:", err)
+		return nil, nil, nil
+	}
+	if err != nil {
+		return nil, nil, err
+	}
+	if len(newUpdate) > 0 {
+		router.ConnectionMaker.Refresh()
+		router.Routes.Recalculate()
+	}
+	return origUpdate, newUpdate, nil
+}

mesh/routes.go

@@ -0,0 +1,272 @@
+package mesh
+
+import (
+	"math"
+	"sync"
+)
+
+type unicastRoutes map[PeerName]PeerName
+type broadcastRoutes map[PeerName][]PeerName
+
+type Routes struct {
+	sync.RWMutex
+	ourself      *LocalPeer
+	peers        *Peers
+	onChange     []func()
+	unicast      unicastRoutes
+	unicastAll   unicastRoutes // [1]
+	broadcast    broadcastRoutes
+	broadcastAll broadcastRoutes // [1]
+	recalculate  chan<- *struct{}
+	wait         chan<- chan struct{}
+	// [1] based on *all* connections, not just established &
+	// symmetric ones
+}
+
+func NewRoutes(ourself *LocalPeer, peers *Peers) *Routes {
+	recalculate := make(chan *struct{}, 1)
+	wait := make(chan chan struct{})
+	routes := &Routes{
+		ourself:      ourself,
+		peers:        peers,
+		unicast:      make(unicastRoutes),
+		unicastAll:   make(unicastRoutes),
+		broadcast:    make(broadcastRoutes),
+		broadcastAll: make(broadcastRoutes),
+		recalculate:  recalculate,
+		wait:         wait}
+	routes.unicast[ourself.Name] = UnknownPeerName
+	routes.unicastAll[ourself.Name] = UnknownPeerName
+	routes.broadcast[ourself.Name] = []PeerName{}
+	routes.broadcastAll[ourself.Name] = []PeerName{}
+	go routes.run(recalculate, wait)
+	return routes
+}
+
+func (routes *Routes) OnChange(callback func()) {
+	routes.Lock()
+	defer routes.Unlock()
+	routes.onChange = append(routes.onChange, callback)
+}
+
+func (routes *Routes) PeerNames() PeerNameSet {
+	return routes.peers.Names()
+}
+
+func (routes *Routes) Unicast(name PeerName) (PeerName, bool) {
+	routes.RLock()
+	defer routes.RUnlock()
+	hop, found := routes.unicast[name]
+	return hop, found
+}
+
+func (routes *Routes) UnicastAll(name PeerName) (PeerName, bool) {
+	routes.RLock()
+	defer routes.RUnlock()
+	hop, found := routes.unicastAll[name]
+	return hop, found
+}
+
+func (routes *Routes) Broadcast(name PeerName) []PeerName {
+	routes.RLock()
+	defer routes.RUnlock()
+	return routes.broadcast[name]
+}
+
+func (routes *Routes) BroadcastAll(name PeerName) []PeerName {
+	routes.RLock()
+	defer routes.RUnlock()
+	return routes.broadcastAll[name]
+}
+
+// Choose min(log2(n_peers), n_neighbouring_peers) neighbours, with a
+// random distribution that is topology-sensitive, favouring
+// neighbours at the end of "bottleneck links". We determine the
+// latter based on the unicast routing table. If a neighbour appears
+// as the value more frequently than others - meaning that we reach a
+// higher proportion of peers via that neighbour than other neighbours
+// - then it is chosen with a higher probability.
+//
+// Note that we choose log2(n_peers) *neighbours*, not
+// peers. Consequently, on sparsely connected peers this function
+// returns a higher proportion of neighbours than elsewhere. In
+// extremis, on peers with fewer than log2(n_peers) neighbours, all
+// neighbours are returned.
+func (routes *Routes) RandomNeighbours(except PeerName) PeerNameSet {
+	res := make(PeerNameSet)
+	routes.RLock()
+	defer routes.RUnlock()
+	count := int(math.Log2(float64(len(routes.unicastAll))))
+	// depends on go's random map iteration
+	for _, dst := range routes.unicastAll {
+		if dst != UnknownPeerName && dst != except {
+			res[dst] = void
+			if len(res) >= count {
+				break
+			}
+		}
+	}
+	return res
+}
+
+// Request recalculation of the routing table. This is async but can
+// effectively be made synchronous with a subsequent call to
+// EnsureRecalculated.
+func (routes *Routes) Recalculate() {
+	// The use of a 1-capacity channel in combination with the
+	// non-blocking send is an optimisation that results in multiple
+	// requests being coalesced.
+	select {
+	case routes.recalculate <- nil:
+	default:
+	}
+}
+
+// Wait for any preceding Recalculate requests to be processed.
+func (routes *Routes) EnsureRecalculated() {
+	done := make(chan struct{})
+	routes.wait <- done
+	<-done
+}
+
+func (routes *Routes) run(recalculate <-chan *struct{}, wait <-chan chan struct{}) {
+	for {
+		select {
+		case <-recalculate:
+			routes.calculate()
+		case done := <-wait:
+			select {
+			case <-recalculate:
+				routes.calculate()
+			default:
+			}
+			close(done)
+		}
+	}
+}
+
+func (routes *Routes) calculate() {
+	routes.peers.RLock()
+	routes.ourself.RLock()
+	var (
+		oldUnicast   = routes.unicast
+		oldBroadcast = routes.broadcast
+		unicast      = routes.calculateUnicast(true)
+		unicastAll   = routes.calculateUnicast(false)
+		broadcast    = routes.calculateBroadcast(true)
+		broadcastAll = routes.calculateBroadcast(false)
+	)
+	routes.ourself.RUnlock()
+	routes.peers.RUnlock()
+
+	routes.Lock()
+	routes.unicast = unicast
+	routes.unicastAll = unicastAll
+	routes.broadcast = broadcast
+	routes.broadcastAll = broadcastAll
+	onChange := routes.onChange
+	routes.Unlock()
+
+	if !unicast.equals(oldUnicast) || !broadcast.equals(oldBroadcast) {
+		for _, callback := range onChange {
+			callback()
+		}
+	}
+}
+
+// Calculate all the routes for the question: if *we* want to send a
+// packet to Peer X, what is the next hop?
+//
+// When we sniff a packet, we determine the destination peer
+// ourself. Consequently, we can relay the packet via any
+// arbitrary peers - the intermediate peers do not have to have
+// any knowledge of the MAC address at all. Thus there's no need
+// to exchange knowledge of MAC addresses, nor any constraints on
+// the routes that we construct.
+func (routes *Routes) calculateUnicast(establishedAndSymmetric bool) unicastRoutes {
+	_, unicast := routes.ourself.Routes(nil, establishedAndSymmetric)
+	return unicast
+}
+
+// Calculate all the routes for the question: if we receive a
+// broadcast originally from Peer X, which peers should we pass the
+// frames on to?
+//
+// When the topology is stable, and thus all peers perform route
+// calculations based on the same data, the algorithm ensures that
+// broadcasts reach every peer exactly once.
+//
+// This is largely due to properties of the Peer.Routes algorithm. In
+// particular:
+//
+// ForAll X,Y,Z in Peers.
+//     X.Routes(Y) <= X.Routes(Z) \/
+//     X.Routes(Z) <= X.Routes(Y)
+// ForAll X,Y,Z in Peers.
+//     Y =/= Z /\ X.Routes(Y) <= X.Routes(Z) =>
+//     X.Routes(Y) u [P | Y.HasSymmetricConnectionTo(P)] <= X.Routes(Z)
+// where <= is the subset relationship on keys of the returned map.
+func (routes *Routes) calculateBroadcast(establishedAndSymmetric bool) broadcastRoutes {
+	broadcast := make(broadcastRoutes)
+	for _, peer := range routes.peers.byName {
+		hops := []PeerName{}
+		if found, reached := peer.Routes(routes.ourself.Peer, establishedAndSymmetric); found {
+			routes.ourself.ForEachConnectedPeer(establishedAndSymmetric, reached,
+				func(remotePeer *Peer) { hops = append(hops, remotePeer.Name) })
+		}
+		broadcast[peer.Name] = hops
+	}
+	return broadcast
+}
+
+func (a unicastRoutes) equals(b unicastRoutes) bool {
+	for key, aval := range a {
+		if bval, ok := b[key]; !ok || bval != aval {
+			return false
+		}
+	}
+
+	for key := range b {
+		if _, ok := a[key]; !ok {
+			return false
+		}
+	}
+
+	return true
+}
+
+func (a broadcastRoutes) equals(b broadcastRoutes) bool {
+	set := make(map[PeerName]struct{})
+
+	for key, aval := range a {
+		bval, ok := b[key]
+		if !ok {
+			return false
+		}
+
+		for _, peer := range aval {
+			set[peer] = struct{}{}
+		}
+
+		for _, peer := range bval {
+			if _, ok := set[peer]; !ok {
+				return false
+			}
+
+			delete(set, peer)
+		}
+
+		if len(set) != 0 {
+			return false
+		}
+
+	}
+
+	for key := range b {
+		if _, ok := a[key]; !ok {
+			return false
+		}
+	}
+
+	return true
+}

mesh/status.go

@@ -0,0 +1,193 @@
+package mesh
+
+import (
+	"fmt"
+)
+
+type Status struct {
+	Protocol           string
+	ProtocolMinVersion int
+	ProtocolMaxVersion int
+	Encryption         bool
+	PeerDiscovery      bool
+	Name               string
+	NickName           string
+	Port               int
+	Peers              []PeerStatus
+	UnicastRoutes      []UnicastRouteStatus
+	BroadcastRoutes    []BroadcastRouteStatus
+	Connections        []LocalConnectionStatus
+	Targets            []string
+	OverlayDiagnostics interface{}
+}
+
+type PeerStatus struct {
+	Name        string
+	NickName    string
+	UID         PeerUID
+	ShortID     PeerShortID
+	Version     uint64
+	Connections []ConnectionStatus
+}
+
+type ConnectionStatus struct {
+	Name        string
+	NickName    string
+	Address     string
+	Outbound    bool
+	Established bool
+}
+
+type UnicastRouteStatus struct {
+	Dest, Via string
+}
+
+type BroadcastRouteStatus struct {
+	Source string
+	Via    []string
+}
+
+type LocalConnectionStatus struct {
+	Address  string
+	Outbound bool
+	State    string
+	Info     string
+}
+
+func NewStatus(router *Router) *Status {
+	return &Status{
+		Protocol,
+		ProtocolMinVersion,
+		ProtocolMaxVersion,
+		router.UsingPassword(),
+		router.PeerDiscovery,
+		router.Ourself.Name.String(),
+		router.Ourself.NickName,
+		router.Port,
+		NewPeerStatusSlice(router.Peers),
+		NewUnicastRouteStatusSlice(router.Routes),
+		NewBroadcastRouteStatusSlice(router.Routes),
+		NewLocalConnectionStatusSlice(router.ConnectionMaker),
+		NewTargetSlice(router.ConnectionMaker),
+		router.Overlay.Diagnostics()}
+}
+
+func NewPeerStatusSlice(peers *Peers) []PeerStatus {
+	var slice []PeerStatus
+
+	peers.ForEach(func(peer *Peer) {
+		var connections []ConnectionStatus
+		if peer == peers.ourself.Peer {
+			for conn := range peers.ourself.Connections() {
+				connections = append(connections, newConnectionStatus(conn))
+			}
+		} else {
+			// Modifying peer.connections requires a write lock on
+			// Peers, and since we are holding a read lock (due to the
+			// ForEach), access without locking the peer is safe.
+			for _, conn := range peer.connections {
+				connections = append(connections, newConnectionStatus(conn))
+			}
+		}
+		slice = append(slice, PeerStatus{
+			peer.Name.String(),
+			peer.NickName,
+			peer.UID,
+			peer.ShortID,
+			peer.Version,
+			connections})
+	})
+
+	return slice
+}
+
+func newConnectionStatus(c Connection) ConnectionStatus {
+	return ConnectionStatus{
+		c.Remote().Name.String(),
+		c.Remote().NickName,
+		c.RemoteTCPAddr(),
+		c.Outbound(),
+		c.Established()}
+}
+
+func NewUnicastRouteStatusSlice(routes *Routes) []UnicastRouteStatus {
+	routes.RLock()
+	defer routes.RUnlock()
+
+	var slice []UnicastRouteStatus
+	for dest, via := range routes.unicast {
+		slice = append(slice, UnicastRouteStatus{dest.String(), via.String()})
+	}
+	return slice
+}
+
+func NewBroadcastRouteStatusSlice(routes *Routes) []BroadcastRouteStatus {
+	routes.RLock()
+	defer routes.RUnlock()
+
+	var slice []BroadcastRouteStatus
+	for source, via := range routes.broadcast {
+		var hops []string
+		for _, hop := range via {
+			hops = append(hops, hop.String())
+		}
+		slice = append(slice, BroadcastRouteStatus{source.String(), hops})
+	}
+	return slice
+}
+
+func NewLocalConnectionStatusSlice(cm *ConnectionMaker) []LocalConnectionStatus {
+	resultChan := make(chan []LocalConnectionStatus, 0)
+	cm.actionChan <- func() bool {
+		var slice []LocalConnectionStatus
+		for conn := range cm.connections {
+			state := "pending"
+			if conn.Established() {
+				state = "established"
+			}
+			lc, _ := conn.(*LocalConnection)
+			info := fmt.Sprintf("%-6v %v", lc.OverlayConn.DisplayName(), conn.Remote())
+			slice = append(slice, LocalConnectionStatus{conn.RemoteTCPAddr(), conn.Outbound(), state, info})
+		}
+		for address, target := range cm.targets {
+			add := func(state, info string) {
+				slice = append(slice, LocalConnectionStatus{address, true, state, info})
+			}
+			switch target.state {
+			case TargetWaiting:
+				until := "never"
+				if !target.tryAfter.IsZero() {
+					until = target.tryAfter.String()
+				}
+				if target.lastError == nil { // shouldn't happen
+					add("waiting", "until: "+until)
+				} else {
+					add("failed", target.lastError.Error()+", retry: "+until)
+				}
+			case TargetAttempting:
+				if target.lastError == nil {
+					add("connecting", "")
+				} else {
+					add("retrying", target.lastError.Error())
+				}
+			case TargetConnected:
+			}
+		}
+		resultChan <- slice
+		return false
+	}
+	return <-resultChan
+}
+
+func NewTargetSlice(cm *ConnectionMaker) []string {
+	resultChan := make(chan []string, 0)
+	cm.actionChan <- func() bool {
+		var slice []string
+		for peer := range cm.directPeers {
+			slice = append(slice, peer)
+		}
+		resultChan <- slice
+		return false
+	}
+	return <-resultChan
+}

mesh/surrogate_gossiper.go

@@ -0,0 +1,41 @@
+package mesh
+
+type SurrogateGossipData struct {
+	messages [][]byte
+}
+
+func NewSurrogateGossipData(msg []byte) *SurrogateGossipData {
+	return &SurrogateGossipData{messages: [][]byte{msg}}
+}
+
+func (d *SurrogateGossipData) Encode() [][]byte {
+	return d.messages
+}
+
+func (d *SurrogateGossipData) Merge(other GossipData) {
+	d.messages = append(d.messages, other.(*SurrogateGossipData).messages...)
+}
+
+// SurrogateGossiper ignores unicasts and relays broadcasts and gossips.
+
+type SurrogateGossiper struct{}
+
+func (*SurrogateGossiper) OnGossipUnicast(sender PeerName, msg []byte) error {
+	return nil
+}
+
+func (*SurrogateGossiper) OnGossipBroadcast(_ PeerName, update []byte) (GossipData, error) {
+	return NewSurrogateGossipData(update), nil
+}
+
+func (*SurrogateGossiper) Gossip() GossipData {
+	return nil
+}
+
+func (*SurrogateGossiper) OnGossip(update []byte) (GossipData, error) {
+	return NewSurrogateGossipData(update), nil
+}
+
+var (
+	surrogateGossiper SurrogateGossiper
+)

mesh/token_bucket.go

@@ -0,0 +1,42 @@
+package mesh
+
+import (
+	"time"
+)
+
+type TokenBucket struct {
+	capacity             int64         // Maximum capacity of bucket
+	tokenInterval        time.Duration // Token replenishment rate
+	refillDuration       time.Duration // Time to refill from empty
+	earliestUnspentToken time.Time
+}
+
+func NewTokenBucket(capacity int64, tokenInterval time.Duration) *TokenBucket {
+	tb := TokenBucket{
+		capacity:       capacity,
+		tokenInterval:  tokenInterval,
+		refillDuration: tokenInterval * time.Duration(capacity)}
+
+	tb.earliestUnspentToken = tb.capacityToken()
+
+	return &tb
+}
+
+func (tb *TokenBucket) Wait() {
+	// If earliest unspent token is in the future, sleep until then
+	time.Sleep(tb.earliestUnspentToken.Sub(time.Now()))
+
+	// Alternatively, enforce bucket capacity if necessary
+	capacityToken := tb.capacityToken()
+	if tb.earliestUnspentToken.Before(capacityToken) {
+		tb.earliestUnspentToken = capacityToken
+	}
+
+	// 'Remove' a token from the bucket
+	tb.earliestUnspentToken = tb.earliestUnspentToken.Add(tb.tokenInterval)
+}
+
+// Determine the historic token timestamp representing a full bucket
+func (tb *TokenBucket) capacityToken() time.Time {
+	return time.Now().Add(-tb.refillDuration).Truncate(tb.tokenInterval)
+}