Dry up creating new AURP peers

aurp/routing_info.go

@@ -173,9 +173,12 @@ func (e EventTuples) WriteTo(w io.Writer) (int64, error) {
 }
 
 func parseEventTuples(p []byte) (EventTuples, error) {
-	// Each event tuple is at least 4 bytes, so we need to store at most
-	// len(p)/4 of them.
-	e := make(EventTuples, 0, len(p)/4)
+	// Event tuples can be 1, 4, or 6 bytes long. But the only type of length 1
+	// is the Null event type sent to probe whether or not the data receiver is
+	// still listening. If that's present there probably aren't any other
+	// tuples. Hence len(p)/4 (rounded up) is a reasonable estimate of max tuple
+	// count.
+	e := make(EventTuples, 0, (len(p)+3)/4)
 	for len(p) > 0 {
 		et, nextp, err := parseEventTuple(p)
 		if err != nil {
@@ -198,6 +201,10 @@ type EventTuple struct {
 func (et *EventTuple) WriteTo(w io.Writer) (int64, error) {
 	a := acc(w)
 	a.write8(uint8(et.EventCode))
+	if et.EventCode == EventCodeNull {
+		// null tuple
+		return a.ret()
+	}
 	a.write16(uint16(et.RangeStart))
 	if !et.Extended {
 		// non-extended tuple
@@ -211,12 +218,18 @@ func (et *EventTuple) WriteTo(w io.Writer) (int64, error) {
 }
 
 func parseEventTuple(p []byte) (EventTuple, []byte, error) {
-	if len(p) < 4 {
-		return EventTuple{}, p, fmt.Errorf("insufficient input length %d for network event tuple", len(p))
+	if len(p) < 1 {
+		return EventTuple{}, p, fmt.Errorf("insufficient input length %d for any network event tuple", len(p))
 	}
 
 	var et EventTuple
 	et.EventCode = EventCode(p[0])
+	if et.EventCode == EventCodeNull {
+		return et, p[1:], nil
+	}
+	if len(p) < 4 {
+		return EventTuple{}, p, fmt.Errorf("insufficient input length %d for non-Null network event tuple", len(p))
+	}
 	et.RangeStart = ddp.Network(binary.BigEndian.Uint16(p[1:3]))
 	et.RangeEnd = et.RangeStart
 	et.Distance = p[3]

main.go

@@ -65,13 +65,13 @@ const routingTableTemplate = `
 		<td>{{$route.LastSeenAgo}}</td>
 		<td>
 			{{- with $route.AURPPeer -}}
-				{{.RemoteAddr}}
+				[redacted]
 			{{- end -}}
 			{{- with $route.EtherTalkPeer -}}
 				{{.Port.Device}} {{.PeerAddr.Network}}.{{.PeerAddr.Node}}
 			{{- end -}}
 			{{- with $route.EtherTalkDirect -}}
-				{{.Device}} {{.NetStart}}-{{.NetEnd}}
+				{{.Device}}
 			{{- end -}}
 		</td>
 	</tr>
@@ -87,14 +87,24 @@ const peerTableTemplate = `
 		<th>Remote addr</th>
 		<th>Receiver state</th>
 		<th>Sender state</th>
+		<th>Last heard from</th>
+		<th>Last reconnect</th>
+		<th>Last update</th>
+		<th>Last send</th>
+		<th>Send retries</th>
 	</tr></thead>
 	<tbody>
 {{range $peer := . }}
 	<tr>
-		<td>{{$peer.ConfiguredAddr}}</td>
-		<td>{{$peer.RemoteAddr}}</td>
+		<td>[redacted]</td>
+		<td>[redacted]</td>
 		<td>{{$peer.ReceiverState}}</td>
 		<td>{{$peer.SenderState}}</td>
+		<td>{{$peer.LastHeardFromAgo}}</td>
+		<td>{{$peer.LastReconnectAgo}}</td>
+		<td>{{$peer.LastUpdateAgo}}</td>
+		<td>{{$peer.LastSendAgo}}</td>
+		<td>{{$peer.SendRetries}}</td>		
 	</tr>
 {{end}}
 	</tbody>
@@ -142,9 +152,9 @@ func main() {
 	}
 	localDI := aurp.IPDomainIdentifier(localIP)
 
-	log.Printf("Using %v as local domain identifier", localIP)
+	// log.Printf("Using %v as local domain identifier", localIP)
 
-	log.Printf("EtherTalk configuration: %+v", cfg.EtherTalk)
+	// log.Printf("EtherTalk configuration: %+v", cfg.EtherTalk)
 
 	ln, err := net.ListenUDP("udp4", &net.UDPAddr{Port: int(cfg.ListenPort)})
 	if err != nil {
@@ -246,12 +256,12 @@ func main() {
 
 	// ------------------------- Configured peer setup ------------------------
 	if cfg.PeerListURL != "" {
-		log.Printf("Fetching peer list from %s...", cfg.PeerListURL)
+		log.Print("Fetching peer list...")
 		existing := len(cfg.Peers)
 		func() {
 			resp, err := http.Get(cfg.PeerListURL)
 			if err != nil {
-				log.Fatalf("Couldn't fetch peer list: %v", err)
+				log.Fatalf("Couldn't fetch peer list!")
 			}
 			defer resp.Body.Close()
 
@@ -277,28 +287,17 @@ func main() {
 
 		raddr, err := net.ResolveUDPAddr("udp4", peerStr)
 		if err != nil {
-			log.Printf("couldn't resolve UDP address, skipping: %v", err)
+			log.Print("couldn't resolve UDP address, skipping peer")
 			continue
 		}
-		log.Printf("resolved %q to %v", peerStr, raddr)
+		//log.Printf("resolved %q to %v", peerStr, raddr)
 
 		if raddr.IP.Equal(localIP) {
-			log.Printf("%v == %v == me, skipping", peerStr, raddr)
+			//log.Print("peer == me, skipping")
 			continue
 		}
 
-		peer := &router.AURPPeer{
-			Transport: &aurp.Transport{
-				LocalDI:     localDI,
-				RemoteDI:    aurp.IPDomainIdentifier(raddr.IP),
-				LocalConnID: nextConnID,
-			},
-			UDPConn:        ln,
-			ConfiguredAddr: peerStr,
-			RemoteAddr:     raddr,
-			ReceiveCh:      make(chan aurp.Packet, 1024),
-			RouteTable:     routes,
-		}
+		peer := router.NewAURPPeer(routes, ln, peerStr, raddr, localDI, nil, nextConnID)
 		aurp.Inc(&nextConnID)
 		peersMu.Lock()
 		peers[udpAddrFromNet(raddr)] = peer
@@ -381,7 +380,7 @@ func main() {
 				return
 			}
 
-			log.Printf("AURP: Got %T from %v (%v)", pkt, raddr, dh.SourceDI)
+			// log.Printf("AURP: Got %T from %v (%v)", pkt, raddr, dh.SourceDI)
 
 			// Existing peer?
 			ra := udpAddrFromNet(raddr)
@@ -389,22 +388,12 @@ func main() {
 			pr := peers[ra]
 			if pr == nil {
 				if !cfg.OpenPeering {
-					log.Printf("AURP: Got packet from %v but it's not in my config and open peering is disabled; dropping the packet", raddr)
+					log.Print("AURP: Got packet but it's not in my config and open peering is disabled; dropping the packet")
 					peersMu.Unlock()
 					continue
 				}
 				// New peer!
-				pr = &router.AURPPeer{
-					Transport: &aurp.Transport{
-						LocalDI:     localDI,
-						RemoteDI:    dh.SourceDI, // platinum rule
-						LocalConnID: nextConnID,
-					},
-					UDPConn:    ln,
-					RemoteAddr: raddr,
-					ReceiveCh:  make(chan aurp.Packet, 1024),
-					RouteTable: routes,
-				}
+				pr = router.NewAURPPeer(routes, ln, "", raddr, localDI, dh.SourceDI, nextConnID)
 				aurp.Inc(&nextConnID)
 				peers[ra] = pr
 				goPeerHandler(pr)

router/aarp.go

@@ -378,10 +378,7 @@ func (e AMTEntry) Valid() bool {
 // LastUpdatedAgo is a friendly string reporting how long ago the entry was
 // updated/resolved.
 func (e AMTEntry) LastUpdatedAgo() string {
-	if e.LastUpdated.IsZero() {
-		return "never"
-	}
-	return fmt.Sprintf("%v ago", time.Since(e.LastUpdated).Truncate(time.Millisecond))
+	return ago(e.LastUpdated)
 }
 
 // addressMappingTable implements a concurrent-safe Address Mapping Table for

router/misc.go

@@ -16,6 +16,11 @@
 
 package router
 
+import (
+	"fmt"
+	"time"
+)
+
 // StringSet is a set of strings.
 // Yep, yet another string set implementation. Took me 2 minutes to write *shrug*
 type StringSet map[string]struct{}
@@ -50,3 +55,11 @@ func SetFromSlice(ss []string) StringSet {
 	set.Insert(ss...)
 	return set
 }
+
+// ago is a helper for formatting times.
+func ago(t time.Time) string {
+	if t.IsZero() {
+		return "never"
+	}
+	return fmt.Sprintf("%v ago", time.Since(t).Truncate(time.Millisecond))
+}

router/peer_aurp.go

@@ -114,9 +114,73 @@ type AURPPeer struct {
 	// Route table (the peer will add/remove/update routes and zones)
 	RouteTable *RouteTable
 
-	mu     sync.RWMutex
-	rstate ReceiverState
-	sstate SenderState
+	// Event tuples yet to be sent to this peer in an RI-Upd.
+	pendingEvents chan aurp.EventTuple
+
+	// The internal states below are only set within the Handle loop, but can
+	// be read concurrently from outside.
+	mu            sync.RWMutex
+	rstate        ReceiverState
+	sstate        SenderState
+	lastReconnect time.Time
+	lastHeardFrom time.Time
+	lastSend      time.Time // TODO: clarify use of lastSend / sendRetries
+	lastUpdate    time.Time
+	sendRetries   int
+}
+
+func NewAURPPeer(routes *RouteTable, udpConn *net.UDPConn, peerAddr string, raddr *net.UDPAddr, localDI, remoteDI aurp.DomainIdentifier, connID uint16) *AURPPeer {
+	if remoteDI == nil {
+		remoteDI = aurp.IPDomainIdentifier(raddr.IP)
+	}
+	return &AURPPeer{
+		Transport: &aurp.Transport{
+			LocalDI:     localDI,
+			RemoteDI:    remoteDI,
+			LocalConnID: connID,
+		},
+		UDPConn:        udpConn,
+		ConfiguredAddr: peerAddr,
+		RemoteAddr:     raddr,
+		ReceiveCh:      make(chan aurp.Packet, 1024),
+		RouteTable:     routes,
+		pendingEvents:  make(chan aurp.EventTuple, 1024),
+	}
+}
+
+func (p *AURPPeer) addPendingEvent(ec aurp.EventCode, route *Route) {
+	// Don't advertise routes to AURP peers to other AURP peers
+	if route.AURPPeer != nil {
+		return
+	}
+	et := aurp.EventTuple{
+		EventCode:  ec,
+		Extended:   route.Extended,
+		RangeStart: route.NetStart,
+		Distance:   route.Distance,
+		RangeEnd:   route.NetEnd,
+	}
+	switch ec {
+	case aurp.EventCodeND, aurp.EventCodeNRC:
+		et.Distance = 0 // "The distance field does not apply to ND or NRC event tuples and should be set to 0."
+	}
+	p.pendingEvents <- et
+}
+
+func (p *AURPPeer) RouteAdded(route *Route) {
+	p.addPendingEvent(aurp.EventCodeNA, route)
+}
+
+func (p *AURPPeer) RouteDeleted(route *Route) {
+	p.addPendingEvent(aurp.EventCodeND, route)
+}
+
+func (p *AURPPeer) RouteDistanceChanged(route *Route) {
+	p.addPendingEvent(aurp.EventCodeNDC, route)
+}
+
+func (p *AURPPeer) RouteForwarderChanged(route *Route) {
+	p.addPendingEvent(aurp.EventCodeNRC, route)
 }
 
 func (p *AURPPeer) Forward(ddpkt *ddp.ExtPacket) error {
@@ -124,7 +188,7 @@ func (p *AURPPeer) Forward(ddpkt *ddp.ExtPacket) error {
 	if err != nil {
 		return err
 	}
-	_, err = p.Send(p.Transport.NewAppleTalkPacket(outPkt))
+	_, err = p.send(p.Transport.NewAppleTalkPacket(outPkt))
 	return err
 }
 
@@ -140,6 +204,36 @@ func (p *AURPPeer) SenderState() SenderState {
 	return p.sstate
 }
 
+func (p *AURPPeer) LastReconnectAgo() string {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+	return ago(p.lastReconnect)
+}
+
+func (p *AURPPeer) LastHeardFromAgo() string {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+	return ago(p.lastHeardFrom)
+}
+
+func (p *AURPPeer) LastSendAgo() string {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+	return ago(p.lastSend)
+}
+
+func (p *AURPPeer) LastUpdateAgo() string {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+	return ago(p.lastUpdate)
+}
+
+func (p *AURPPeer) SendRetries() int {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+	return p.sendRetries
+}
+
 func (p *AURPPeer) setRState(rstate ReceiverState) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
@@ -152,6 +246,42 @@ func (p *AURPPeer) setSState(sstate SenderState) {
 	p.sstate = sstate
 }
 
+func (p *AURPPeer) incSendRetries() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.sendRetries++
+}
+
+func (p *AURPPeer) resetSendRetries() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.sendRetries = 0
+}
+
+func (p *AURPPeer) bumpLastHeardFrom() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.lastHeardFrom = time.Now()
+}
+
+func (p *AURPPeer) bumpLastReconnect() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.lastReconnect = time.Now()
+}
+
+func (p *AURPPeer) bumpLastSend() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.lastSend = time.Now()
+}
+
+func (p *AURPPeer) bumpLastUpdate() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.lastUpdate = time.Now()
+}
+
 func (p *AURPPeer) disconnect() {
 	p.mu.Lock()
 	defer p.mu.Unlock()
@@ -159,13 +289,13 @@ func (p *AURPPeer) disconnect() {
 	p.sstate = SenderUnconnected
 }
 
-// Send encodes and sends pkt to the remote host.
-func (p *AURPPeer) Send(pkt aurp.Packet) (int, error) {
+// send encodes and sends pkt to the remote host.
+func (p *AURPPeer) send(pkt aurp.Packet) (int, error) {
 	var b bytes.Buffer
 	if _, err := pkt.WriteTo(&b); err != nil {
 		return 0, err
 	}
-	log.Printf("AURP Peer: Sending %T (len %d) to %v", pkt, b.Len(), p.RemoteAddr)
+	// log.Printf("AURP Peer: Sending %T (len %d) to %v", pkt, b.Len(), p.RemoteAddr)
 	return p.UDPConn.WriteToUDP(b.Bytes(), p.RemoteAddr)
 }
 
@@ -175,18 +305,20 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 	sticker := time.NewTicker(1 * time.Second)
 	defer sticker.Stop()
 
-	lastReconnect := time.Now()
-	lastHeardFrom := time.Now()
-	lastSend := time.Now() // TODO: clarify use of lastSend / sendRetries
-	lastUpdate := time.Now()
-	sendRetries := 0
+	p.mu.Lock()
+	p.lastReconnect = time.Now()
+	p.lastHeardFrom = time.Now()
+	p.lastSend = time.Now() // TODO: clarify use of lastSend / sendRetries
+	p.lastUpdate = time.Now()
+	p.sendRetries = 0
+	p.mu.Unlock()
 
 	var lastRISent aurp.Packet
 
 	p.disconnect()
 
 	// Write an Open-Req packet
-	if _, err := p.Send(p.Transport.NewOpenReqPacket(nil)); err != nil {
+	if _, err := p.send(p.Transport.NewOpenReqPacket(nil)); err != nil {
 		log.Printf("AURP Peer: Couldn't send Open-Req packet: %v", err)
 		return err
 	}
@@ -202,7 +334,7 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 			}
 			// Send a best-effort Router Down before returning
 			lastRISent = p.Transport.NewRDPacket(aurp.ErrCodeNormalClose)
-			if _, err := p.Send(lastRISent); err != nil {
+			if _, err := p.send(lastRISent); err != nil {
 				log.Printf("Couldn't send RD packet: %v", err)
 			}
 			return ctx.Err()
@@ -210,60 +342,60 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 		case <-rticker.C:
 			switch p.rstate {
 			case ReceiverWaitForOpenRsp:
-				if time.Since(lastSend) <= sendRetryTimer {
+				if time.Since(p.lastSend) <= sendRetryTimer {
 					break
 				}
-				if sendRetries >= sendRetryLimit {
+				if p.sendRetries >= sendRetryLimit {
 					log.Printf("AURP Peer: Send retry limit reached while waiting for Open-Rsp, closing connection")
 					p.setRState(ReceiverUnconnected)
 					break
 				}
 
 				// Send another Open-Req
-				sendRetries++
-				lastSend = time.Now()
-				if _, err := p.Send(p.Transport.NewOpenReqPacket(nil)); err != nil {
+				p.incSendRetries()
+				p.bumpLastSend()
+				if _, err := p.send(p.Transport.NewOpenReqPacket(nil)); err != nil {
 					log.Printf("AURP Peer: Couldn't send Open-Req packet: %v", err)
 					return err
 				}
 
 			case ReceiverConnected:
 				// Check LHFT, send tickle?
-				if time.Since(lastHeardFrom) <= lastHeardFromTimer {
+				if time.Since(p.lastHeardFrom) <= lastHeardFromTimer {
 					break
 				}
-				if _, err := p.Send(p.Transport.NewTicklePacket()); err != nil {
+				if _, err := p.send(p.Transport.NewTicklePacket()); err != nil {
 					log.Printf("AURP Peer: Couldn't send Tickle: %v", err)
 					return err
 				}
 				p.setRState(ReceiverWaitForTickleAck)
-				sendRetries = 0
-				lastSend = time.Now()
+				p.resetSendRetries()
+				p.bumpLastSend()
 
 			case ReceiverWaitForTickleAck:
-				if time.Since(lastSend) <= sendRetryTimer {
+				if time.Since(p.lastSend) <= sendRetryTimer {
 					break
 				}
-				if sendRetries >= tickleRetryLimit {
+				if p.sendRetries >= tickleRetryLimit {
 					log.Printf("AURP Peer: Send retry limit reached while waiting for Tickle-Ack, closing connection")
 					p.setRState(ReceiverUnconnected)
 					p.RouteTable.DeleteAURPPeer(p)
 					break
 				}
 
-				sendRetries++
-				lastSend = time.Now()
-				if _, err := p.Send(p.Transport.NewTicklePacket()); err != nil {
+				p.incSendRetries()
+				p.bumpLastSend()
+				if _, err := p.send(p.Transport.NewTicklePacket()); err != nil {
 					log.Printf("AURP Peer: Couldn't send Tickle: %v", err)
 					return err
 				}
 				// still in Wait For Tickle-Ack
 
 			case ReceiverWaitForRIRsp:
-				if time.Since(lastSend) <= sendRetryTimer {
+				if time.Since(p.lastSend) <= sendRetryTimer {
 					break
 				}
-				if sendRetries >= sendRetryLimit {
+				if p.sendRetries >= sendRetryLimit {
 					log.Printf("AURP Peer: Send retry limit reached while waiting for RI-Rsp, closing connection")
 					p.setRState(ReceiverUnconnected)
 					p.RouteTable.DeleteAURPPeer(p)
@@ -272,8 +404,9 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 
 				// RI-Req is stateless, so we don't need to cache the one we
 				// sent earlier just to send it again
-				sendRetries++
-				if _, err := p.Send(p.Transport.NewRIReqPacket()); err != nil {
+				p.incSendRetries()
+				p.bumpLastSend()
+				if _, err := p.send(p.Transport.NewRIReqPacket()); err != nil {
 					log.Printf("AURP Peer: Couldn't send RI-Req packet: %v", err)
 					return err
 				}
@@ -282,18 +415,18 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 			case ReceiverUnconnected:
 				// Data receiver is unconnected. If data sender is connected,
 				// send a null RI-Upd to check if the sender is also unconnected
-				if p.sstate == SenderConnected && time.Since(lastSend) > sendRetryTimer {
-					if sendRetries >= sendRetryLimit {
+				if p.sstate == SenderConnected && time.Since(p.lastSend) > sendRetryTimer {
+					if p.sendRetries >= sendRetryLimit {
 						log.Printf("AURP Peer: Send retry limit reached while probing sender connect, closing connection")
 					}
-					sendRetries++
-					lastSend = time.Now()
+					p.incSendRetries()
+					p.bumpLastSend()
 					aurp.Inc(&p.Transport.LocalSeq)
 					events := aurp.EventTuples{{
 						EventCode: aurp.EventCodeNull,
 					}}
 					lastRISent = p.Transport.NewRIUpdPacket(events)
-					if _, err := p.Send(lastRISent); err != nil {
+					if _, err := p.send(lastRISent); err != nil {
 						log.Printf("AURP Peer: Couldn't send RI-Upd packet: %v", err)
 						return err
 					}
@@ -302,23 +435,23 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 
 				if p.ConfiguredAddr != "" {
 					// Periodically try to reconnect, if this peer is in the config file
-					if time.Since(lastReconnect) <= reconnectTimer {
+					if time.Since(p.lastReconnect) <= reconnectTimer {
 						break
 					}
 
 					// In case it's a DNS name, re-resolve it before reconnecting
 					raddr, err := net.ResolveUDPAddr("udp4", p.ConfiguredAddr)
 					if err != nil {
-						log.Printf("couldn't resolve UDP address, skipping: %v", err)
+						log.Print("couldn't resolve UDP address, skipping")
 						break
 					}
-					log.Printf("AURP Peer: resolved %q to %v", p.ConfiguredAddr, raddr)
+					// log.Printf("AURP Peer: resolved %q to %v", p.ConfiguredAddr, raddr)
 					p.RemoteAddr = raddr
 
-					lastReconnect = time.Now()
-					sendRetries = 0
-					lastSend = time.Now()
-					if _, err := p.Send(p.Transport.NewOpenReqPacket(nil)); err != nil {
+					p.bumpLastReconnect()
+					p.resetSendRetries()
+					p.bumpLastSend()
+					if _, err := p.send(p.Transport.NewOpenReqPacket(nil)); err != nil {
 						log.Printf("AURP Peer: Couldn't send Open-Req packet: %v", err)
 						return err
 					}
@@ -332,40 +465,41 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 				// Do nothing
 
 			case SenderConnected:
-				if time.Since(lastUpdate) <= updateTimer {
+				if time.Since(p.lastUpdate) <= updateTimer {
 					break
 				}
 				// TODO: is there a routing update to send?
+				p.bumpLastUpdate()
 
 			case SenderWaitForRIRspAck, SenderWaitForRIUpdAck:
-				if time.Since(lastSend) <= sendRetryTimer {
+				if time.Since(p.lastSend) <= sendRetryTimer {
 					break
 				}
 				if lastRISent == nil {
 					log.Print("AURP Peer: sender retry: lastRISent = nil?")
 					continue
 				}
-				if sendRetries >= sendRetryLimit {
+				if p.sendRetries >= sendRetryLimit {
 					log.Printf("AURP Peer: Send retry limit reached, closing connection")
 					p.setSState(SenderUnconnected)
 					continue
 				}
-				sendRetries++
-				lastSend = time.Now()
-				if _, err := p.Send(lastRISent); err != nil {
+				p.incSendRetries()
+				p.bumpLastSend()
+				if _, err := p.send(lastRISent); err != nil {
 					log.Printf("AURP Peer: Couldn't re-send %T: %v", lastRISent, err)
 					return err
 				}
 
 			case SenderWaitForRDAck:
-				if time.Since(lastSend) <= sendRetryTimer {
+				if time.Since(p.lastSend) <= sendRetryTimer {
 					break
 				}
 				p.setSState(SenderUnconnected)
 			}
 
 		case pkt := <-p.ReceiveCh:
-			lastHeardFrom = time.Now()
+			p.bumpLastHeardFrom()
 
 			switch pkt := pkt.(type) {
 			case *aurp.OpenReqPacket:
@@ -392,7 +526,7 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 					orsp = p.Transport.NewOpenRspPacket(0, 1, nil)
 				}
 
-				if _, err := p.Send(orsp); err != nil {
+				if _, err := p.send(orsp); err != nil {
 					log.Printf("AURP Peer: Couldn't send Open-Rsp: %v", err)
 					return err
 				}
@@ -402,9 +536,9 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 
 				// If receiver is unconnected, commence connecting
 				if p.rstate == ReceiverUnconnected {
-					lastSend = time.Now()
-					sendRetries = 0
-					if _, err := p.Send(p.Transport.NewOpenReqPacket(nil)); err != nil {
+					p.resetSendRetries()
+					p.bumpLastSend()
+					if _, err := p.send(p.Transport.NewOpenReqPacket(nil)); err != nil {
 						log.Printf("AURP Peer: Couldn't send Open-Req packet: %v", err)
 						return err
 					}
@@ -425,8 +559,8 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 				p.setRState(ReceiverConnected)
 
 				// Send an RI-Req
-				sendRetries = 0
-				if _, err := p.Send(p.Transport.NewRIReqPacket()); err != nil {
+				p.resetSendRetries()
+				if _, err := p.send(p.Transport.NewRIReqPacket()); err != nil {
 					log.Printf("AURP Peer: Couldn't send RI-Req packet: %v", err)
 					return err
 				}
@@ -449,7 +583,7 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 				p.Transport.LocalSeq = 1
 				// TODO: Split tuples across multiple packets as required
 				lastRISent = p.Transport.NewRIRspPacket(aurp.RoutingFlagLast, nets)
-				if _, err := p.Send(lastRISent); err != nil {
+				if _, err := p.send(lastRISent); err != nil {
 					log.Printf("AURP Peer: Couldn't send RI-Rsp packet: %v", err)
 					return err
 				}
@@ -474,7 +608,7 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 
 				// TODO: track which networks we don't have zone info for, and
 				// only set SZI for those ?
-				if _, err := p.Send(p.Transport.NewRIAckPacket(pkt.ConnectionID, pkt.Sequence, aurp.RoutingFlagSendZoneInfo)); err != nil {
+				if _, err := p.send(p.Transport.NewRIAckPacket(pkt.ConnectionID, pkt.Sequence, aurp.RoutingFlagSendZoneInfo)); err != nil {
 					log.Printf("AURP Peer: Couldn't send RI-Ack packet: %v", err)
 					return err
 				}
@@ -500,7 +634,7 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 				}
 
 				p.setSState(SenderConnected)
-				sendRetries = 0
+				p.resetSendRetries()
 
 				// If SZI flag is set, send ZI-Rsp (transaction)
 				if pkt.Flags&aurp.RoutingFlagSendZoneInfo != 0 {
@@ -525,7 +659,7 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 					}
 					zones := p.RouteTable.ZonesForNetworks(nets)
 					// TODO: split ZI-Rsp packets similarly to ZIP Replies
-					if _, err := p.Send(p.Transport.NewZIRspPacket(zones)); err != nil {
+					if _, err := p.send(p.Transport.NewZIRspPacket(zones)); err != nil {
 						log.Printf("AURP Peer: Couldn't send ZI-Rsp packet: %v", err)
 					}
 				}
@@ -536,9 +670,9 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 					// Receiver is unconnected, but their receiver sent us an
 					// RI-Ack for something
 					// Try to reconnect?
-					lastSend = time.Now()
-					sendRetries = 0
-					if _, err := p.Send(p.Transport.NewOpenReqPacket(nil)); err != nil {
+					p.resetSendRetries()
+					p.bumpLastSend()
+					if _, err := p.send(p.Transport.NewOpenReqPacket(nil)); err != nil {
 						log.Printf("AURP Peer: Couldn't send Open-Req packet: %v", err)
 						return err
 					}
@@ -546,7 +680,6 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 				}
 
 			case *aurp.RIUpdPacket:
-
 				var ackFlag aurp.RoutingFlag
 
 				for _, et := range pkt.Events {
@@ -586,7 +719,7 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 					}
 				}
 
-				if _, err := p.Send(p.Transport.NewRIAckPacket(pkt.ConnectionID, pkt.Sequence, ackFlag)); err != nil {
+				if _, err := p.send(p.Transport.NewRIAckPacket(pkt.ConnectionID, pkt.Sequence, ackFlag)); err != nil {
 					log.Printf("AURP Peer: Couldn't send RI-Ack: %v", err)
 					return err
 				}
@@ -600,7 +733,7 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 				p.RouteTable.DeleteAURPPeer(p)
 
 				// Respond with RI-Ack
-				if _, err := p.Send(p.Transport.NewRIAckPacket(pkt.ConnectionID, pkt.Sequence, 0)); err != nil {
+				if _, err := p.send(p.Transport.NewRIAckPacket(pkt.ConnectionID, pkt.Sequence, 0)); err != nil {
 					log.Printf("AURP Peer: Couldn't send RI-Ack: %v", err)
 					return err
 				}
@@ -610,7 +743,7 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 			case *aurp.ZIReqPacket:
 				// TODO: split ZI-Rsp packets similarly to ZIP Replies
 				zones := p.RouteTable.ZonesForNetworks(pkt.Networks)
-				if _, err := p.Send(p.Transport.NewZIRspPacket(zones)); err != nil {
+				if _, err := p.send(p.Transport.NewZIRspPacket(zones)); err != nil {
 					log.Printf("AURP Peer: Couldn't send ZI-Rsp packet: %v", err)
 					return err
 				}
@@ -622,7 +755,7 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 				}
 
 			case *aurp.GDZLReqPacket:
-				if _, err := p.Send(p.Transport.NewGDZLRspPacket(-1, nil)); err != nil {
+				if _, err := p.send(p.Transport.NewGDZLRspPacket(-1, nil)); err != nil {
 					log.Printf("AURP Peer: Couldn't send GDZL-Rsp packet: %v", err)
 					return err
 				}
@@ -631,7 +764,7 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 				log.Printf("AURP Peer: Received a GDZL-Rsp, but I wouldn't have sent a GDZL-Req - that's weird")
 
 			case *aurp.GZNReqPacket:
-				if _, err := p.Send(p.Transport.NewGZNRspPacket(pkt.ZoneName, false, nil)); err != nil {
+				if _, err := p.send(p.Transport.NewGZNRspPacket(pkt.ZoneName, false, nil)); err != nil {
 					log.Printf("AURP Peer: Couldn't send GZN-Rsp packet: %v", err)
 					return err
 				}
@@ -641,7 +774,7 @@ func (p *AURPPeer) Handle(ctx context.Context) error {
 
 			case *aurp.TicklePacket:
 				// Immediately respond with Tickle-Ack
-				if _, err := p.Send(p.Transport.NewTickleAckPacket()); err != nil {
+				if _, err := p.send(p.Transport.NewTickleAckPacket()); err != nil {
 					log.Printf("AURP Peer: Couldn't send Tickle-Ack: %v", err)
 					return err
 				}

router/route.go

@@ -45,10 +45,7 @@ type Route struct {
 }
 
 func (r Route) LastSeenAgo() string {
-	if r.LastSeen.IsZero() {
-		return "never"
-	}
-	return fmt.Sprintf("%v ago", time.Since(r.LastSeen).Truncate(time.Millisecond))
+	return ago(r.LastSeen)
 }
 
 // Valid reports whether the route is valid.
@@ -58,17 +55,40 @@ func (r *Route) Valid() bool {
 	return len(r.ZoneNames) > 0 && (r.EtherTalkPeer == nil || time.Since(r.LastSeen) <= maxRouteAge)
 }
 
+type RouteTableObserver interface {
+	RouteAdded(*Route)
+	RouteDeleted(*Route)
+	RouteDistanceChanged(*Route)
+	RouteForwarderChanged(*Route)
+}
+
 type RouteTable struct {
-	mu     sync.Mutex
-	routes map[*Route]struct{}
+	routesMu sync.RWMutex
+	routes   map[*Route]struct{}
+
+	observersMu sync.RWMutex
+	observers   map[RouteTableObserver]struct{}
 }
 
 func NewRouteTable() *RouteTable {
 	return &RouteTable{
-		routes: make(map[*Route]struct{}),
+		routes:    make(map[*Route]struct{}),
+		observers: make(map[RouteTableObserver]struct{}),
 	}
 }
 
+func (rt *RouteTable) AddObserver(obs RouteTableObserver) {
+	rt.observersMu.Lock()
+	defer rt.observersMu.Unlock()
+	rt.observers[obs] = struct{}{}
+}
+
+func (rt *RouteTable) RemoveObserver(obs RouteTableObserver) {
+	rt.observersMu.Lock()
+	defer rt.observersMu.Unlock()
+	delete(rt.observers, obs)
+}
+
 func (rt *RouteTable) InsertEtherTalkDirect(port *EtherTalkPort) {
 	r := &Route{
 		Extended:        true,
@@ -80,14 +100,14 @@ func (rt *RouteTable) InsertEtherTalkDirect(port *EtherTalkPort) {
 		EtherTalkDirect: port,
 	}
 
-	rt.mu.Lock()
-	defer rt.mu.Unlock()
+	rt.routesMu.Lock()
+	defer rt.routesMu.Unlock()
 	rt.routes[r] = struct{}{}
 }
 
 func (rt *RouteTable) Dump() []Route {
-	rt.mu.Lock()
-	defer rt.mu.Unlock()
+	rt.routesMu.Lock()
+	defer rt.routesMu.Unlock()
 
 	table := make([]Route, 0, len(rt.routes))
 	for r := range rt.routes {
@@ -97,8 +117,8 @@ func (rt *RouteTable) Dump() []Route {
 }
 
 func (rt *RouteTable) LookupRoute(network ddp.Network) *Route {
-	rt.mu.Lock()
-	defer rt.mu.Unlock()
+	rt.routesMu.Lock()
+	defer rt.routesMu.Unlock()
 
 	var bestRoute *Route
 	for r := range rt.routes {
@@ -120,8 +140,8 @@ func (rt *RouteTable) LookupRoute(network ddp.Network) *Route {
 }
 
 func (rt *RouteTable) DeleteAURPPeer(peer *AURPPeer) {
-	rt.mu.Lock()
-	defer rt.mu.Unlock()
+	rt.routesMu.Lock()
+	defer rt.routesMu.Unlock()
 
 	for route := range rt.routes {
 		if route.AURPPeer == peer {
@@ -131,8 +151,8 @@ func (rt *RouteTable) DeleteAURPPeer(peer *AURPPeer) {
 }
 
 func (rt *RouteTable) DeleteAURPPeerNetwork(peer *AURPPeer, network ddp.Network) {
-	rt.mu.Lock()
-	defer rt.mu.Unlock()
+	rt.routesMu.Lock()
+	defer rt.routesMu.Unlock()
 
 	for route := range rt.routes {
 		if route.AURPPeer == peer && route.NetStart == network {
@@ -142,8 +162,8 @@ func (rt *RouteTable) DeleteAURPPeerNetwork(peer *AURPPeer, network ddp.Network)
 }
 
 func (rt *RouteTable) UpdateAURPRouteDistance(peer *AURPPeer, network ddp.Network, distance uint8) {
-	rt.mu.Lock()
-	defer rt.mu.Unlock()
+	rt.routesMu.Lock()
+	defer rt.routesMu.Unlock()
 
 	for route := range rt.routes {
 		if route.AURPPeer == peer && route.NetStart == network {
@@ -161,8 +181,8 @@ func (rt *RouteTable) UpsertEtherTalkRoute(peer *EtherTalkPeer, extended bool, n
 		return nil, fmt.Errorf("invalid network range [%d, %d] for nonextended network", netStart, netEnd)
 	}
 
-	rt.mu.Lock()
-	defer rt.mu.Unlock()
+	rt.routesMu.Lock()
+	defer rt.routesMu.Unlock()
 
 	// Update?
 	for r := range rt.routes {
@@ -213,16 +233,16 @@ func (rt *RouteTable) InsertAURPRoute(peer *AURPPeer, extended bool, netStart, n
 		AURPPeer: peer,
 	}
 
-	rt.mu.Lock()
-	defer rt.mu.Unlock()
+	rt.routesMu.Lock()
+	defer rt.routesMu.Unlock()
 	rt.routes[r] = struct{}{}
 	return nil
 }
 
 // ValidRoutes returns all valid routes.
 func (rt *RouteTable) ValidRoutes() []*Route {
-	rt.mu.Lock()
-	defer rt.mu.Unlock()
+	rt.routesMu.RLock()
+	defer rt.routesMu.RUnlock()
 	valid := make([]*Route, 0, len(rt.routes))
 	for r := range rt.routes {
 		if r.Valid() {
@@ -234,8 +254,8 @@ func (rt *RouteTable) ValidRoutes() []*Route {
 
 // ValidNonAURPRoutes returns all valid routes that were not learned via AURP.
 func (rt *RouteTable) ValidNonAURPRoutes() []*Route {
-	rt.mu.Lock()
-	defer rt.mu.Unlock()
+	rt.routesMu.RLock()
+	defer rt.routesMu.RUnlock()
 	valid := make([]*Route, 0, len(rt.routes))
 	for r := range rt.routes {
 		if r.AURPPeer != nil {

router/zones.go

@@ -23,8 +23,8 @@ import (
 )
 
 func (rt *RouteTable) AddZonesToNetwork(n ddp.Network, zs ...string) {
-	rt.mu.Lock()
-	defer rt.mu.Unlock()
+	rt.routesMu.Lock()
+	defer rt.routesMu.Unlock()
 	for r := range rt.routes {
 		if n < r.NetStart || n > r.NetEnd {
 			continue
@@ -39,8 +39,8 @@ func (rt *RouteTable) AddZonesToNetwork(n ddp.Network, zs ...string) {
 func (rt *RouteTable) ZonesForNetworks(ns []ddp.Network) map[ddp.Network][]string {
 	zs := make(map[ddp.Network][]string)
 
-	rt.mu.Lock()
-	defer rt.mu.Unlock()
+	rt.routesMu.Lock()
+	defer rt.routesMu.Unlock()
 	for r := range rt.routes {
 		if !r.Valid() {
 			continue
@@ -55,8 +55,8 @@ func (rt *RouteTable) ZonesForNetworks(ns []ddp.Network) map[ddp.Network][]strin
 }
 
 func (rt *RouteTable) RoutesForZone(zone string) []*Route {
-	rt.mu.Lock()
-	defer rt.mu.Unlock()
+	rt.routesMu.Lock()
+	defer rt.routesMu.Unlock()
 
 	var routes []*Route
 	for r := range rt.routes {
@@ -73,8 +73,8 @@ func (rt *RouteTable) RoutesForZone(zone string) []*Route {
 func (rt *RouteTable) AllZoneNames() (zones []string) {
 	defer slices.Sort(zones)
 
-	rt.mu.Lock()
-	defer rt.mu.Unlock()
+	rt.routesMu.Lock()
+	defer rt.routesMu.Unlock()
 
 	zs := make(StringSet)
 	for r := range rt.routes {