Version 0.4.7

Merge pull request #986 from yggdrasil-network/develop

CHANGELOG.md

@@ -26,6 +26,24 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 - in case of vulnerabilities.
 -->
 
+## [0.4.7] - 2022-11-20
+
+### Added
+
+- Dropped outbound peerings will now try to reconnect after a single second, rather than waiting up to 60 seconds for the normal peer timer
+
+### Changed
+
+- Session encryption keys are now rotated at most once per minute, which reduces CPU usage and improves throughput on fast low latency links
+- Buffers are now reused in the session encryption handler, which improves session throughput and reduces memory allocations
+- Buffers are now reused in the router for DHT and path traffic, which improves overall routing throughput and reduces memory allocations
+
+### Fixed
+
+- A bug in the admin socket where requests fail unless `arguments` is specified has been fixed
+- Certificates on TLS listeners will no longer expire after a year
+- The `-address` and `-subnet` command line options now return a useful warning when no configuration is specified
+
 ## [0.4.6] - 2022-10-25
 
 ### Added

cmd/yggdrasil/main.go

@@ -247,7 +247,12 @@ func run(args yggArgs, ctx context.Context) {
 		return
 	default:
 		// No flags were provided, therefore print the list of flags to stdout.
+		fmt.Println("Usage:")
 		flag.PrintDefaults()
+
+		if args.getaddr || args.getsnet {
+			fmt.Println("\nError: You need to specify some config data using -useconf or -useconffile.")
+		}
 	}
 	// Have we got a working configuration? If we don't then it probably means
 	// that neither -autoconf, -useconf or -useconffile were set above. Stop
@@ -339,7 +344,7 @@ func run(args yggArgs, ctx context.Context) {
 				Beacon:   intf.Beacon,
 				Listen:   intf.Listen,
 				Port:     intf.Port,
-				Priority: intf.Priority,
+				Priority: uint8(intf.Priority),
 			})
 		}
 		if n.multicast, err = multicast.New(n.core, logger, options...); err != nil {

contrib/mobile/mobile.go

@@ -87,7 +87,7 @@ func (m *Yggdrasil) StartJSON(configjson []byte) error {
 				Beacon:   intf.Beacon,
 				Listen:   intf.Listen,
 				Port:     intf.Port,
-				Priority: intf.Priority,
+				Priority: uint8(intf.Priority),
 			})
 		}
 		m.multicast, err = multicast.New(m.core, logger, options...)
@@ -115,6 +115,18 @@ func (m *Yggdrasil) Send(p []byte) error {
 	return nil
 }
 
+// Send sends a packet from given buffer to Yggdrasil. From first byte up to length.
+func (m *Yggdrasil) SendBuffer(p []byte, length int) error {
+	if m.iprwc == nil {
+		return nil
+	}
+	if len(p) < length {
+		return nil
+	}
+	_, _ = m.iprwc.Write(p[:length])
+	return nil
+}
+
 // Recv waits for and reads a packet coming from Yggdrasil. It
 // will be a fully formed IPv6 packet
 func (m *Yggdrasil) Recv() ([]byte, error) {
@@ -126,6 +138,15 @@ func (m *Yggdrasil) Recv() ([]byte, error) {
 	return buf[:n], nil
 }
 
+// Recv waits for and reads a packet coming from Yggdrasil to given buffer, returning size of packet
+func (m *Yggdrasil) RecvBuffer(buf []byte) (int, error) {
+	if m.iprwc == nil {
+		return 0, nil
+	}
+	n, _ := m.iprwc.Read(buf)
+	return n, nil
+}
+
 // Stop the mobile Yggdrasil instance
 func (m *Yggdrasil) Stop() error {
 	logger := log.New(m.log, "", 0)
@@ -138,6 +159,11 @@ func (m *Yggdrasil) Stop() error {
 	return nil
 }
 
+// Retry resets the peer connection timer and tries to dial them immediately.
+func (m *Yggdrasil) RetryPeersNow() {
+	m.core.RetryPeersNow()
+}
+
 // GenerateConfigJSON generates mobile-friendly configuration in JSON format
 func GenerateConfigJSON() []byte {
 	nc := defaults.GenerateConfig()

go.mod

@@ -3,7 +3,7 @@ module github.com/yggdrasil-network/yggdrasil-go
 go 1.17
 
 require (
-	github.com/Arceliar/ironwood v0.0.0-20221025225125-45b4281814c2
+	github.com/Arceliar/ironwood v0.0.0-20221115123222-ec61cea2f439
 	github.com/Arceliar/phony v0.0.0-20210209235338-dde1a8dca979
 	github.com/cheggaaa/pb/v3 v3.0.8
 	github.com/gologme/log v1.2.0
@@ -12,7 +12,7 @@ require (
 	github.com/kardianos/minwinsvc v1.0.2
 	github.com/mitchellh/mapstructure v1.4.1
 	github.com/vishvananda/netlink v1.1.0
-	golang.org/x/mobile v0.0.0-20221012134814-c746ac228303
+	golang.org/x/mobile v0.0.0-20221110043201-43a038452099
 	golang.org/x/net v0.0.0-20221014081412-f15817d10f9b
 	golang.org/x/sys v0.0.0-20221013171732-95e765b1cc43
 	golang.org/x/text v0.3.8

go.sum

@@ -1,5 +1,5 @@
-github.com/Arceliar/ironwood v0.0.0-20221025225125-45b4281814c2 h1:Usab30pNT2i/vZvpXcN9uOr5IO1RZPcUqoGH0DIAPnU=
+github.com/Arceliar/ironwood v0.0.0-20221115123222-ec61cea2f439 h1:eOW6/XIs06TnUn9GPCnfv71CQZw8edP3u3mH3lZt6iM=
-github.com/Arceliar/ironwood v0.0.0-20221025225125-45b4281814c2/go.mod h1:RP72rucOFm5udrnEzTmIWLRVGQiV/fSUAQXJ0RST/nk=
+github.com/Arceliar/ironwood v0.0.0-20221115123222-ec61cea2f439/go.mod h1:RP72rucOFm5udrnEzTmIWLRVGQiV/fSUAQXJ0RST/nk=
 github.com/Arceliar/phony v0.0.0-20210209235338-dde1a8dca979 h1:WndgpSW13S32VLQ3ugUxx2EnnWmgba1kCqPkd4Gk1yQ=
 github.com/Arceliar/phony v0.0.0-20210209235338-dde1a8dca979/go.mod h1:6Lkn+/zJilRMsKmbmG1RPoamiArC6HS73xbwRyp3UyI=
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
@@ -55,8 +55,8 @@ golang.org/x/exp v0.0.0-20190731235908-ec7cb31e5a56/go.mod h1:JhuoJpWY28nO4Vef9t
 golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js=
 golang.org/x/image v0.0.0-20190802002840-cff245a6509b/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
 golang.org/x/mobile v0.0.0-20190312151609-d3739f865fa6/go.mod h1:z+o9i4GpDbdi3rU15maQ/Ox0txvL9dWGYEHz965HBQE=
-golang.org/x/mobile v0.0.0-20221012134814-c746ac228303 h1:K4fp1rDuJBz0FCPAWzIJwnzwNEM7S6yobdZzMrZ/Zws=
+golang.org/x/mobile v0.0.0-20221110043201-43a038452099 h1:aIu0lKmfdgtn2uTj7JI2oN4TUrQvgB+wzTPO23bCKt8=
-golang.org/x/mobile v0.0.0-20221012134814-c746ac228303/go.mod h1:M32cGdzp91A8Ex9qQtyZinr19EYxzkFqDjW2oyHzTDQ=
+golang.org/x/mobile v0.0.0-20221110043201-43a038452099/go.mod h1:aAjjkJNdrh3PMckS4B10TGS2nag27cbKR1y2BpUxsiY=
 golang.org/x/mod v0.1.0/go.mod h1:0QHyrYULN0/3qlju5TqG8bIK38QM8yzMo5ekMj3DlcY=
 golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 h1:6zppjxzCulZykYSLyVDYbneBfbaBIQPYMevg0bEwv2s=
@@ -108,7 +108,6 @@ golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGm
 golang.org/x/tools v0.0.0-20190312151545-0bb0c0a6e846/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.1.7/go.mod h1:LGqMHiF4EqQNHR1JncWGqT5BVaXmza+X+BDGol+dOxo=
-golang.org/x/tools v0.1.8-0.20211022200916-316ba0b74098/go.mod h1:LGqMHiF4EqQNHR1JncWGqT5BVaXmza+X+BDGol+dOxo=
 golang.org/x/tools v0.1.12 h1:VveCTK38A2rkS8ZqFY25HIDFscX5X9OoEhJd3quQmXU=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

src/admin/admin.go

@@ -327,6 +327,7 @@ func (a *AdminSocket) handleRequest(conn net.Conn) {
 		var buf json.RawMessage
 		var req AdminSocketRequest
 		var resp AdminSocketResponse
+		req.Arguments = []byte("{}")
 		if err := func() error {
 			if err = decoder.Decode(&buf); err != nil {
 				return fmt.Errorf("Failed to find request")

src/admin/getpeers.go

@@ -19,7 +19,7 @@ type PeerEntry struct {
 	IPAddress string   `json:"address"`
 	PublicKey string   `json:"key"`
 	Port      uint64   `json:"port"`
-	Priority  uint8    `json:"priority"`
+	Priority  uint64   `json:"priority"`
 	Coords    []uint64 `json:"coords"`
 	Remote    string   `json:"remote"`
 	RXBytes   DataUnit `json:"bytes_recvd"`
@@ -36,7 +36,7 @@ func (a *AdminSocket) getPeersHandler(req *GetPeersRequest, res *GetPeersRespons
 			IPAddress: net.IP(addr[:]).String(),
 			PublicKey: hex.EncodeToString(p.Key),
 			Port:      p.Port,
-			Priority:  p.Priority,
+			Priority:  uint64(p.Priority), // can't be uint8 thanks to gobind
 			Coords:    p.Coords,
 			Remote:    p.Remote,
 			RXBytes:   DataUnit(p.RXBytes),

src/config/config.go

@@ -44,7 +44,7 @@ type MulticastInterfaceConfig struct {
 	Beacon   bool
 	Listen   bool
 	Port     uint16
-	Priority uint8
+	Priority uint64 // really uint8, but gobind won't export it
 }
 
 // NewSigningKeys replaces the signing keypair in the NodeConfig with a new

src/core/api.go

@@ -194,7 +194,7 @@ func (c *Core) AddPeer(uri string, sourceInterface string) error {
 	if err != nil {
 		return err
 	}
-	info, err := c.links.call(u, sourceInterface)
+	info, err := c.links.call(u, sourceInterface, nil)
 	if err != nil {
 		return err
 	}
@@ -236,7 +236,7 @@ func (c *Core) RemovePeer(uri string, sourceInterface string) error {
 // This does not add the peer to the peer list, so if the connection drops, the
 // peer will not be called again automatically.
 func (c *Core) CallPeer(u *url.URL, sintf string) error {
-	_, err := c.links.call(u, sintf)
+	_, err := c.links.call(u, sintf, nil)
 	return err
 }
 

src/core/core.go

@@ -121,6 +121,13 @@ func (c *Core) _addPeerLoop() {
 	})
 }
 
+func (c *Core) RetryPeersNow() {
+	if c.addPeerTimer != nil && !c.addPeerTimer.Stop() {
+		<-c.addPeerTimer.C
+	}
+	c.Act(nil, c._addPeerLoop)
+}
+
 // Stop shuts down the Yggdrasil node.
 func (c *Core) Stop() {
 	phony.Block(c, func() {

src/core/link.go

@@ -34,6 +34,11 @@ type linkInfo struct {
 	remote   string // Remote name or address
 }
 
+type linkDial struct {
+	url   *url.URL
+	sintf string
+}
+
 type link struct {
 	lname    string
 	links    *links
@@ -105,9 +110,12 @@ func (l *links) isConnectedTo(info linkInfo) bool {
 	return isConnected
 }
 
-func (l *links) call(u *url.URL, sintf string) (linkInfo, error) {
+func (l *links) call(u *url.URL, sintf string, errch chan<- error) (info linkInfo, err error) {
-	info := linkInfoFor(u.Scheme, sintf, u.Host)
+	info = linkInfoFor(u.Scheme, sintf, u.Host)
 	if l.isConnectedTo(info) {
+		if errch != nil {
+			close(errch) // already connected, no error
+		}
 		return info, nil
 	}
 	options := linkOptions{
@@ -116,6 +124,9 @@ func (l *links) call(u *url.URL, sintf string) (linkInfo, error) {
 	for _, pubkey := range u.Query()["key"] {
 		sigPub, err := hex.DecodeString(pubkey)
 		if err != nil {
+			if errch != nil {
+				close(errch)
+			}
 			return info, fmt.Errorf("pinned key contains invalid hex characters")
 		}
 		var sigPubKey keyArray
@@ -125,6 +136,9 @@ func (l *links) call(u *url.URL, sintf string) (linkInfo, error) {
 	if p := u.Query().Get("priority"); p != "" {
 		pi, err := strconv.ParseUint(p, 10, 8)
 		if err != nil {
+			if errch != nil {
+				close(errch)
+			}
 			return info, fmt.Errorf("priority invalid: %w", err)
 		}
 		options.priority = uint8(pi)
@@ -132,15 +146,27 @@ func (l *links) call(u *url.URL, sintf string) (linkInfo, error) {
 	switch info.linkType {
 	case "tcp":
 		go func() {
+			if errch != nil {
+				defer close(errch)
+			}
 			if err := l.tcp.dial(u, options, sintf); err != nil && err != io.EOF {
 				l.core.log.Warnf("Failed to dial TCP %s: %s\n", u.Host, err)
+				if errch != nil {
+					errch <- err
+				}
 			}
 		}()
 
 	case "socks":
 		go func() {
+			if errch != nil {
+				defer close(errch)
+			}
 			if err := l.socks.dial(u, options); err != nil && err != io.EOF {
 				l.core.log.Warnf("Failed to dial SOCKS %s: %s\n", u.Host, err)
+				if errch != nil {
+					errch <- err
+				}
 			}
 		}()
 
@@ -163,19 +189,34 @@ func (l *links) call(u *url.URL, sintf string) (linkInfo, error) {
 			}
 		}
 		go func() {
+			if errch != nil {
+				defer close(errch)
+			}
 			if err := l.tls.dial(u, options, sintf, tlsSNI); err != nil && err != io.EOF {
 				l.core.log.Warnf("Failed to dial TLS %s: %s\n", u.Host, err)
+				if errch != nil {
+					errch <- err
+				}
 			}
 		}()
 
 	case "unix":
 		go func() {
+			if errch != nil {
+				defer close(errch)
+			}
 			if err := l.unix.dial(u, options, sintf); err != nil && err != io.EOF {
 				l.core.log.Warnf("Failed to dial UNIX %s: %s\n", u.Host, err)
+				if errch != nil {
+					errch <- err
+				}
 			}
 		}()
 
 	default:
+		if errch != nil {
+			close(errch)
+		}
 		return info, errors.New("unknown call scheme: " + u.Scheme)
 	}
 	return info, nil
@@ -197,7 +238,7 @@ func (l *links) listen(u *url.URL, sintf string) (*Listener, error) {
 	return listener, err
 }
 
-func (l *links) create(conn net.Conn, name string, info linkInfo, incoming, force bool, options linkOptions) error {
+func (l *links) create(conn net.Conn, dial *linkDial, name string, info linkInfo, incoming, force bool, options linkOptions) error {
 	intf := link{
 		conn: &linkConn{
 			Conn: conn,
@@ -211,14 +252,14 @@ func (l *links) create(conn net.Conn, name string, info linkInfo, incoming, forc
 		force:    force,
 	}
 	go func() {
-		if err := intf.handler(); err != nil {
+		if err := intf.handler(dial); err != nil {
 			l.core.log.Errorf("Link handler %s error (%s): %s", name, conn.RemoteAddr(), err)
 		}
 	}()
 	return nil
 }
 
-func (intf *link) handler() error {
+func (intf *link) handler(dial *linkDial) error {
 	defer intf.conn.Close() // nolint:errcheck
 
 	// Don't connect to this link more than once.
@@ -321,6 +362,30 @@ func (intf *link) handler() error {
 		intf.links.core.log.Infof("Disconnected %s %s: %s, source %s; error: %s",
 			dir, strings.ToUpper(intf.info.linkType), remoteStr, localStr, err)
 	}
+
+	if !intf.incoming && dial != nil {
+		// The connection was one that we dialled, so wait a second and try to
+		// dial it again.
+		var retry func(attempt int)
+		retry = func(attempt int) {
+			// intf.links.core.log.Infof("Retrying %s (attempt %d of 5)...", dial.url.String(), attempt)
+			errch := make(chan error, 1)
+			if _, err := intf.links.call(dial.url, dial.sintf, errch); err != nil {
+				return
+			}
+			if err := <-errch; err != nil {
+				if attempt < 3 {
+					time.AfterFunc(time.Second, func() {
+						retry(attempt + 1)
+					})
+				}
+			}
+		}
+		time.AfterFunc(time.Second, func() {
+			retry(1)
+		})
+	}
+
 	return nil
 }
 

src/core/link_socks.go

@@ -37,16 +37,20 @@ func (l *linkSOCKS) dial(url *url.URL, options linkOptions) error {
 	if err != nil {
 		return err
 	}
-	return l.handler(url.String(), info, conn, options, false)
+	dial := &linkDial{
+		url: url,
+	}
+	return l.handler(dial, info, conn, options, false)
 }
 
-func (l *linkSOCKS) handler(name string, info linkInfo, conn net.Conn, options linkOptions, incoming bool) error {
+func (l *linkSOCKS) handler(dial *linkDial, info linkInfo, conn net.Conn, options linkOptions, incoming bool) error {
 	return l.links.create(
-		conn,     // connection
+		conn,              // connection
-		name,     // connection name
+		dial,              // connection URL
-		info,     // connection info
+		dial.url.String(), // connection name
-		incoming, // not incoming
+		info,              // connection info
-		false,    // not forced
+		incoming,          // not incoming
-		options,  // connection options
+		false,             // not forced
+		options,           // connection options
 	)
 }

src/core/link_tcp.go

@@ -47,8 +47,12 @@ func (l *linkTCP) dial(url *url.URL, options linkOptions, sintf string) error {
 	if err != nil {
 		return err
 	}
-	uri := strings.TrimRight(strings.SplitN(url.String(), "?", 2)[0], "/")
+	name := strings.TrimRight(strings.SplitN(url.String(), "?", 2)[0], "/")
-	return l.handler(uri, info, conn, options, false, false)
+	dial := &linkDial{
+		url:   url,
+		sintf: sintf,
+	}
+	return l.handler(dial, name, info, conn, options, false, false)
 }
 
 func (l *linkTCP) listen(url *url.URL, sintf string) (*Listener, error) {
@@ -86,7 +90,7 @@ func (l *linkTCP) listen(url *url.URL, sintf string) (*Listener, error) {
 			raddr := conn.RemoteAddr().(*net.TCPAddr)
 			name := fmt.Sprintf("tcp://%s", raddr)
 			info := linkInfoFor("tcp", sintf, tcpIDFor(laddr, raddr))
-			if err = l.handler(name, info, conn, linkOptionsForListener(url), true, raddr.IP.IsLinkLocalUnicast()); err != nil {
+			if err = l.handler(nil, name, info, conn, linkOptionsForListener(url), true, raddr.IP.IsLinkLocalUnicast()); err != nil {
 				l.core.log.Errorln("Failed to create inbound link:", err)
 			}
 		}
@@ -97,9 +101,10 @@ func (l *linkTCP) listen(url *url.URL, sintf string) (*Listener, error) {
 	return entry, nil
 }
 
-func (l *linkTCP) handler(name string, info linkInfo, conn net.Conn, options linkOptions, incoming, force bool) error {
+func (l *linkTCP) handler(dial *linkDial, name string, info linkInfo, conn net.Conn, options linkOptions, incoming, force bool) error {
 	return l.links.create(
 		conn,     // connection
+		dial,     // connection URL
 		name,     // connection name
 		info,     // connection info
 		incoming, // not incoming

src/core/link_tls.go

@@ -69,8 +69,12 @@ func (l *linkTLS) dial(url *url.URL, options linkOptions, sintf, sni string) err
 	if err != nil {
 		return err
 	}
-	uri := strings.TrimRight(strings.SplitN(url.String(), "?", 2)[0], "/")
+	name := strings.TrimRight(strings.SplitN(url.String(), "?", 2)[0], "/")
-	return l.handler(uri, info, conn, options, false, false)
+	dial := &linkDial{
+		url:   url,
+		sintf: sintf,
+	}
+	return l.handler(dial, name, info, conn, options, false, false)
 }
 
 func (l *linkTLS) listen(url *url.URL, sintf string) (*Listener, error) {
@@ -109,7 +113,7 @@ func (l *linkTLS) listen(url *url.URL, sintf string) (*Listener, error) {
 			raddr := conn.RemoteAddr().(*net.TCPAddr)
 			name := fmt.Sprintf("tls://%s", raddr)
 			info := linkInfoFor("tls", sintf, tcpIDFor(laddr, raddr))
-			if err = l.handler(name, info, conn, linkOptionsForListener(url), true, raddr.IP.IsLinkLocalUnicast()); err != nil {
+			if err = l.handler(nil, name, info, conn, linkOptionsForListener(url), true, raddr.IP.IsLinkLocalUnicast()); err != nil {
 				l.core.log.Errorln("Failed to create inbound link:", err)
 			}
 		}
@@ -120,20 +124,18 @@ func (l *linkTLS) listen(url *url.URL, sintf string) (*Listener, error) {
 	return entry, nil
 }
 
+// RFC5280 section 4.1.2.5
+var notAfterNeverExpires = time.Date(9999, time.December, 31, 23, 59, 59, 0, time.UTC)
+
 func (l *linkTLS) generateConfig() (*tls.Config, error) {
 	certBuf := &bytes.Buffer{}
-
-	// TODO: because NotAfter is finite, we should add some mechanism to
-	// regenerate the certificate and restart the listeners periodically
-	// for nodes with very high uptimes. Perhaps regenerate certs and restart
-	// listeners every few months or so.
 	cert := x509.Certificate{
 		SerialNumber: big.NewInt(1),
 		Subject: pkix.Name{
 			CommonName: hex.EncodeToString(l.links.core.public[:]),
 		},
 		NotBefore:             time.Now(),
-		NotAfter:              time.Now().Add(time.Hour * 24 * 365),
+		NotAfter:              notAfterNeverExpires,
 		KeyUsage:              x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature,
 		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
 		BasicConstraintsValid: true,
@@ -167,6 +169,6 @@ func (l *linkTLS) generateConfig() (*tls.Config, error) {
 	}, nil
 }
 
-func (l *linkTLS) handler(name string, info linkInfo, conn net.Conn, options linkOptions, incoming, force bool) error {
+func (l *linkTLS) handler(dial *linkDial, name string, info linkInfo, conn net.Conn, options linkOptions, incoming, force bool) error {
-	return l.tcp.handler(name, info, conn, options, incoming, force)
+	return l.tcp.handler(dial, name, info, conn, options, incoming, force)
 }

src/core/link_unix.go

@@ -45,7 +45,10 @@ func (l *linkUNIX) dial(url *url.URL, options linkOptions, _ string) error {
 	if err != nil {
 		return err
 	}
-	return l.handler(url.String(), info, conn, options, false)
+	dial := &linkDial{
+		url: url,
+	}
+	return l.handler(dial, url.String(), info, conn, options, false)
 }
 
 func (l *linkUNIX) listen(url *url.URL, _ string) (*Listener, error) {
@@ -74,7 +77,7 @@ func (l *linkUNIX) listen(url *url.URL, _ string) (*Listener, error) {
 				break
 			}
 			info := linkInfoFor("unix", "", url.String())
-			if err = l.handler(url.String(), info, conn, linkOptionsForListener(url), true); err != nil {
+			if err = l.handler(nil, url.String(), info, conn, linkOptionsForListener(url), true); err != nil {
 				l.core.log.Errorln("Failed to create inbound link:", err)
 			}
 		}
@@ -85,9 +88,10 @@ func (l *linkUNIX) listen(url *url.URL, _ string) (*Listener, error) {
 	return entry, nil
 }
 
-func (l *linkUNIX) handler(name string, info linkInfo, conn net.Conn, options linkOptions, incoming bool) error {
+func (l *linkUNIX) handler(dial *linkDial, name string, info linkInfo, conn net.Conn, options linkOptions, incoming bool) error {
 	return l.links.create(
 		conn,     // connection
+		dial,     // connection URL
 		name,     // connection name
 		info,     // connection info
 		incoming, // not incoming

src/multicast/multicast.go

@@ -30,6 +30,7 @@ type Multicast struct {
 	_isOpen     bool
 	_listeners  map[string]*listenerInfo
 	_interfaces map[string]*interfaceInfo
+	_timer      *time.Timer
 	config      struct {
 		_groupAddr  GroupAddress
 		_interfaces map[MulticastInterface]struct{}
@@ -207,6 +208,15 @@ func (m *Multicast) _getAllowedInterfaces() map[string]*interfaceInfo {
 	return interfaces
 }
 
+func (m *Multicast) AnnounceNow() {
+	phony.Block(m, func() {
+		if m._timer != nil && !m._timer.Stop() {
+			<-m._timer.C
+		}
+		m.Act(nil, m._announce)
+	})
+}
+
 func (m *Multicast) _announce() {
 	if !m._isOpen {
 		return
@@ -329,7 +339,7 @@ func (m *Multicast) _announce() {
 			break
 		}
 	}
-	time.AfterFunc(time.Second, func() {
+	m._timer = time.AfterFunc(time.Second, func() {
 		m.Act(nil, m._announce)
 	})
 }