refactor: polyamide api and wireguard transports

core/ipc_handler.go

@@ -10,13 +10,15 @@ import (
 	"slices"
 	"time"
 
-	"github.com/encodeous/nylon/polyamide/transports/wireguard/device"
+	"github.com/encodeous/nylon/polyamide"
 	"github.com/encodeous/nylon/protocol"
 	"github.com/encodeous/nylon/state"
 	"github.com/goccy/go-yaml"
 	"google.golang.org/protobuf/encoding/protojson"
 )
 
+var ErrIPCStatusHandled = errors.New("nylon IPC response handled")
+
 var pjMarshal = protojson.MarshalOptions{EmitUnpopulated: true}
 var pjUnmarshal = protojson.UnmarshalOptions{DiscardUnknown: true}
 
@@ -63,7 +65,7 @@ func HandleNylonIPC(n *Nylon, rw *bufio.ReadWriter) error {
 		if err := writeResponse(rw, errResponse(err.Error())); err != nil {
 			return err
 		}
-		return device.ErrIPCStatusHandled
+		return ErrIPCStatusHandled
 	}
 
 	// trace is blocking, so we dont dispatch
@@ -75,7 +77,7 @@ func HandleNylonIPC(n *Nylon, rw *bufio.ReadWriter) error {
 		if err := writeResponse(rw, resp); err != nil {
 			return err
 		}
-		return device.ErrIPCStatusHandled
+		return ErrIPCStatusHandled
 	}
 
 	done := make(chan *protocol.IpcResponse, 1)
@@ -105,7 +107,7 @@ func HandleNylonIPC(n *Nylon, rw *bufio.ReadWriter) error {
 	if err := writeResponse(rw, resp); err != nil {
 		return err
 	}
-	return device.ErrIPCStatusHandled
+	return ErrIPCStatusHandled
 }
 
 func handleStatus(n *Nylon, req *protocol.StatusRequest) *protocol.IpcResponse {
@@ -127,8 +129,11 @@ func handleStatus(n *Nylon, req *protocol.StatusRequest) *protocol.IpcResponse {
 	}
 
 	listenPort := uint32(n.LocalCfg.Port)
-	if n.Device != nil {
-		listenPort = uint32(n.Device.ListenPort())
+	// The WireGuard transport knows its bound port once it starts.
+	if n.wireGuard != nil {
+		if port := n.wireGuard.transport.ListenPort(); port != 0 {
+			listenPort = uint32(port)
+		}
 	}
 
 	return &protocol.IpcResponse{
@@ -188,7 +193,7 @@ func buildSeqnos(n *Nylon) []*protocol.SeqnoEntry {
 	return entries
 }
 
-func buildNeighbours(n *Nylon, wgStats map[state.NyPublicKey]device.PeerStatus) []*protocol.NeighbourInfo {
+func buildNeighbours(n *Nylon, wgStats map[state.NyPublicKey]polyamide.PeerStats) []*protocol.NeighbourInfo {
 	ids := slices.Clone(n.GetPeers(n.LocalCfg.Id))
 	slices.Sort(ids)
 	neighbours := make([]*protocol.NeighbourInfo, 0, len(ids))
@@ -316,26 +321,28 @@ func advertisementsForNode(n *Nylon, id state.NodeId) []*protocol.Advertisement
 	return ads
 }
 
-func wireGuardPeerStats(n *Nylon) map[state.NyPublicKey]device.PeerStatus {
-	stats := make(map[state.NyPublicKey]device.PeerStatus)
-	if n.Device == nil {
+func wireGuardPeerStats(n *Nylon) map[state.NyPublicKey]polyamide.PeerStats {
+	stats := make(map[state.NyPublicKey]polyamide.PeerStats)
+	if len(n.Transports) == 0 {
 		return stats
 	}
-	for _, peer := range n.Device.GetPeers() {
-		stat := peer.Status()
-		stats[state.NyPublicKey(stat.PublicKey)] = stat
+	for _, stat := range n.transportPeers() {
+		if key, ok := peerPublicKey(stat.Peer); ok {
+			stats[key] = stat
+		}
 	}
 	return stats
 }
-
-func wireGuardPeerStatsProto(stat device.PeerStatus) *protocol.WireGuardPeerStats {
-	return &protocol.WireGuardPeerStats{
-		LatestHandshakeUnix:         stat.LatestHandshakeTime().UnixNano(),
-		TxBytes:                     stat.TxBytes,
-		RxBytes:                     stat.RxBytes,
-		PersistentKeepaliveInterval: stat.PersistentKeepaliveInterval,
-		Endpoint:                    &stat.Endpoint,
+func wireGuardPeerStatsProto(stat polyamide.PeerStats) *protocol.WireGuardPeerStats {
+	var handshake int64
+	if !stat.LastHandshake.IsZero() {
+		handshake = stat.LastHandshake.UnixNano()
+	}
+	endpoint := ""
+	if stat.PreferredEndpoint != nil {
+		endpoint = stat.PreferredEndpoint.Address()
 	}
+	return &protocol.WireGuardPeerStats{LatestHandshakeUnix: handshake, TxBytes: stat.TxBytes, RxBytes: stat.RxBytes, PersistentKeepaliveInterval: uint32(stat.Keepalive / time.Second), Endpoint: &endpoint}
 }
 
 func sourceProto(source state.Source) *protocol.Source {
@@ -496,7 +503,7 @@ func handleTrace(n *Nylon, rw *bufio.ReadWriter) error {
 		if err := writeResponse(rw, errResponse("tracing not enabled; restart with --dbg-trace-tc")); err != nil {
 			return err
 		}
-		return device.ErrIPCStatusHandled
+		return ErrIPCStatusHandled
 	}
 	// Send initial OK
 	if err := writeResponse(rw, &protocol.IpcResponse{Ok: true}); err != nil {
@@ -513,15 +520,15 @@ func handleTrace(n *Nylon, rw *bufio.ReadWriter) error {
 	for {
 		select {
 		case <-ctx.Done():
-			return device.ErrIPCStatusHandled
+			return ErrIPCStatusHandled
 		case msg := <-ch:
 			if str, ok := msg.(string); ok {
 				resp := &protocol.IpcResponse{
 					Ok:       true,
 					Response: &protocol.IpcResponse_Trace{Trace: &protocol.TraceEvent{Line: str}},
 				}
 				if err := writeResponse(rw, resp); err != nil {
-					return device.ErrIPCStatusHandled
+					return ErrIPCStatusHandled
 				}
 			}
 		}

core/nylon.go

@@ -4,21 +4,20 @@ import (
 	"context"
 	"errors"
 	"log/slog"
-	"net"
 	"net/netip"
 	"os"
 	"os/signal"
 	"path"
 	"reflect"
 	"runtime"
+	"slices"
 	"sync"
 	"sync/atomic"
 	"syscall"
 	"time"
 
 	"github.com/encodeous/nylon/perf"
-	"github.com/encodeous/nylon/polyamide/transports/wireguard/device"
-	"github.com/encodeous/nylon/polyamide/transports/wireguard/tun"
+	"github.com/encodeous/nylon/polyamide"
 	"github.com/encodeous/nylon/state"
 	"github.com/encodeous/tint"
 	"github.com/jellydator/ttlcache/v3"
@@ -58,11 +57,25 @@ type Nylon struct {
 	ConfigPath      string
 
 	// resources
-	Tun           tun.Device
-	wgUapi        net.Listener
-	Interface     string
-	Device        *device.Device
-	observability *observabilityServer
+	Tun       polyamide.HostDevice
+	Interface string
+	// Transports are the registered transports. The first is the default.
+	Transports        []polyamide.Transport
+	linksForNeighbour func(state.NodeId) []LinkConfig
+	observability     *observabilityServer
+	peerHandles       map[state.NodeId]map[polyamide.Transport]polyamide.Peer
+	neighbourLinks    map[state.NodeId][]neighbourLink
+	// peerReceived is when each peer last received a packet. It ranks links without health.
+	peerReceived map[polyamide.Peer]time.Time
+	tunWorkers   sync.WaitGroup
+	// wireGuard serves legacy UAPI. It is nil for custom runtimes.
+	wireGuard       *legacyWireGuard
+	resourceCleanup sync.Once
+	// systemConfigured is set once startup begins configuring the host with
+	// pre-up commands. Only then does cleanup undo host configuration, so a
+	// failed start cannot tear down another instance using the same interface.
+	systemConfigured   bool
+	resourceCleanupErr error
 
 	// only used for debugging & tests
 	AuxConfig map[string]any
@@ -79,7 +92,67 @@ type AppliedSystemState struct {
 	Peers   map[state.NodeId]state.NyPublicKey
 }
 
+// Runtime supplies the transports and host device. Nylon owns them and closes
+// them if initialization fails.
+type Runtime struct {
+	// The first transport is the default. It carries configured endpoints.
+	Transports []polyamide.Transport
+	// Called on the dispatcher. Nil binds configured endpoints to the default transport.
+	LinksForNeighbour func(state.NodeId) []LinkConfig
+	Host              polyamide.HostDevice
+}
+
+func (rt Runtime) validate() error {
+	if len(rt.Transports) == 0 || slices.Contains(rt.Transports, nil) || rt.Host == nil {
+		return errors.New("runtime requires a host device and at least one transport")
+	}
+	for i, transport := range rt.Transports {
+		if slices.Contains(rt.Transports[:i], transport) {
+			return errors.New("runtime contains a duplicate transport")
+		}
+	}
+	return nil
+}
+
+func (rt Runtime) close() {
+	if rt.Host != nil {
+		_ = rt.Host.Close()
+	}
+	for _, transport := range rt.Transports {
+		if transport != nil {
+			_ = transport.Close()
+		}
+	}
+}
+
+// useRuntime attaches a validated runtime.
+func (n *Nylon) useRuntime(rt Runtime) {
+	n.Transports = slices.Clone(rt.Transports)
+	n.linksForNeighbour = rt.LinksForNeighbour
+	n.Tun = rt.Host
+}
+
 func NewNylon(ccfg state.CentralCfg, ncfg state.LocalCfg, logLevel slog.Level, configPath string, aux map[string]any, opts state.NylonOptions, tunables *state.RouterTunables) (*Nylon, error) {
+	return newNylon(ccfg, ncfg, logLevel, configPath, aux, opts, tunables, nil)
+}
+
+// NewNylonWithRuntime uses rt instead of creating the platform runtime.
+func NewNylonWithRuntime(ccfg state.CentralCfg, ncfg state.LocalCfg, logLevel slog.Level, configPath string, aux map[string]any, opts state.NylonOptions, tunables *state.RouterTunables, rt Runtime) (*Nylon, error) {
+	return newNylon(ccfg, ncfg, logLevel, configPath, aux, opts, tunables, &rt)
+}
+
+func newNylon(ccfg state.CentralCfg, ncfg state.LocalCfg, logLevel slog.Level, configPath string, aux map[string]any, opts state.NylonOptions, tunables *state.RouterTunables, supplied *Runtime) (*Nylon, error) {
+	attached := false
+	defer func() {
+		if !attached && supplied != nil {
+			supplied.close()
+		}
+	}()
+	if supplied != nil {
+		if err := supplied.validate(); err != nil {
+			return nil, err
+		}
+	}
 	ctx, cancel := context.WithCancelCause(context.Background())
 
 	dispatch := make(chan func() error, 128)
@@ -155,12 +228,19 @@ func NewNylon(ccfg state.CentralCfg, ncfg state.LocalCfg, logLevel slog.Level, c
 		AuxConfig:        aux,
 		DNSResolver:      dnsResolver,
 		EndpointResolver: state.NewEndpointResolver(dnsResolver),
+		peerHandles:      make(map[state.NodeId]map[polyamide.Transport]polyamide.Peer),
+	}
+	if supplied != nil {
+		n.useRuntime(*supplied)
 	}
 
+	attached = true
 	n.Log.Info("init modules")
 
 	err = n.Init()
 	if err != nil {
+		n.Cancel(err)
+		_ = n.Cleanup()
 		return nil, err
 	}
 	return n, nil
@@ -196,8 +276,7 @@ func (n *Nylon) Init() error {
 		return nylonGc(n)
 	}, n.GcDelay)
 
-	// wireguard configuration
-	err = n.initWireGuard()
+	err = n.initTransport()
 	if err != nil {
 		return err
 	}
@@ -327,16 +406,26 @@ endLoop:
 }
 
 func (n *Nylon) Cleanup() error {
+	n.resourceCleanup.Do(func() { n.resourceCleanupErr = n.cleanup() })
+	return n.resourceCleanupErr
+}
+
+func (n *Nylon) cleanup() error {
+	n.Cancel(context.Canceled)
 	if n.observability != nil {
 		n.observability.close()
 	}
-	n.PingBuf.Stop()
+	if n.PingBuf != nil {
+		n.PingBuf.Stop()
+	}
 	for _, health := range n.prefixHealth {
 		health.monitor.Stop()
 	}
 
+	err := n.cleanupTransport()
 	n.CleanupRouter()
-	n.Trace.Cleanup()
-
-	return n.cleanupWireGuard()
+	if n.Trace != nil && n.Trace.Broadcaster != nil {
+		_ = n.Trace.Cleanup()
+	}
+	return err
 }

core/nylon_apply.go

@@ -46,10 +46,10 @@ func (n *Nylon) ApplyCentralConfig(cfg *state.CentralCfg) (ApplyResult, error) {
 }
 
 func (n *Nylon) SyncApplicationState() error {
-	if n.Device == nil {
+	if len(n.Transports) == 0 {
 		return nil
 	}
-	if err := n.SyncWireGuard(); err != nil {
+	if err := n.SyncTransport(); err != nil {
 		return err
 	}
 	ComputeRoutes(n.RouterState, n)
@@ -105,7 +105,7 @@ func (n *Nylon) reconcileRouterState(next *state.CentralCfg) error {
 			Eps:    make([]state.Endpoint, 0, len(cfg.Endpoints)),
 		}
 		for _, ep := range cfg.Endpoints {
-			stNeigh.Eps = append(stNeigh.Eps, state.NewEndpoint(ep, false, nil, &n.RouterTunables))
+			stNeigh.Eps = append(stNeigh.Eps, state.NewEndpoint(ep, false, &n.RouterTunables))
 		}
 		neighs = append(neighs, stNeigh)
 	}
@@ -165,7 +165,7 @@ func reconcileConfiguredEndpoints(neigh *state.Neighbour, desired []string, t *s
 		if _, ok := seen[address]; ok {
 			continue
 		}
-		eps = append(eps, state.NewEndpoint(address, false, nil, t))
+		eps = append(eps, state.NewEndpoint(address, false, t))
 	}
 	neigh.Eps = eps
 }