yangerd: Add missing files

src/yangerd/.gitignore

@@ -1,3 +1,3 @@
-# Build artifacts
-yangerd
-yangerctl
+# Build artifacts (root-level binaries only; do not match cmd/ source dirs)
+/yangerd
+/yangerctl

src/yangerd/cmd/yangerctl/main.go

@@ -0,0 +1,110 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/kernelkit/infix/src/yangerd/internal/ipc"
+)
+
+const defaultSocket = "/run/yangerd.sock"
+const defaultTimeout = 5 * time.Second
+
+func main() {
+	socket := defaultSocket
+	timeout := defaultTimeout
+
+	args := os.Args[1:]
+	for len(args) > 0 && len(args[0]) > 0 && args[0][0] == '-' {
+		switch args[0] {
+		case "--socket":
+			if len(args) < 2 {
+				die("--socket requires an argument")
+			}
+			socket = args[1]
+			args = args[2:]
+		case "--timeout":
+			if len(args) < 2 {
+				die("--timeout requires an argument")
+			}
+			d, err := time.ParseDuration(args[1])
+			if err != nil {
+				die("invalid duration: %v", err)
+			}
+			timeout = d
+			args = args[2:]
+		default:
+			die("unknown flag: %s", args[0])
+		}
+	}
+
+	if len(args) == 0 {
+		usage()
+	}
+
+	client := ipc.NewClient(socket, timeout)
+
+	switch args[0] {
+	case "get":
+		if len(args) < 2 {
+			die("get requires a path argument")
+		}
+		resp, err := client.Get(args[1])
+		if err != nil {
+			die("get: %v", err)
+		}
+		printResponse(resp)
+	case "dump":
+		resp, err := client.Get("/")
+		if err != nil {
+			die("dump: %v", err)
+		}
+		printResponse(resp)
+	case "health":
+		resp, err := client.Health()
+		if err != nil {
+			die("health: %v", err)
+		}
+		printResponse(resp)
+	default:
+		die("unknown command: %s", args[0])
+	}
+}
+
+func printResponse(resp *ipc.Response) {
+	if resp.Code == 503 {
+		fmt.Fprintf(os.Stderr, "yangerd is starting up\n")
+		os.Exit(3)
+	}
+	if resp.Status == "error" {
+		fmt.Fprintf(os.Stderr, "error %d: %s\n", resp.Code, resp.Message)
+		if resp.Code == 404 {
+			os.Exit(2)
+		}
+		os.Exit(1)
+	}
+
+	var out []byte
+	if resp.Data != nil {
+		out, _ = json.MarshalIndent(json.RawMessage(resp.Data), "", "  ")
+	} else {
+		out, _ = json.MarshalIndent(resp, "", "  ")
+	}
+	fmt.Println(string(out))
+}
+
+func usage() {
+	fmt.Fprintf(os.Stderr, "Usage: yangerctl [--socket path] [--timeout dur] <command> [args]\n\n")
+	fmt.Fprintf(os.Stderr, "Commands:\n")
+	fmt.Fprintf(os.Stderr, "  get <path>   Query a YANG subtree\n")
+	fmt.Fprintf(os.Stderr, "  dump         Dump entire tree\n")
+	fmt.Fprintf(os.Stderr, "  health       Show daemon health\n")
+	os.Exit(1)
+}
+
+func die(format string, args ...interface{}) {
+	fmt.Fprintf(os.Stderr, "yangerctl: "+format+"\n", args...)
+	os.Exit(1)
+}

src/yangerd/cmd/yangerd/main.go

@@ -0,0 +1,427 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"log"
+	"log/slog"
+	"os"
+	"os/signal"
+	"path/filepath"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"syscall"
+	"time"
+
+	"github.com/kernelkit/infix/src/yangerd/internal/bridgebatch"
+	"github.com/kernelkit/infix/src/yangerd/internal/collector"
+	"github.com/kernelkit/infix/src/yangerd/internal/config"
+	"github.com/kernelkit/infix/src/yangerd/internal/dbusmonitor"
+	"github.com/kernelkit/infix/src/yangerd/internal/ethmonitor"
+	"github.com/kernelkit/infix/src/yangerd/internal/frrvty"
+	"github.com/kernelkit/infix/src/yangerd/internal/fswatcher"
+	"github.com/kernelkit/infix/src/yangerd/internal/ipbatch"
+	"github.com/kernelkit/infix/src/yangerd/internal/ipc"
+	"github.com/kernelkit/infix/src/yangerd/internal/iwmonitor"
+	"github.com/kernelkit/infix/src/yangerd/internal/lldpmonitor"
+	"github.com/kernelkit/infix/src/yangerd/internal/monitor"
+	"github.com/kernelkit/infix/src/yangerd/internal/sysreaders"
+	"github.com/kernelkit/infix/src/yangerd/internal/tree"
+	"github.com/kernelkit/infix/src/yangerd/internal/wgquery"
+	"github.com/kernelkit/infix/src/yangerd/internal/zapiwatcher"
+)
+
+// osFileChecker implements iface.FileChecker using the real filesystem.
+type osFileChecker struct{}
+
+func (osFileChecker) Exists(path string) bool {
+	_, err := os.Stat(path)
+	return err == nil
+}
+
+func (osFileChecker) ReadFile(path string) (string, error) {
+	b, err := os.ReadFile(path)
+	if err != nil {
+		return "", err
+	}
+	return string(b), nil
+}
+
+func main() {
+	cfg := config.Load()
+	log.SetFlags(0)
+
+	t := tree.New()
+	ready := &atomic.Bool{}
+
+	srv := ipc.NewServer(t, ready)
+	if err := srv.Listen(cfg.Socket); err != nil {
+		log.Fatalf("listen %s: %v", cfg.Socket, err)
+	}
+	defer os.Remove(cfg.Socket)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	var wg sync.WaitGroup
+	cmd := collector.ExecRunner{}
+	fs := collector.OSFileReader{}
+	collectors := []collector.Collector{
+		collector.NewSystemCollector(cmd, fs, cfg.PollSystem),
+		collector.NewRoutingCollector(cmd, cfg.PollRouting),
+		collector.NewNTPCollector(cmd, cfg.PollNTP),
+		collector.NewHardwareCollector(cmd, fs, cfg.PollHardware, cfg.EnableWifi, cfg.EnableGPS),
+	}
+	pokeCh := make(chan struct{}, len(collectors))
+	collector.RunAll(ctx, &wg, t, collectors, pokeCh)
+
+	inst := collector.DBusInstaller{}
+	t.RegisterProvider("ietf-system:system-state", func() json.RawMessage {
+		live := collector.LiveSystemState(fs)
+		installerOverlay := collector.MergeInstaller(t.GetCached("ietf-system:system-state"), inst)
+		if installerOverlay == nil {
+			return live
+		}
+		var base map[string]json.RawMessage
+		if json.Unmarshal(live, &base) != nil {
+			return live
+		}
+		var overlay map[string]json.RawMessage
+		if json.Unmarshal(installerOverlay, &overlay) != nil {
+			return live
+		}
+		for k, v := range overlay {
+			base[k] = v
+		}
+		merged, err := json.Marshal(base)
+		if err != nil {
+			return live
+		}
+		return merged
+	})
+
+	if data := collector.BootPlatform(fs); data != nil {
+		t.Merge("ietf-system:system-state", data)
+	}
+	if data := collector.BootSoftware(ctx, cmd); data != nil {
+		t.Merge("ietf-system:system-state", data)
+	}
+
+	slogLog := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slogLevel(cfg.LogLevel)}))
+
+	linkBatch, err := ipbatch.New(ctx, slogLog, ipbatch.WithStats(), ipbatch.WithDetails())
+	if err != nil {
+		log.Fatalf("start link batch: %v", err)
+	}
+	defer linkBatch.Close()
+
+	addrBatch, err := ipbatch.New(ctx, slogLog, ipbatch.WithDetails())
+	if err != nil {
+		log.Fatalf("start addr batch: %v", err)
+	}
+	defer addrBatch.Close()
+
+	neighBatch, err := ipbatch.New(ctx, slogLog)
+	if err != nil {
+		log.Fatalf("start neigh batch: %v", err)
+	}
+	defer neighBatch.Close()
+
+	brBatch, err := bridgebatch.New(ctx, slogLog)
+	if err != nil {
+		log.Fatalf("start bridge batch: %v", err)
+	}
+	defer brBatch.Close()
+
+	nlmon := monitor.New(linkBatch, addrBatch, neighBatch, brBatch, t, osFileChecker{}, slogLog)
+
+	ethMon, err := ethmonitor.New(slogLog, cmd)
+	if err != nil {
+		slogLog.Warn("ethmonitor unavailable, continuing without it", "err", err)
+	} else {
+		ethMon.SetOnUpdate(nlmon.SetEthernetData)
+		nlmon.SetEthRefresh(ethMon.RefreshInterface)
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if err := ethMon.Run(ctx); err != nil && ctx.Err() == nil {
+				slogLog.Error("ethmonitor exited", "err", err)
+			}
+		}()
+	}
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		ticker := time.NewTicker(10 * time.Second)
+		defer ticker.Stop()
+		<-nlmon.WaitReady()
+		for {
+			links := nlmon.Links()
+			for ifname, data := range wgquery.Query(links) {
+				nlmon.SetWireguardData(ifname, data)
+			}
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+			}
+		}
+	}()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		ticker := time.NewTicker(cfg.PollSTP)
+		defer ticker.Stop()
+		<-nlmon.WaitReady()
+		for {
+			nlmon.RefreshSTP()
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+			}
+		}
+	}()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			if err := nlmon.Run(ctx); err != nil {
+				if ctx.Err() != nil {
+					return
+				}
+				slogLog.Error("nlmonitor exited, restarting in 1s", "err", err)
+				select {
+				case <-ctx.Done():
+					return
+				case <-time.After(time.Second):
+				}
+			} else {
+				return
+			}
+		}
+	}()
+
+	if cfg.EnableWifi {
+		iwmon := iwmonitor.New(slogLog)
+		iwmon.SetOnUpdate(nlmon.SetWifiData)
+		iwmon.SetOnPhyChange(func() {
+			select {
+			case pokeCh <- struct{}{}:
+			default:
+			}
+		})
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if err := iwmon.Run(ctx); err != nil && ctx.Err() == nil {
+				slogLog.Error("iwmonitor exited", "err", err)
+			}
+		}()
+	}
+
+	if cfg.EnableLLDP {
+		lldpmon := lldpmonitor.New(t, slogLog)
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if err := lldpmon.Run(ctx); err != nil && ctx.Err() == nil {
+				slogLog.Error("lldpmonitor exited", "err", err)
+			}
+		}()
+	}
+
+	zapi := zapiwatcher.New(t, frrvty.New(""), slogLog)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := zapi.Run(ctx); err != nil && ctx.Err() == nil {
+			slogLog.Error("zapiwatcher exited", "err", err)
+		}
+	}()
+
+	if cfg.EnableDHCP || cfg.EnableFirewall {
+		dbusMon := dbusmonitor.New(t, slogLog)
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if err := dbusMon.Run(ctx); err != nil && ctx.Err() == nil {
+				slogLog.Error("dbusmonitor exited", "err", err)
+			}
+		}()
+	}
+
+	fsw, err := fswatcher.New(t, slogLog)
+	if err != nil {
+		log.Fatalf("start fswatcher: %v", err)
+	}
+
+	fwdAgg := sysreaders.NewForwardingAggregator()
+	forwardingPaths := []string{
+		"/proc/sys/net/ipv4/conf/*/forwarding",
+		"/proc/sys/net/ipv6/conf/*/forwarding",
+	}
+	for _, pattern := range forwardingPaths {
+		matches, globErr := filepath.Glob(pattern)
+		if globErr != nil {
+			slogLog.Warn("fswatcher glob failed", "pattern", pattern, "err", globErr)
+			continue
+		}
+		for _, path := range matches {
+			if err := fsw.Watch(path, fswatcher.WatchHandler{
+				TreeKey:  routingTreeKey,
+				ReadFunc: fwdAgg.HandleForwardingChange,
+				Debounce: 100 * time.Millisecond,
+				UseMerge: true,
+			}); err != nil {
+				slogLog.Warn("fswatcher watch failed", "path", path, "err", err)
+			}
+		}
+	}
+	if err := fsw.Watch("/etc/hostname", fswatcher.WatchHandler{
+		TreeKey:  "ietf-system:system",
+		ReadFunc: sysreaders.ReadHostname,
+		Debounce: 200 * time.Millisecond,
+		UseMerge: true,
+	}); err != nil {
+		slogLog.Warn("fswatcher watch failed", "path", "/etc/hostname", "err", err)
+	}
+	if err := fsw.WatchSymlink("/etc/localtime", fswatcher.WatchHandler{
+		TreeKey:  "ietf-system:system",
+		ReadFunc: sysreaders.ReadTimezone,
+		Debounce: 200 * time.Millisecond,
+		UseMerge: true,
+	}); err != nil {
+		slogLog.Warn("fswatcher watch failed", "path", "/etc/localtime", "err", err)
+	}
+	usersHandler := fswatcher.WatchHandler{
+		TreeKey:  "ietf-system:system",
+		ReadFunc: sysreaders.ReadUsers,
+		Debounce: 200 * time.Millisecond,
+		UseMerge: true,
+	}
+	if err := fsw.Watch("/etc/shadow", usersHandler); err != nil {
+		slogLog.Warn("fswatcher watch failed", "path", "/etc/shadow", "err", err)
+	}
+	if err := fsw.WatchDir(sysreaders.SSHDKeysDir, usersHandler); err != nil {
+		slogLog.Warn("fswatcher watch failed", "path", sysreaders.SSHDKeysDir, "err", err)
+	}
+	bootOrderHandler := fswatcher.WatchHandler{
+		TreeKey:  "ietf-system:system-state",
+		ReadFunc: makeBootOrderReader(t, cmd),
+		Debounce: 200 * time.Millisecond,
+		UseMerge: true,
+	}
+	for _, path := range []string{"/mnt/aux/grub/grubenv", "/mnt/aux/uboot.env"} {
+		if err := fsw.Watch(path, bootOrderHandler); err != nil {
+			slogLog.Debug("fswatcher boot-order watch skipped", "path", path, "err", err)
+		}
+	}
+	dnsHandler := fswatcher.WatchHandler{
+		TreeKey:  "ietf-system:system-state",
+		ReadFunc: sysreaders.ReadDNSResolver,
+		Debounce: 200 * time.Millisecond,
+		UseMerge: true,
+	}
+	for _, path := range []string{"/etc/resolv.conf.head", "/var/lib/misc/resolv.conf"} {
+		if err := fsw.WatchSymlink(path, dnsHandler); err != nil {
+			slogLog.Warn("fswatcher dns watch failed", "path", path, "err", err)
+		}
+	}
+	if cfg.EnableContainers {
+		containerHandler := fswatcher.WatchHandler{
+			TreeKey: "infix-containers:containers",
+			ReadFunc: func(_ string) (json.RawMessage, error) {
+				return collector.CollectContainers(cmd, fs), nil
+			},
+			Debounce: 500 * time.Millisecond,
+		}
+		os.MkdirAll("/run/libpod/events", 0755)
+		if err := fsw.WatchDir("/run/libpod/events", containerHandler); err != nil {
+			slogLog.Warn("fswatcher container watch failed", "err", err)
+		}
+	}
+	fsw.InitialRead()
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := fsw.Run(ctx); err != nil && ctx.Err() == nil {
+			slogLog.Error("fswatcher exited", "err", err)
+		}
+	}()
+
+	go func() {
+		<-nlmon.WaitReady()
+		ready.Store(true)
+	}()
+
+	sigCh := make(chan os.Signal, 1)
+	signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT, syscall.SIGHUP)
+
+	go func() {
+		for sig := range sigCh {
+			if sig == syscall.SIGHUP {
+				log.Printf("SIGHUP: triggering immediate re-poll")
+				for range len(collectors) {
+					pokeCh <- struct{}{}
+				}
+				continue
+			}
+			cancel()
+			return
+		}
+	}()
+
+	if err := srv.Serve(ctx); err != nil {
+		log.Fatalf("serve: %v", err)
+	}
+
+	wg.Wait()
+}
+
+func slogLevel(s string) slog.Level {
+	switch strings.ToLower(s) {
+	case "debug":
+		return slog.LevelDebug
+	case "warn", "warning":
+		return slog.LevelWarn
+	case "error":
+		return slog.LevelError
+	default:
+		return slog.LevelInfo
+	}
+}
+
+const routingTreeKey = "ietf-routing:routing"
+
+func makeBootOrderReader(t *tree.Tree, cmd collector.CommandRunner) func(string) (json.RawMessage, error) {
+	return func(_ string) (json.RawMessage, error) {
+		bootOrder := collector.ReadBootOrder(context.TODO(), cmd)
+
+		raw := t.Get("ietf-system:system-state")
+		var state map[string]interface{}
+		if raw != nil {
+			json.Unmarshal(raw, &state)
+		}
+		if state == nil {
+			state = make(map[string]interface{})
+		}
+
+		sw, _ := state["infix-system:software"].(map[string]interface{})
+		if sw == nil {
+			sw = make(map[string]interface{})
+		}
+
+		if bootOrder != nil {
+			sw["boot-order"] = bootOrder
+		} else {
+			delete(sw, "boot-order")
+		}
+
+		return json.Marshal(map[string]interface{}{"infix-system:software": sw})
+	}
+}

src/yangerd/internal/config/config.go

@@ -15,6 +15,7 @@ type Config struct {
 	PollRouting      time.Duration
 	PollNTP          time.Duration
 	PollHardware     time.Duration
+	PollSTP          time.Duration
 	EnableWifi       bool
 	EnableLLDP       bool
 	EnableFirewall   bool
@@ -32,6 +33,7 @@ func Load() *Config {
 		PollRouting:      envDur("YANGERD_POLL_INTERVAL_ROUTING", 10*time.Second),
 		PollNTP:          envDur("YANGERD_POLL_INTERVAL_NTP", 60*time.Second),
 		PollHardware:     envDur("YANGERD_POLL_INTERVAL_HARDWARE", 10*time.Second),
+		PollSTP:          envDur("YANGERD_POLL_INTERVAL_STP", 5*time.Second),
 		EnableWifi:       envBool("YANGERD_ENABLE_WIFI", false),
 		EnableLLDP:       envBool("YANGERD_ENABLE_LLDP", true),
 		EnableFirewall:   envBool("YANGERD_ENABLE_FIREWALL", true),

src/yangerd/internal/ethmonitor/ethmonitor.go

@@ -1,5 +1,9 @@
 // Package ethmonitor subscribes to ethtool genetlink notifications and
 // keeps per-interface ethernet settings updated via a callback.
+//
+// Data is fetched by shelling out to `ethtool --json <ifname>` (matching
+// the Python yanger approach) while genetlink provides reactive change
+// notifications.
 package ethmonitor
 
 import (
@@ -8,40 +12,44 @@ import (
 	"fmt"
 	"log/slog"
 	"net"
+	"strings"
 
-	"github.com/mdlayher/ethtool"
 	"github.com/mdlayher/genetlink"
 	"github.com/mdlayher/netlink"
 )
 
 const (
-	// ETHTOOL_MSG_LINKINFO_NTF is the ethtool generic netlink notification
-	// command for link info changes.
-	ETHTOOL_MSG_LINKINFO_NTF = 28
-	// ETHTOOL_MSG_LINKMODES_NTF is the ethtool generic netlink notification
-	// command for link mode changes.
+	ETHTOOL_MSG_LINKINFO_NTF  = 28
 	ETHTOOL_MSG_LINKMODES_NTF = 29
 
 	ethtoolFamilyName       = "ethtool"
 	ethtoolMonitorGroupName = "monitor"
 
 	nlaHeaderIfindex = 1
+
+	ethtoolSpeedUnknown = (1 << 32) - 1
 )
 
+// CommandRunner executes external commands and returns stdout.
+type CommandRunner interface {
+	Run(ctx context.Context, name string, args ...string) ([]byte, error)
+}
+
 // EthMonitor listens for ethtool genetlink monitor events and updates
 // interface ethernet operational state via a callback.
 type EthMonitor struct {
-	conn     *genetlink.Conn
-	family   genetlink.Family
-	groupID  uint32
-	etClient *ethtool.Client
-	log      *slog.Logger
+	conn    *genetlink.Conn
+	family  genetlink.Family
+	groupID uint32
+	cmd     CommandRunner
+	ctx     context.Context
+	log     *slog.Logger
 	onUpdate func(ifname string, data json.RawMessage)
 }
 
 // New creates an EthMonitor, resolves the ethtool genetlink family,
 // and joins its "monitor" multicast group.
-func New(log *slog.Logger) (*EthMonitor, error) {
+func New(log *slog.Logger, cmd CommandRunner) (*EthMonitor, error) {
 	conn, err := genetlink.Dial(nil)
 	if err != nil {
 		return nil, fmt.Errorf("dial genetlink: %w", err)
@@ -70,18 +78,12 @@ func New(log *slog.Logger) (*EthMonitor, error) {
 		return nil, fmt.Errorf("join ethtool monitor group %d: %w", groupID, err)
 	}
 
-	etClient, err := ethtool.New()
-	if err != nil {
-		_ = conn.Close()
-		return nil, fmt.Errorf("create ethtool client: %w", err)
-	}
-
 	return &EthMonitor{
-		conn:     conn,
-		family:   family,
-		groupID:  groupID,
-		etClient: etClient,
-		log:      log,
+		conn:    conn,
+		family:  family,
+		groupID: groupID,
+		cmd:     cmd,
+		log:     log,
 	}, nil
 }
 
@@ -93,10 +95,8 @@ func (m *EthMonitor) SetOnUpdate(fn func(string, json.RawMessage)) {
 // Run starts the ethtool genetlink receive loop and updates interface
 // ethernet settings when link info or link mode notifications are seen.
 func (m *EthMonitor) Run(ctx context.Context) error {
+	m.ctx = ctx
 	defer func() {
-		if err := m.etClient.Close(); err != nil {
-			m.log.Warn("ethmonitor: close ethtool client", "err", err)
-		}
 		if err := m.conn.Close(); err != nil {
 			m.log.Warn("ethmonitor: close genetlink conn", "err", err)
 		}
@@ -135,35 +135,48 @@ func (m *EthMonitor) RefreshInterface(ifname string) {
 	m.refreshEthernetSettings(ifname)
 }
 
+// ethtoolJSON represents the relevant fields from `ethtool --json <ifname>`.
+type ethtoolJSON struct {
+	Speed              int      `json:"speed"`
+	Duplex             string   `json:"duplex"`
+	Port               string   `json:"port"`
+	AutoNegotiation    bool     `json:"auto-negotiation"`
+	SupportedLinkModes []string `json:"supported-link-modes"`
+	AdvertisedLinkModes []string `json:"advertised-link-modes"`
+}
+
 func (m *EthMonitor) refreshEthernetSettings(ifname string) {
-	iface, err := net.InterfaceByName(ifname)
+	ctx := m.ctx
+	if ctx == nil {
+		ctx = context.Background()
+	}
+
+	out, err := m.cmd.Run(ctx, "ethtool", "--json", ifname)
 	if err != nil {
-		m.log.Warn("ethmonitor: lookup interface", "ifname", ifname, "err", err)
+		m.log.Warn("ethmonitor: run ethtool", "ifname", ifname, "err", err)
 		return
 	}
 
-	ethIface := ethtool.Interface{Index: iface.Index, Name: iface.Name}
-
-	if _, err := m.etClient.LinkInfo(ethIface); err != nil {
-		m.log.Warn("ethmonitor: query link info", "ifname", ifname, "err", err)
+	var results []ethtoolJSON
+	if err := json.Unmarshal(out, &results); err != nil {
+		m.log.Warn("ethmonitor: parse ethtool json", "ifname", ifname, "err", err)
+		return
+	}
+	if len(results) == 0 {
 		return
 	}
 
-	mode, err := m.etClient.LinkMode(ethIface)
-	if err != nil {
-		m.log.Warn("ethmonitor: query link mode", "ifname", ifname, "err", err)
-		return
+	data := results[0]
+	eth, speedBPS := buildEthernetContainer(data)
+
+	// Marshal the result; include interface-level speed as a special key
+	// that mergeAugments will lift onto the interface object.
+	result := map[string]any{"ethernet": eth}
+	if speedBPS > 0 {
+		result["speed"] = fmt.Sprintf("%d", speedBPS)
 	}
 
-	eth := map[string]any{
-		"speed":  speedString(mode.SpeedMegabits),
-		"duplex": duplexString(mode.Duplex),
-		"auto-negotiation": map[string]any{
-			"enable": mode.Autoneg == ethtool.AutonegOn,
-		},
-	}
-
-	raw, err := json.Marshal(eth)
+	raw, err := json.Marshal(result)
 	if err != nil {
 		m.log.Warn("ethmonitor: marshal ethernet settings", "ifname", ifname, "err", err)
 		return
@@ -174,6 +187,55 @@ func (m *EthMonitor) refreshEthernetSettings(ifname string) {
 	}
 }
 
+// buildEthernetContainer builds the ieee802-ethernet-interface:ethernet
+// container and returns (container, interface speed in bits/s or 0).
+func buildEthernetContainer(data ethtoolJSON) (map[string]any, int64) {
+	autoneg := map[string]any{"enable": data.AutoNegotiation}
+	eth := map[string]any{"auto-negotiation": autoneg}
+
+	duplex := strings.ToLower(data.Duplex)
+	if duplex == "full" || duplex == "half" {
+		eth["duplex"] = duplex
+	}
+
+	// Supported PMD types (config-false leaf-list).
+	supported := ethtoolModesToPMD(data.SupportedLinkModes)
+	if len(supported) > 0 {
+		eth["infix-ethernet-interface:supported-pmd-types"] = supported
+	}
+
+	// Advertised PMD types — suppress when identical to supported (default).
+	advertised := ethtoolModesToPMD(data.AdvertisedLinkModes)
+	if len(advertised) > 0 && !stringSliceEqual(advertised, supported) {
+		autoneg["infix-ethernet-interface:advertised-pmd-types"] = advertised
+	}
+
+	// Speed, phy-type, pmd-type.
+	var speedBPS int64
+	speedMbps := data.Speed
+	if speedMbps > 0 && speedMbps < ethtoolSpeedUnknown {
+		speedBPS = int64(speedMbps) * 1_000_000
+
+		// Speed inside the ethernet container (decimal64, Gb/s).
+		eth["speed"] = fmt.Sprintf("%.3f", float64(speedMbps)/1000.0)
+
+		key := linkModeKey{Port: data.Port, SpeedMbps: speedMbps, Duplex: duplex}
+		if mapping, ok := linkModes[key]; ok {
+			eth["phy-type"] = "ieee802-ethernet-phy-type:phy-type-" + mapping.PhyType
+			if mapping.PMDType != "" {
+				eth["pmd-type"] = "ieee802-ethernet-phy-type:pmd-type-" + mapping.PMDType
+			}
+		}
+
+		// Refine pmd-type when exactly one supported mode (specific SFP).
+		if len(supported) == 1 {
+			eth["pmd-type"] = supported[0]
+		}
+	}
+
+	return eth, speedBPS
+}
+
 func extractIfname(data []byte) (string, error) {
 	ad, err := netlink.NewAttributeDecoder(data)
 	if err != nil {
@@ -210,20 +272,118 @@ func extractIfname(data []byte) (string, error) {
 	return "", fmt.Errorf("header ifindex attribute not found")
 }
 
-func duplexString(d ethtool.Duplex) string {
-	switch d {
-	case ethtool.Full:
-		return "full"
-	case ethtool.Half:
-		return "half"
-	default:
-		return "unknown"
-	}
+// linkModeKey is the lookup key for phy-type/pmd-type mapping.
+type linkModeKey struct {
+	Port      string
+	SpeedMbps int
+	Duplex    string
 }
 
-// speedString converts a speed in megabits to a decimal64 string in
-// Gb/s with 3 fraction digits, matching the ieee802-ethernet-interface
-// YANG model's eth-if-speed-type (decimal64, fraction-digits 3, units Gb/s).
-func speedString(megabits int) string {
-	return fmt.Sprintf("%.3f", float64(megabits)/1000.0)
+// linkModeMapping holds the IEEE identity suffixes.
+type linkModeMapping struct {
+	PhyType string
+	PMDType string // empty means "cannot determine from this tuple alone"
+}
+
+// linkModes maps (port, speed, duplex) → (phy-type, pmd-type) per
+// IEEE Std 802.3.2-2025 (ieee802-ethernet-phy-type).
+var linkModes = map[linkModeKey]linkModeMapping{
+	{"Twisted Pair", 10, "full"}:       {"10BASE-T", "10BASE-T"},
+	{"Twisted Pair", 10, "half"}:       {"10BASE-T", "10BASE-T"},
+	{"Twisted Pair", 100, "full"}:      {"100BASE-X", "100BASE-TX"},
+	{"Twisted Pair", 100, "half"}:      {"100BASE-X", "100BASE-TX"},
+	{"Twisted Pair", 1000, "full"}:     {"1000BASE-T", "1000BASE-T"},
+	{"Twisted Pair", 1000, "half"}:     {"1000BASE-T", "1000BASE-T"},
+	{"Twisted Pair", 2500, "full"}:     {"2.5GBASE-T", "2.5GBASE-T"},
+	{"Twisted Pair", 5000, "full"}:     {"5GBASE-T", "5GBASE-T"},
+	{"Twisted Pair", 10000, "full"}:    {"10GBASE-T", "10GBASE-T"},
+	{"Twisted Pair", 25000, "full"}:    {"25GBASE-T", "25GBASE-T"},
+	{"Twisted Pair", 40000, "full"}:    {"40GBASE-T", "40GBASE-T"},
+	{"MII", 10, "full"}:               {"10BASE-T", "10BASE-T"},
+	{"MII", 10, "half"}:               {"10BASE-T", "10BASE-T"},
+	{"MII", 100, "full"}:              {"100BASE-X", "100BASE-TX"},
+	{"MII", 100, "half"}:              {"100BASE-X", "100BASE-TX"},
+	{"FIBRE", 100, "full"}:            {"100BASE-X", ""},
+	{"FIBRE", 1000, "full"}:           {"1000BASE-X", ""},
+	{"FIBRE", 10000, "full"}:          {"10GBASE-R", ""},
+	{"FIBRE", 25000, "full"}:          {"25GBASE-R", ""},
+	{"FIBRE", 40000, "full"}:          {"40GBASE-R", ""},
+	{"FIBRE", 100000, "full"}:         {"100GBASE-R", ""},
+	{"Direct Attach Copper", 10000, "full"}:  {"10GBASE-R", ""},
+	{"Direct Attach Copper", 25000, "full"}:  {"25GBASE-R", "25GBASE-CR"},
+	{"Direct Attach Copper", 40000, "full"}:  {"40GBASE-R", "40GBASE-CR4"},
+	{"Direct Attach Copper", 100000, "full"}: {"100GBASE-R", "100GBASE-CR4"},
+}
+
+// ethtoolToPMD maps kernel link-mode base names to IEEE pmd-type
+// identity suffixes. The kernel reports modes like "1000baseT/Full";
+// we strip the "/Full" or "/Half" suffix before lookup.
+var ethtoolToPMD = map[string]string{
+	"10baseT":          "10BASE-T",
+	"10baseT1L":        "10BASE-T1L",
+	"100baseT":         "100BASE-TX",
+	"100baseT1":        "100BASE-T1",
+	"100baseFX":        "100BASE-FX",
+	"1000baseT":        "1000BASE-T",
+	"1000baseT1":       "1000BASE-T1",
+	"1000baseX":        "1000BASE-LX",
+	"1000baseKX":       "1000BASE-KX",
+	"2500baseT":        "2.5GBASE-T",
+	"2500baseX":        "2.5GBASE-X",
+	"5000baseT":        "5GBASE-T",
+	"10000baseT":       "10GBASE-T",
+	"10000baseSR":      "10GBASE-SR",
+	"10000baseLR":      "10GBASE-LR",
+	"10000baseLRM":     "10GBASE-LRM",
+	"10000baseER":      "10GBASE-ER",
+	"10000baseKR":      "10GBASE-KR",
+	"10000baseKX4":     "10GBASE-KX4",
+	"25000baseCR":      "25GBASE-CR",
+	"25000baseSR":      "25GBASE-SR",
+	"25000baseKR":      "25GBASE-KR",
+	"40000baseCR4":     "40GBASE-CR4",
+	"40000baseSR4":     "40GBASE-SR4",
+	"40000baseLR4":     "40GBASE-LR4",
+	"40000baseKR4":     "40GBASE-KR4",
+	"100000baseCR4":    "100GBASE-CR4",
+	"100000baseSR4":    "100GBASE-SR4",
+	"100000baseLR4_ER4": "100GBASE-LR4",
+	"100000baseKR4":    "100GBASE-KR4",
+}
+
+// ethtoolModesToPMD translates a list of ethtool link-mode strings
+// (e.g. "1000baseT/Full") into deduped, order-preserving PMD identity
+// strings.
+func ethtoolModesToPMD(modes []string) []string {
+	seen := make(map[string]bool)
+	var out []string
+	for _, entry := range modes {
+		base := entry
+		if idx := strings.IndexByte(entry, '/'); idx >= 0 {
+			base = entry[:idx]
+		}
+		pmd, ok := ethtoolToPMD[base]
+		if !ok || seen[pmd] {
+			continue
+		}
+		seen[pmd] = true
+		out = append(out, "ieee802-ethernet-phy-type:pmd-type-"+pmd)
+	}
+	return out
+}
+
+func stringSliceEqual(a, b []string) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	set := make(map[string]bool, len(a))
+	for _, s := range a {
+		set[s] = true
+	}
+	for _, s := range b {
+		if !set[s] {
+			return false
+		}
+	}
+	return true
 }

src/yangerd/internal/monitor/monitor.go

@@ -7,6 +7,8 @@ import (
 	"fmt"
 	"log/slog"
 	"net"
+	"sort"
+	"strings"
 	"sync"
 	"syscall"
 	"time"
@@ -64,6 +66,11 @@ type NLMonitor struct {
 	wireguard map[string]json.RawMessage // ifname → WireGuard peer-status JSON
 
 	lastOperStatus map[string]string
+
+	// stpMu guards lastSTP, a fingerprint of the most recent mstpd STP
+	// query, letting the periodic poll rebuild only on actual change.
+	stpMu   sync.Mutex
+	lastSTP string
 }
 
 // New creates a netlink monitor backed by ip/bridge batch query workers.
@@ -249,6 +256,10 @@ func (m *NLMonitor) initialDump() error {
 	if err != nil {
 		neighRaw = json.RawMessage(`[]`)
 	}
+	mdbRaw, err := m.queryBridge("mdb show")
+	if err != nil {
+		mdbRaw = json.RawMessage(`[]`)
+	}
 
 	m.log.Debug("initialDump", "linkBytes", len(linkRaw), "addrBytes", len(addrRaw), "neighBytes", len(neighRaw))
 	m.validateAddrData("initialDump", addrRaw)
@@ -257,6 +268,9 @@ func (m *NLMonitor) initialDump() error {
 	m.links = linkRaw
 	m.addrs = addrRaw
 	m.neighs = neighRaw
+	for _, bridgeName := range mdbBridgeNames(mdbRaw) {
+		m.mdb[bridgeName] = filterByMDBBridge(mdbRaw, bridgeName)
+	}
 	for _, name := range interfaceNames(linkRaw) {
 		if st, ok := extractOperStatus(filterByIfName(linkRaw, name)); ok {
 			m.lastOperStatus[name] = st
@@ -361,8 +375,8 @@ func (m *NLMonitor) handleMDBUpdate() {
 	}
 
 	m.mu.Lock()
-	for _, bridgeName := range bridgeNames(raw) {
-		m.mdb[bridgeName] = filterByBridge(raw, bridgeName)
+	for _, bridgeName := range mdbBridgeNames(raw) {
+		m.mdb[bridgeName] = filterByMDBBridge(raw, bridgeName)
 	}
 	m.mu.Unlock()
 
@@ -448,6 +462,53 @@ func (m *NLMonitor) rebuild() {
 	m.tree.Set(treeKey, doc)
 }
 
+// RefreshSTP re-queries mstpd and rebuilds only when STP data changed.
+// mstpd's control socket is request/response with no event channel, and
+// the bridge-level root-id settles via BPDU exchange without any netlink
+// event, so STP state must be polled to stay current.  No bridges or an
+// unchanged result costs one cheap mstpd query and no document re-marshal.
+func (m *NLMonitor) RefreshSTP() {
+	links := m.Links()
+	resolver := stpquery.NewLinksIfIndexResolver(links)
+	brSTP, ptSTP := stpquery.Query(links, resolver)
+	if len(brSTP) == 0 && len(ptSTP) == 0 {
+		return
+	}
+
+	fp := stpFingerprint(brSTP, ptSTP)
+	m.stpMu.Lock()
+	changed := fp != m.lastSTP
+	m.lastSTP = fp
+	m.stpMu.Unlock()
+
+	if changed {
+		m.rebuild()
+	}
+}
+
+func stpFingerprint(brSTP, ptSTP map[string]json.RawMessage) string {
+	var b strings.Builder
+	writeSortedRaw(&b, "b", brSTP)
+	writeSortedRaw(&b, "p", ptSTP)
+	return b.String()
+}
+
+func writeSortedRaw(b *strings.Builder, prefix string, m map[string]json.RawMessage) {
+	keys := make([]string, 0, len(m))
+	for k := range m {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	for _, k := range keys {
+		b.WriteString(prefix)
+		b.WriteByte(':')
+		b.WriteString(k)
+		b.WriteByte('=')
+		b.Write(m[k])
+		b.WriteByte('\n')
+	}
+}
+
 // mergeAugments adds ethernet, wifi, and bridge data into the
 // complete ietf-interfaces document produced by iface.Transform().
 func mergeAugments(doc json.RawMessage, ethernet, wifi, fdb, mdb, bridgeSTP, portSTP, wireguard map[string]json.RawMessage) json.RawMessage {
@@ -480,9 +541,20 @@ func mergeAugments(doc json.RawMessage, ethernet, wifi, fdb, mdb, bridgeSTP, por
 		}
 
 		if ethData, ok := ethernet[name]; ok {
-			var ethObj any
-			if err := json.Unmarshal(ethData, &ethObj); err == nil {
-				ifaceObj["ieee802-ethernet-interface:ethernet"] = ethObj
+			var wrapper map[string]json.RawMessage
+			if err := json.Unmarshal(ethData, &wrapper); err == nil {
+				if ethRaw, ok := wrapper["ethernet"]; ok {
+					var ethObj any
+					if err := json.Unmarshal(ethRaw, &ethObj); err == nil {
+						ifaceObj["ieee802-ethernet-interface:ethernet"] = ethObj
+					}
+				}
+				if speedRaw, ok := wrapper["speed"]; ok {
+					var speed string
+					if err := json.Unmarshal(speedRaw, &speed); err == nil {
+						ifaceObj["speed"] = speed
+					}
+				}
 			}
 		}
 
@@ -503,9 +575,8 @@ func mergeAugments(doc json.RawMessage, ethernet, wifi, fdb, mdb, bridgeSTP, por
 
 		if mdbData, ok := mdb[name]; ok {
 			bridgeObj := ensureBridgeAugment(ifaceObj)
-			var mdbObj any
-			if err := json.Unmarshal(mdbData, &mdbObj); err == nil {
-				bridgeObj["mdb"] = mdbObj
+			if mf := transformMDB(mdbData); mf != nil {
+				bridgeObj["multicast-filters"] = mf
 			}
 		}
 
@@ -982,3 +1053,119 @@ func filterByBridge(raw json.RawMessage, bridgeName string) json.RawMessage {
 	}
 	return json.RawMessage(out)
 }
+
+// parseMDBEntries extracts the flat list of MDB entries from the
+// bridge batch output format: [{"mdb":[{entries...}],"router":{}}]
+func parseMDBEntries(raw json.RawMessage) []map[string]any {
+	var wrappers []map[string]json.RawMessage
+	if err := json.Unmarshal(raw, &wrappers); err != nil {
+		return nil
+	}
+
+	var all []map[string]any
+	for _, w := range wrappers {
+		mdbRaw, ok := w["mdb"]
+		if !ok {
+			continue
+		}
+		var entries []map[string]any
+		if err := json.Unmarshal(mdbRaw, &entries); err != nil {
+			continue
+		}
+		all = append(all, entries...)
+	}
+	return all
+}
+
+func mdbBridgeNames(raw json.RawMessage) []string {
+	entries := parseMDBEntries(raw)
+	seen := make(map[string]struct{}, len(entries))
+	var names []string
+	for _, e := range entries {
+		name, _ := e["dev"].(string)
+		if name == "" {
+			continue
+		}
+		if _, ok := seen[name]; ok {
+			continue
+		}
+		seen[name] = struct{}{}
+		names = append(names, name)
+	}
+	return names
+}
+
+func filterByMDBBridge(raw json.RawMessage, bridgeName string) json.RawMessage {
+	entries := parseMDBEntries(raw)
+	var filtered []map[string]any
+	for _, e := range entries {
+		if dev, _ := e["dev"].(string); dev == bridgeName {
+			filtered = append(filtered, e)
+		}
+	}
+	if len(filtered) == 0 {
+		return json.RawMessage(`[]`)
+	}
+	out, err := json.Marshal(filtered)
+	if err != nil {
+		return json.RawMessage(`[]`)
+	}
+	return json.RawMessage(out)
+}
+
+func transformMDB(raw json.RawMessage) map[string]any {
+	var entries []map[string]any
+	if err := json.Unmarshal(raw, &entries); err != nil || len(entries) == 0 {
+		return nil
+	}
+
+	type portEntry struct {
+		Port  string `json:"port"`
+		State string `json:"state"`
+	}
+
+	groups := make(map[string][]portEntry)
+	var order []string
+
+	for _, e := range entries {
+		grp, _ := e["grp"].(string)
+		port, _ := e["port"].(string)
+		state, _ := e["state"].(string)
+		if grp == "" || port == "" {
+			continue
+		}
+
+		if _, seen := groups[grp]; !seen {
+			order = append(order, grp)
+		}
+		groups[grp] = append(groups[grp], portEntry{
+			Port:  port,
+			State: mdbStateToYANG(state),
+		})
+	}
+
+	if len(groups) == 0 {
+		return nil
+	}
+
+	filters := make([]map[string]any, 0, len(order))
+	for _, grp := range order {
+		filters = append(filters, map[string]any{
+			"group": grp,
+			"ports": groups[grp],
+		})
+	}
+
+	return map[string]any{"multicast-filter": filters}
+}
+
+func mdbStateToYANG(state string) string {
+	switch state {
+	case "temp":
+		return "temporary"
+	case "permanent":
+		return "permanent"
+	default:
+		return state
+	}
+}

src/yangerd/internal/monitor/monitor_test.go

@@ -338,7 +338,7 @@ func TestMergeAugments(t *testing.T) {
 	doc := json.RawMessage(`{"interface":[{"name":"eth0","type":"infix-if-type:ethernet"},{"name":"br0","type":"infix-if-type:bridge"}]}`)
 
 	eth := map[string]json.RawMessage{
-		"eth0": json.RawMessage(`{"speed":1000,"duplex":"full"}`),
+		"eth0": json.RawMessage(`{"ethernet":{"speed":"1.000","duplex":"full"},"speed":"1000000000"}`),
 	}
 	fdb := map[string]json.RawMessage{
 		"br0": json.RawMessage(`[{"mac":"00:11:22:33:44:55"}]`),
@@ -390,3 +390,69 @@ func TestTreeKey(t *testing.T) {
 		t.Fatalf("treeKey = %q, want %q", treeKey, "ietf-interfaces:interfaces")
 	}
 }
+
+func TestTransformMDB(t *testing.T) {
+	// transformMDB receives the output of filterByMDBBridge: a flat array of entries
+	raw := json.RawMessage(`[{"dev":"br0","port":"e3","grp":"224.1.1.1","state":"temp"},{"dev":"br0","port":"e4","grp":"224.1.1.1","state":"permanent"},{"dev":"br0","port":"e3","grp":"ff02::6a","state":"temp"}]`)
+
+	result := transformMDB(raw)
+	if result == nil {
+		t.Fatal("expected non-nil result")
+	}
+
+	filters, ok := result["multicast-filter"].([]map[string]any)
+	if !ok {
+		t.Fatalf("unexpected type: %T", result["multicast-filter"])
+	}
+	if len(filters) != 2 {
+		t.Fatalf("expected 2 filters, got %d", len(filters))
+	}
+
+	if filters[0]["group"] != "224.1.1.1" {
+		t.Fatalf("unexpected group: %v", filters[0]["group"])
+	}
+
+	out, _ := json.Marshal(result)
+	if !json.Valid(out) {
+		t.Fatalf("invalid JSON: %s", out)
+	}
+}
+
+func TestTransformMDBEmpty(t *testing.T) {
+	if transformMDB(json.RawMessage(`[]`)) != nil {
+		t.Fatal("expected nil for empty")
+	}
+	if transformMDB(json.RawMessage(`[{"dev":"br0","port":"br0","grp":"ff02::6a","state":"temp"}]`)) == nil {
+		t.Fatal("expected non-nil for router-only entry")
+	}
+}
+
+func TestMDBBridgeNames(t *testing.T) {
+	raw := json.RawMessage(`[{"mdb":[{"dev":"br0","port":"e3","grp":"224.1.1.1","state":"temp"}],"router":{}},{"mdb":[{"dev":"br1","port":"e5","grp":"ff02::1","state":"temp"}],"router":{}}]`)
+	names := mdbBridgeNames(raw)
+	if len(names) != 2 || names[0] != "br0" || names[1] != "br1" {
+		t.Fatalf("unexpected names: %v", names)
+	}
+}
+
+func TestSTPFingerprintDeterministic(t *testing.T) {
+	br1 := map[string]json.RawMessage{
+		"br0": json.RawMessage(`{"root-id":"1.000.00:a0:85:00:01:00"}`),
+		"br1": json.RawMessage(`{"root-id":"2.000.00:a0:85:00:02:00"}`),
+	}
+	br2 := map[string]json.RawMessage{
+		"br1": json.RawMessage(`{"root-id":"2.000.00:a0:85:00:02:00"}`),
+		"br0": json.RawMessage(`{"root-id":"1.000.00:a0:85:00:01:00"}`),
+	}
+	pt1 := map[string]json.RawMessage{"e1": json.RawMessage(`{"state":"forwarding"}`)}
+	pt2 := map[string]json.RawMessage{"e1": json.RawMessage(`{"state":"forwarding"}`)}
+
+	if stpFingerprint(br1, pt1) != stpFingerprint(br2, pt2) {
+		t.Fatal("fingerprint must be independent of map iteration order")
+	}
+
+	br2["br0"] = json.RawMessage(`{"root-id":"8.000.00:a0:85:00:01:00"}`)
+	if stpFingerprint(br1, pt1) == stpFingerprint(br2, pt2) {
+		t.Fatal("fingerprint must change when STP data changes")
+	}
+}

src/yangerd/internal/zapiwatcher/zapiwatcher.go

@@ -6,9 +6,9 @@ import (
 	"fmt"
 	"log/slog"
 	"net"
-	"sort"
+	"regexp"
 	"strconv"
-	"sync"
+	"strings"
 	"time"
 
 	"github.com/kernelkit/infix/src/yangerd/internal/backoff"
@@ -16,8 +16,23 @@ import (
 	"github.com/kernelkit/infix/src/yangerd/internal/zapi"
 )
 
-const zapiSocketPath = "/var/run/frr/zserv.api"
+const (
+	zapiSocketPath = "/var/run/frr/zserv.api"
+	routingTreeKey = "ietf-routing:routing"
 
+	// debounceDelay coalesces a burst of ZAPI route notifications into a
+	// single RIB read.  FRR emits many RouteAdd/Del messages while a
+	// protocol converges; without debouncing we would re-read the table
+	// dozens of times in a few milliseconds.
+	debounceDelay = 200 * time.Millisecond
+)
+
+// subscribeTypes are the route types we ask zebra to redistribute.  We do
+// not use the route payloads themselves -- redistribution only ever
+// delivers the selected route per destination and does not reliably send
+// a delete when a route is superseded.  The subscription exists purely so
+// zebra notifies us that *something* changed; the authoritative table is
+// then read from zebra's vty socket (see RouteQuerier).
 var subscribeTypes = []zapi.RouteType{
 	zapi.RouteKernel,
 	zapi.RouteConnect,
@@ -27,39 +42,45 @@ var subscribeTypes = []zapi.RouteType{
 	zapi.RouteOSPF,
 }
 
-var routeTypeToProtocol = map[zapi.RouteType]string{
-	zapi.RouteKernel:  "infix-routing:kernel",
-	zapi.RouteConnect: "ietf-routing:direct",
-	zapi.RouteLocal:   "ietf-routing:direct",
-	zapi.RouteStatic:  "ietf-routing:static",
-	zapi.RouteOSPF:    "ietf-ospf:ospfv2",
-	zapi.RouteRIP:     "ietf-rip:rip",
-}
-
-// routeEntry pairs a ZAPI route with the wall-clock time it was first
-// received.  The timestamp is used for the YANG last-updated leaf.
-type routeEntry struct {
-	route      *zapi.Route
-	receivedAt time.Time
+// RouteQuerier runs a "show ... json" command against FRR and returns its
+// raw output.  Production code uses an frrvty.Client (zebra's vty socket);
+// tests inject a fake.
+type RouteQuerier interface {
+	Query(ctx context.Context, command string) ([]byte, error)
 }
 
+// ZAPIWatcher keeps the operational RIB (ietf-routing:routing/ribs) in
+// sync with FRR.  It does NOT reconstruct routes from the ZAPI stream:
+// the ZAPI socket is used only as a change trigger, and the full routing
+// table -- every candidate per destination, with FRR's own
+// selected/installed flags, exactly as "show ip route" renders it -- is
+// read from zebra's vty socket on each change.  Because every refresh is
+// a complete snapshot, a route removed from zebra simply disappears; we
+// never depend on receiving a ZAPI delete.
 type ZAPIWatcher struct {
-	tree   *tree.Tree
-	log    *slog.Logger
-	mu     sync.Mutex
-	routes map[string]*routeEntry
+	tree    *tree.Tree
+	querier RouteQuerier
+	log     *slog.Logger
+	refresh chan struct{}
 }
 
-const routingTreeKey = "ietf-routing:routing"
-
-func New(t *tree.Tree, log *slog.Logger) *ZAPIWatcher {
+func New(t *tree.Tree, querier RouteQuerier, log *slog.Logger) *ZAPIWatcher {
 	if log == nil {
 		log = slog.Default()
 	}
-	return &ZAPIWatcher{tree: t, log: log, routes: make(map[string]*routeEntry)}
+	return &ZAPIWatcher{
+		tree:    t,
+		querier: querier,
+		log:     log,
+		refresh: make(chan struct{}, 1),
+	}
 }
 
 func (w *ZAPIWatcher) Run(ctx context.Context) error {
+	// The refresh worker owns all writes to the tree and runs for the
+	// lifetime of the watcher, independent of the ZAPI connection.
+	go w.refreshLoop(ctx)
+
 	bo := backoff.Default()
 	delay := bo.Initial
 
@@ -81,6 +102,10 @@ func (w *ZAPIWatcher) Run(ctx context.Context) error {
 		delay = bo.Initial
 		w.log.Info("zapi watcher: connected", "socket", zapiSocketPath)
 
+		// Read the current table now that we are subscribed, so we have
+		// data even if no further events arrive.
+		w.triggerRefresh()
+
 		err = w.processMessages(ctx, conn)
 		_ = conn.Close()
 		if ctx.Err() != nil {
@@ -88,7 +113,6 @@ func (w *ZAPIWatcher) Run(ctx context.Context) error {
 		}
 
 		w.log.Warn("zapi watcher: disconnected", "err", err)
-		w.clearAllRoutes()
 	}
 }
 
@@ -127,6 +151,9 @@ func (w *ZAPIWatcher) connect(ctx context.Context) (net.Conn, error) {
 	return conn, nil
 }
 
+// processMessages drains the ZAPI stream.  We only care *that* a route
+// changed, not what changed -- each route add/delete triggers a debounced
+// re-read of the full table.
 func (w *ZAPIWatcher) processMessages(ctx context.Context, conn net.Conn) error {
 	for {
 		select {
@@ -135,235 +162,310 @@ func (w *ZAPIWatcher) processMessages(ctx context.Context, conn net.Conn) error
 		default:
 		}
 
-		hdr, body, err := zapi.ReadMessage(conn)
+		hdr, _, err := zapi.ReadMessage(conn)
 		if err != nil {
 			return fmt.Errorf("read message: %w", err)
 		}
 
-		w.handleMessage(hdr, body)
+		switch hdr.Command {
+		case zapi.CmdRedistRouteAdd, zapi.CmdRedistRouteDel:
+			w.log.Debug("zapi watcher: route change", "cmd", hdr.Command, "vrf", hdr.VrfID)
+			w.triggerRefresh()
+		}
 	}
 }
 
-func (w *ZAPIWatcher) handleMessage(hdr zapi.Header, body []byte) {
-	w.log.Debug("zapi watcher: message", "cmd", hdr.Command, "vrf", hdr.VrfID, "len", hdr.Length)
+// triggerRefresh requests a table re-read.  The buffered channel collapses
+// multiple pending requests into one.
+func (w *ZAPIWatcher) triggerRefresh() {
+	select {
+	case w.refresh <- struct{}{}:
+	default:
+	}
+}
 
-	switch hdr.Command {
-	case zapi.CmdRedistRouteAdd:
-		route, err := zapi.DecodeRouteLog(body, w.log)
-		if err != nil {
-			w.log.Warn("zapi watcher: decode route add", "err", err, "bodyLen", len(body))
+func (w *ZAPIWatcher) refreshLoop(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
 			return
+		case <-w.refresh:
 		}
-		w.log.Debug("zapi watcher: route add",
-			"type", route.Type,
-			"prefix", route.Prefix.String(),
-			"distance", route.Distance,
-			"metric", route.Metric,
-			"nexthops", len(route.Nexthops),
-			"msg", route.Message,
-		)
-		w.addRoute(route)
-	case zapi.CmdRedistRouteDel:
-		route, err := zapi.DecodeRoute(body)
-		if err != nil {
-			w.log.Warn("zapi watcher: decode route del", "err", err, "bodyLen", len(body))
+
+		// Let a burst of notifications settle before reading.
+		select {
+		case <-ctx.Done():
 			return
+		case <-time.After(debounceDelay):
 		}
-		w.log.Debug("zapi watcher: route del",
-			"type", route.Type,
-			"prefix", route.Prefix.String(),
-		)
-		w.deleteRoute(route)
-	}
-}
-
-// routeKey returns a unique map key for a redistributed route.
-// FRR's redistribution sends a single RouteAdd per prefix+protocol —
-// nexthop changes are updates (no preceding RouteDel), so the key
-// must NOT include nexthops or distance.
-func routeKey(route *zapi.Route) string {
-	return ribName(route.Prefix) + ":" + route.Prefix.String() + ":" + routeProtocol(route.Type)
-}
-
-func (w *ZAPIWatcher) addRoute(route *zapi.Route) {
-	key := routeKey(route)
-	now := time.Now()
-
-	w.mu.Lock()
-	if existing, ok := w.routes[key]; ok {
-		now = existing.receivedAt
-	}
-	w.routes[key] = &routeEntry{route: route, receivedAt: now}
-	routeCount := len(w.routes)
-	w.mu.Unlock()
-
-	w.log.Debug("zapi watcher: stored route", "key", key, "totalRoutes", routeCount)
-	w.writeRibs()
-}
-
-func (w *ZAPIWatcher) deleteRoute(route *zapi.Route) {
-	key := routeKey(route)
-
-	w.mu.Lock()
-	delete(w.routes, key)
-	w.mu.Unlock()
-
-	w.writeRibs()
-}
-
-func (w *ZAPIWatcher) clearAllRoutes() {
-	w.mu.Lock()
-	w.routes = make(map[string]*routeEntry)
-	w.mu.Unlock()
-
-	w.writeRibs()
-}
-
-// destKey returns the grouping key for active route computation:
-// "rib:prefix" — all routes sharing the same destination compete.
-func destKey(route *zapi.Route) string {
-	return ribName(route.Prefix) + ":" + route.Prefix.String()
-}
-
-func (w *ZAPIWatcher) writeRibs() {
-	w.mu.Lock()
-
-	bestDist := make(map[string]uint8)
-	for _, entry := range w.routes {
-		dk := destKey(entry.route)
-		if d, ok := bestDist[dk]; !ok || entry.route.Distance < d {
-			bestDist[dk] = entry.route.Distance
+		// Drain a request that arrived during the debounce window; the
+		// upcoming read already reflects it.
+		select {
+		case <-w.refresh:
+		default:
 		}
+
+		w.writeRibs(ctx)
 	}
+}
 
-	// Collect entries into a slice for deterministic ordering.
-	allEntries := make([]*routeEntry, 0, len(w.routes))
-	for _, entry := range w.routes {
-		allEntries = append(allEntries, entry)
+// writeRibs reads the full IPv4 and IPv6 routing tables from zebra and
+// replaces the ribs subtree.  On a query error it leaves the previous
+// data untouched rather than blanking the table.
+func (w *ZAPIWatcher) writeRibs(ctx context.Context) {
+	ipv4, err := w.collectRoutes(ctx, "ipv4")
+	if err != nil {
+		w.log.Warn("zapi watcher: read ipv4 routes", "err", err)
+		return
 	}
-
-	// Sort by destination prefix so routes for the same destination are
-	// grouped together.  Within the same prefix, lowest distance first
-	// (active route on top).
-	sort.Slice(allEntries, func(i, j int) bool {
-		a, b := allEntries[i].route, allEntries[j].route
-		ap, bp := a.Prefix.String(), b.Prefix.String()
-		if ap != bp {
-			return ap < bp
-		}
-		return a.Distance < b.Distance
-	})
-
-	ipv4Routes := make([]json.RawMessage, 0, len(allEntries))
-	ipv6Routes := make([]json.RawMessage, 0, len(allEntries))
-
-	for _, entry := range allEntries {
-		dk := destKey(entry.route)
-		active := entry.route.Distance == bestDist[dk]
-		routeData := transformRoute(entry.route, active, entry.receivedAt)
-
-		if entry.route.Prefix.IP.To4() != nil {
-			ipv4Routes = append(ipv4Routes, routeData)
-		} else {
-			ipv6Routes = append(ipv6Routes, routeData)
-		}
+	ipv6, err := w.collectRoutes(ctx, "ipv6")
+	if err != nil {
+		w.log.Warn("zapi watcher: read ipv6 routes", "err", err)
+		return
 	}
-	w.mu.Unlock()
 
 	ribs := map[string]any{
 		"rib": []map[string]any{
 			{
 				"name":           "ipv4",
 				"address-family": "ietf-routing:ipv4",
-				"routes": map[string]any{
-					"route": ipv4Routes,
-				},
+				"routes":         map[string]any{"route": ipv4},
 			},
 			{
 				"name":           "ipv6",
 				"address-family": "ietf-routing:ipv6",
-				"routes": map[string]any{
-					"route": ipv6Routes,
-				},
+				"routes":         map[string]any{"route": ipv6},
 			},
 		},
 	}
 
-	ribsJSON, err := json.Marshal(map[string]any{"ribs": ribs})
+	data, err := json.Marshal(map[string]any{"ribs": ribs})
 	if err != nil {
 		w.log.Error("zapi watcher: marshal ribs", "err", err)
 		return
 	}
 
-	w.tree.Merge(routingTreeKey, ribsJSON)
+	w.tree.Merge(routingTreeKey, data)
 }
 
-func ribName(prefix net.IPNet) string {
-	if prefix.IP.To4() != nil {
-		return "ipv4"
-	}
-	return "ipv6"
-}
-
-func transformRoute(route *zapi.Route, active bool, receivedAt time.Time) json.RawMessage {
-	isIPv4 := route.Prefix.IP.To4() != nil
-
-	addrKey := "ietf-ipv6-unicast-routing:address"
-	dpKey := "ietf-ipv6-unicast-routing:destination-prefix"
-	if isIPv4 {
-		addrKey = "ietf-ipv4-unicast-routing:address"
-		dpKey = "ietf-ipv4-unicast-routing:destination-prefix"
+// collectRoutes runs "show ip route json" / "show ipv6 route json" and
+// transforms every entry into an ietf-routing route node.
+func (w *ZAPIWatcher) collectRoutes(ctx context.Context, family string) ([]json.RawMessage, error) {
+	command := "show ip route json"
+	if family == "ipv6" {
+		command = "show ipv6 route json"
 	}
 
-	nextHops := make([]map[string]any, 0, len(route.Nexthops))
-	for _, nh := range route.Nexthops {
-		hop := make(map[string]any)
-
-		if len(nh.Gate) > 0 && !nh.Gate.IsUnspecified() {
-			hop[addrKey] = nh.Gate.String()
-		}
-
-		if nh.Ifindex > 0 {
-			ifi, err := net.InterfaceByIndex(int(nh.Ifindex))
-			if err == nil && ifi != nil && ifi.Name != "" {
-				hop["outgoing-interface"] = ifi.Name
-			} else {
-				hop["outgoing-interface"] = strconv.FormatUint(uint64(nh.Ifindex), 10)
-			}
-		}
-
-		if len(hop) > 0 {
-			nextHops = append(nextHops, hop)
-		}
-	}
-
-	routeNode := map[string]any{
-		dpKey:              route.Prefix.String(),
-		"source-protocol":  routeProtocol(route.Type),
-		"route-preference": route.Distance,
-		"last-updated":     receivedAt.Format(time.RFC3339),
-		"next-hop": map[string]any{
-			"next-hop-list": map[string]any{
-				"next-hop": nextHops,
-			},
-		},
-	}
-
-	if active {
-		routeNode["active"] = []any{nil}
-	}
-
-	encoded, err := json.Marshal(routeNode)
+	out, err := w.querier.Query(ctx, command)
 	if err != nil {
-		return json.RawMessage(`{}`)
+		return nil, err
 	}
 
-	return json.RawMessage(encoded)
+	// FRR prints "{}" for an empty table; otherwise a map of
+	// prefix -> [route, ...] (multiple candidates per prefix).
+	var table map[string][]map[string]any
+	if err := json.Unmarshal(out, &table); err != nil {
+		return nil, fmt.Errorf("parse %q: %w", command, err)
+	}
+
+	now := time.Now()
+	routes := make([]json.RawMessage, 0, len(table))
+	for prefix, entries := range table {
+		if !strings.Contains(prefix, "/") {
+			continue
+		}
+		for _, entry := range entries {
+			routes = append(routes, transformRoute(family, prefix, entry, now))
+		}
+	}
+	return routes, nil
 }
 
-func routeProtocol(rt zapi.RouteType) string {
-	if protocol, ok := routeTypeToProtocol[rt]; ok {
-		return protocol
+// protocolMap maps FRR's protocol names to IETF routing-protocol
+// identities.  Unknown protocols fall back to kernel so they still
+// validate against the model.
+var protocolMap = map[string]string{
+	"kernel":    "infix-routing:kernel",
+	"connected": "ietf-routing:direct",
+	"local":     "ietf-routing:direct",
+	"static":    "ietf-routing:static",
+	"ospf":      "ietf-ospf:ospfv2",
+	"ospf6":     "ietf-ospf:ospfv3",
+	"rip":       "ietf-rip:rip",
+	"ripng":     "ietf-rip:rip",
+}
+
+func protocolName(frr string) string {
+	if p, ok := protocolMap[frr]; ok {
+		return p
 	}
 	return "infix-routing:kernel"
 }
+
+// transformRoute converts one FRR JSON route entry into an ietf-routing
+// route node.  It mirrors the legacy yanger ietf_routing.py:add_protocol.
+func transformRoute(family, prefixKey string, route map[string]any, now time.Time) json.RawMessage {
+	addrKey := "ietf-ipv4-unicast-routing:address"
+	dpKey := "ietf-ipv4-unicast-routing:destination-prefix"
+	nhAddrKey := "ietf-ipv4-unicast-routing:next-hop-address"
+	hostLen := "32"
+	if family == "ipv6" {
+		addrKey = "ietf-ipv6-unicast-routing:address"
+		dpKey = "ietf-ipv6-unicast-routing:destination-prefix"
+		nhAddrKey = "ietf-ipv6-unicast-routing:next-hop-address"
+		hostLen = "128"
+	}
+
+	dst := stringField(route, "prefix")
+	if dst == "" {
+		dst = prefixKey
+	}
+	if !strings.Contains(dst, "/") {
+		plen := hostLen
+		if v, ok := route["prefixLen"]; ok {
+			plen = strconv.Itoa(toInt(v))
+		}
+		dst = dst + "/" + plen
+	}
+
+	frr := stringField(route, "protocol")
+
+	node := map[string]any{
+		dpKey:              dst,
+		"source-protocol":  protocolName(frr),
+		"route-preference": toInt(route["distance"]),
+		"last-updated":     now.Add(-parseUptime(stringField(route, "uptime"))).Format(time.RFC3339),
+	}
+
+	// Metric is modelled only for OSPF and RIP routes.
+	switch {
+	case strings.Contains(frr, "ospf"):
+		node["ietf-ospf:metric"] = toInt(route["metric"])
+	case strings.Contains(frr, "rip"):
+		node["ietf-rip:metric"] = toInt(route["metric"])
+	}
+
+	// "selected" is FRR's own best-path decision -- the '>' in
+	// "show ip route".  active is a presence leaf, encoded as [null].
+	if boolField(route, "selected") {
+		node["active"] = []any{nil}
+	}
+
+	installed := boolField(route, "installed")
+
+	nextHops := make([]map[string]any, 0)
+	if hops, ok := route["nexthops"].([]any); ok {
+		for _, h := range hops {
+			hop, ok := h.(map[string]any)
+			if !ok {
+				continue
+			}
+			nh := map[string]any{}
+			if ip := stringField(hop, "ip"); ip != "" {
+				nh[addrKey] = ip
+			} else if ifn := stringField(hop, "interfaceName"); ifn != "" {
+				nh["outgoing-interface"] = ifn
+			}
+			// zebra marks the nexthop programmed into the FIB with
+			// "fib":true (see zebra/zebra_vty.c).
+			if installed && boolField(hop, "fib") {
+				nh["infix-routing:installed"] = []any{nil}
+			}
+			if len(nh) > 0 {
+				nextHops = append(nextHops, nh)
+			}
+		}
+	}
+
+	if len(nextHops) > 0 {
+		node["next-hop"] = map[string]any{
+			"next-hop-list": map[string]any{
+				"next-hop": nextHops,
+			},
+		}
+	} else {
+		nh := map[string]any{}
+		switch frr {
+		case "blackhole":
+			nh["special-next-hop"] = "blackhole"
+		case "unreachable":
+			nh["special-next-hop"] = "unreachable"
+		default:
+			if ifn := stringField(route, "interfaceName"); ifn != "" {
+				nh["outgoing-interface"] = ifn
+			}
+			if gw := stringField(route, "nexthop"); gw != "" {
+				nh[nhAddrKey] = gw
+			}
+		}
+		node["next-hop"] = nh
+	}
+
+	encoded, err := json.Marshal(node)
+	if err != nil {
+		return json.RawMessage(`{}`)
+	}
+	return encoded
+}
+
+func stringField(m map[string]any, key string) string {
+	if s, ok := m[key].(string); ok {
+		return s
+	}
+	return ""
+}
+
+func boolField(m map[string]any, key string) bool {
+	b, _ := m[key].(bool)
+	return b
+}
+
+func toInt(v any) int {
+	switch n := v.(type) {
+	case float64:
+		return int(n)
+	case int:
+		return n
+	case int64:
+		return int(n)
+	case json.Number:
+		i, _ := n.Int64()
+		return int(i)
+	case string:
+		i, _ := strconv.Atoi(n)
+		return i
+	}
+	return 0
+}
+
+// FRR uptime string formats (frrtime), ported from yanger's
+// uptime2datetime: "HH:MM:SS", "XdXXhXXm", "XXwXdXXh".
+var (
+	uptimeHMS = regexp.MustCompile(`^(\d{2}):(\d{2}):(\d{2})$`)
+	uptimeDHM = regexp.MustCompile(`^(\d+)d(\d{2})h(\d{2})m$`)
+	uptimeWDH = regexp.MustCompile(`^(\d{2})w(\d)d(\d{2})h$`)
+)
+
+// parseUptime converts an FRR uptime string into a duration.  The
+// last-updated leaf is then computed as now-uptime.  Unrecognised input
+// yields zero (i.e. last-updated == now).
+func parseUptime(s string) time.Duration {
+	atoi := func(x string) int { n, _ := strconv.Atoi(x); return n }
+
+	if m := uptimeHMS.FindStringSubmatch(s); m != nil {
+		return time.Duration(atoi(m[1]))*time.Hour +
+			time.Duration(atoi(m[2]))*time.Minute +
+			time.Duration(atoi(m[3]))*time.Second
+	}
+	if m := uptimeDHM.FindStringSubmatch(s); m != nil {
+		return time.Duration(atoi(m[1]))*24*time.Hour +
+			time.Duration(atoi(m[2]))*time.Hour +
+			time.Duration(atoi(m[3]))*time.Minute
+	}
+	if m := uptimeWDH.FindStringSubmatch(s); m != nil {
+		return time.Duration(atoi(m[1]))*7*24*time.Hour +
+			time.Duration(atoi(m[2]))*24*time.Hour +
+			time.Duration(atoi(m[3]))*time.Hour
+	}
+	return 0
+}

src/yangerd/internal/zapiwatcher/zapiwatcher_test.go

@@ -1,427 +1,269 @@
 package zapiwatcher
 
 import (
+	"context"
 	"encoding/json"
-	"net"
 	"testing"
 	"time"
 
 	"github.com/kernelkit/infix/src/yangerd/internal/tree"
-	"github.com/kernelkit/infix/src/yangerd/internal/zapi"
 )
 
-func TestRouteProtocol(t *testing.T) {
-	tests := []struct {
-		name     string
-		rt       zapi.RouteType
-		expected string
-	}{
-		{"kernel", zapi.RouteKernel, "infix-routing:kernel"},
-		{"connect", zapi.RouteConnect, "ietf-routing:direct"},
-		{"static", zapi.RouteStatic, "ietf-routing:static"},
-		{"ospf", zapi.RouteOSPF, "ietf-ospf:ospfv2"},
-		{"rip", zapi.RouteRIP, "ietf-rip:rip"},
-		{"unknown defaults to kernel", zapi.RouteType(99), "infix-routing:kernel"},
-		{"zero defaults to kernel", zapi.RouteType(0), "infix-routing:kernel"},
-	}
-
-	for _, tc := range tests {
-		t.Run(tc.name, func(t *testing.T) {
-			got := routeProtocol(tc.rt)
-			if got != tc.expected {
-				t.Errorf("routeProtocol(%d) = %q, want %q", tc.rt, got, tc.expected)
-			}
-		})
-	}
+// fakeQuerier returns canned vtysh JSON per command.
+type fakeQuerier struct {
+	ipv4 string
+	ipv6 string
+	err  error
 }
 
-func TestRibName(t *testing.T) {
-	tests := []struct {
-		name     string
-		prefix   net.IPNet
-		expected string
-	}{
-		{
-			"ipv4",
-			net.IPNet{IP: net.ParseIP("10.0.0.0").To4(), Mask: net.CIDRMask(24, 32)},
-			"ipv4",
-		},
-		{
-			"ipv6",
-			net.IPNet{IP: net.ParseIP("2001:db8::"), Mask: net.CIDRMask(64, 128)},
-			"ipv6",
-		},
-		{
-			"loopback v4",
-			net.IPNet{IP: net.ParseIP("127.0.0.0").To4(), Mask: net.CIDRMask(8, 32)},
-			"ipv4",
-		},
+func (f fakeQuerier) Query(_ context.Context, command string) ([]byte, error) {
+	if f.err != nil {
+		return nil, f.err
 	}
-
-	for _, tc := range tests {
-		t.Run(tc.name, func(t *testing.T) {
-			got := ribName(tc.prefix)
-			if got != tc.expected {
-				t.Errorf("ribName(%v) = %q, want %q", tc.prefix, got, tc.expected)
-			}
-		})
+	if command == "show ipv6 route json" {
+		if f.ipv6 == "" {
+			return []byte("{}"), nil
+		}
+		return []byte(f.ipv6), nil
 	}
+	if f.ipv4 == "" {
+		return []byte("{}"), nil
+	}
+	return []byte(f.ipv4), nil
 }
 
-func TestTransformRoute(t *testing.T) {
-	route := &zapi.Route{
-		Type:     zapi.RouteStatic,
-		Distance: 1,
-		Metric:   100,
-		Prefix: net.IPNet{
-			IP:   net.ParseIP("10.0.0.0").To4(),
-			Mask: net.CIDRMask(24, 32),
-		},
-	}
-
-	received := time.Date(2025, 6, 15, 10, 30, 0, 0, time.UTC)
-	result := transformRoute(route, false, received)
-
-	var parsed map[string]any
-	if err := json.Unmarshal(result, &parsed); err != nil {
-		t.Fatalf("unmarshal transformRoute result: %v", err)
-	}
-
-	if got := parsed["ietf-ipv4-unicast-routing:destination-prefix"]; got != "10.0.0.0/24" {
-		t.Errorf("destination-prefix = %v, want %q", got, "10.0.0.0/24")
-	}
-
-	if got := parsed["source-protocol"]; got != "ietf-routing:static" {
-		t.Errorf("source-protocol = %v, want %q", got, "ietf-routing:static")
-	}
-
-	if got, ok := parsed["route-preference"].(float64); !ok || int(got) != 1 {
-		t.Errorf("route-preference = %v, want 1 (admin distance)", parsed["route-preference"])
-	}
-
-	nhContainer, ok := parsed["next-hop"].(map[string]any)
-	if !ok {
-		t.Fatalf("next-hop not a map: %T", parsed["next-hop"])
-	}
-	nhList, ok := nhContainer["next-hop-list"].(map[string]any)
-	if !ok {
-		t.Fatalf("next-hop-list not a map: %T", nhContainer["next-hop-list"])
-	}
-	hops, ok := nhList["next-hop"].([]any)
-	if !ok {
-		t.Fatalf("next-hop not an array: %T", nhList["next-hop"])
-	}
-	if len(hops) != 0 {
-		t.Errorf("expected 0 next-hops, got %d", len(hops))
-	}
-
-	if _, ok := parsed["active"]; ok {
-		t.Error("non-selected route should not have 'active' leaf")
-	}
-
-	lastUpdated, ok := parsed["last-updated"].(string)
-	if !ok {
-		t.Fatal("last-updated missing or not a string")
-	}
-	if lastUpdated != "2025-06-15T10:30:00Z" {
-		t.Errorf("last-updated = %q, want %q", lastUpdated, "2025-06-15T10:30:00Z")
-	}
-}
-
-func TestTransformRouteActiveParam(t *testing.T) {
-	route := &zapi.Route{
-		Type:     zapi.RouteStatic,
-		Distance: 5,
-		Prefix: net.IPNet{
-			IP:   net.ParseIP("0.0.0.0").To4(),
-			Mask: net.CIDRMask(0, 32),
-		},
-	}
-
-	result := transformRoute(route, true, time.Now())
-
-	var parsed map[string]any
-	if err := json.Unmarshal(result, &parsed); err != nil {
-		t.Fatalf("unmarshal: %v", err)
-	}
-
-	active, ok := parsed["active"].([]any)
-	if !ok {
-		t.Fatalf("active leaf missing or wrong type: %v", parsed["active"])
-	}
-	if len(active) != 1 || active[0] != nil {
-		t.Errorf("active = %v, want [null]", active)
-	}
-}
-
-func TestRouteKeyDifferentProtocols(t *testing.T) {
-	ospf := &zapi.Route{
-		Type:     zapi.RouteOSPF,
-		Distance: 110,
-		Prefix: net.IPNet{
-			IP:   net.IPv4(10, 0, 0, 0),
-			Mask: net.CIDRMask(24, 32),
-		},
-		Nexthops: []zapi.Nexthop{{
-			Gate: net.ParseIP("192.168.10.3").To4(),
-		}},
-	}
-	static := &zapi.Route{
-		Type:     zapi.RouteStatic,
-		Distance: 120,
-		Prefix: net.IPNet{
-			IP:   net.IPv4(10, 0, 0, 0),
-			Mask: net.CIDRMask(24, 32),
-		},
-		Nexthops: []zapi.Nexthop{{
-			Gate: net.ParseIP("192.168.50.2").To4(),
-		}},
-	}
-
-	keyOSPF := routeKey(ospf)
-	keyStatic := routeKey(static)
-
-	if keyOSPF == keyStatic {
-		t.Errorf("routes with different protocols must have different keys: %q == %q", keyOSPF, keyStatic)
-	}
-}
-
-func TestRouteKeySamePrefixProtoIgnoresDistance(t *testing.T) {
-	old := &zapi.Route{
-		Type:     zapi.RouteStatic,
-		Distance: 120,
-		Prefix: net.IPNet{
-			IP:   net.IPv4(0, 0, 0, 0),
-			Mask: net.CIDRMask(0, 32),
-		},
-	}
-	updated := &zapi.Route{
-		Type:     zapi.RouteStatic,
-		Distance: 1,
-		Prefix: net.IPNet{
-			IP:   net.IPv4(0, 0, 0, 0),
-			Mask: net.CIDRMask(0, 32),
-		},
-	}
-
-	if routeKey(old) != routeKey(updated) {
-		t.Errorf("same prefix+proto must produce same key regardless of distance: %q vs %q", routeKey(old), routeKey(updated))
-	}
-}
-
-func TestAddRouteSamePrefixDifferentProtocol(t *testing.T) {
-	tr := tree.New()
-	w := New(tr, nil)
-
-	ospfRoute := &zapi.Route{
-		Type:     zapi.RouteOSPF,
-		Distance: 110,
-		Prefix: net.IPNet{
-			IP:   net.IPv4(0, 0, 0, 0),
-			Mask: net.CIDRMask(0, 32),
-		},
-		Nexthops: []zapi.Nexthop{{
-			Type: zapi.NHIPv4IFIndex,
-			Gate: net.ParseIP("192.168.10.3").To4(),
-		}},
-	}
-	staticRoute := &zapi.Route{
-		Type:     zapi.RouteStatic,
-		Distance: 120,
-		Prefix: net.IPNet{
-			IP:   net.IPv4(0, 0, 0, 0),
-			Mask: net.CIDRMask(0, 32),
-		},
-		Nexthops: []zapi.Nexthop{{
-			Type: zapi.NHIPv4IFIndex,
-			Gate: net.ParseIP("192.168.50.2").To4(),
-		}},
-	}
-
-	w.addRoute(ospfRoute)
-	w.addRoute(staticRoute)
-
-	w.mu.Lock()
-	count := len(w.routes)
-	w.mu.Unlock()
-
-	if count != 2 {
-		t.Errorf("expected 2 routes (different protocols), got %d", count)
-	}
-
+func ipv4Routes(t *testing.T, tr *tree.Tree) []map[string]any {
+	t.Helper()
 	data := tr.Get(routingTreeKey)
 	if data == nil {
 		t.Fatal("routing tree key not set")
 	}
-
 	var routing map[string]any
 	if err := json.Unmarshal(data, &routing); err != nil {
 		t.Fatalf("unmarshal routing: %v", err)
 	}
-
-	ribs, ok := routing["ribs"].(map[string]any)
-	if !ok {
-		t.Fatal("ribs not found")
-	}
-	ribList, ok := ribs["rib"].([]any)
-	if !ok {
-		t.Fatal("rib list not found")
-	}
-
-	var ipv4Routes []any
-	for _, rib := range ribList {
-		ribMap := rib.(map[string]any)
-		if ribMap["name"] == "ipv4" {
-			routes := ribMap["routes"].(map[string]any)
-			ipv4Routes = routes["route"].([]any)
-			break
-		}
-	}
-
-	if len(ipv4Routes) != 2 {
-		t.Fatalf("expected 2 IPv4 routes, got %d", len(ipv4Routes))
-	}
-
-	var activeCount int
-	for _, r := range ipv4Routes {
-		rm := r.(map[string]any)
-		pref := int(rm["route-preference"].(float64))
-		_, hasActive := rm["active"]
-		if hasActive {
-			activeCount++
-			if pref != 110 {
-				t.Errorf("active route has preference %d, want 110", pref)
+	ribs := routing["ribs"].(map[string]any)
+	for _, rib := range ribs["rib"].([]any) {
+		rm := rib.(map[string]any)
+		if rm["name"] == "ipv4" {
+			out := []map[string]any{}
+			for _, r := range rm["routes"].(map[string]any)["route"].([]any) {
+				out = append(out, r.(map[string]any))
 			}
-		}
-		if pref == 120 && hasActive {
-			t.Error("route with preference 120 should not be active")
-		}
-		if _, ok := rm["last-updated"].(string); !ok {
-			t.Errorf("route with preference %d missing last-updated", pref)
+			return out
 		}
 	}
-	if activeCount != 1 {
-		t.Errorf("expected exactly 1 active route, got %d", activeCount)
+	t.Fatal("ipv4 rib not found")
+	return nil
+}
+
+func TestProtocolName(t *testing.T) {
+	cases := map[string]string{
+		"kernel":    "infix-routing:kernel",
+		"connected": "ietf-routing:direct",
+		"local":     "ietf-routing:direct",
+		"static":    "ietf-routing:static",
+		"ospf":      "ietf-ospf:ospfv2",
+		"rip":       "ietf-rip:rip",
+		"bgp":       "infix-routing:kernel", // unknown -> kernel
+		"":          "infix-routing:kernel",
+	}
+	for in, want := range cases {
+		if got := protocolName(in); got != want {
+			t.Errorf("protocolName(%q) = %q, want %q", in, got, want)
+		}
 	}
 }
 
-func TestTransformRouteIPv6(t *testing.T) {
-	route := &zapi.Route{
-		Type:     zapi.RouteOSPF,
-		Distance: 110,
-		Metric:   10,
-		Prefix: net.IPNet{
-			IP:   net.ParseIP("2001:db8::").To16(),
-			Mask: net.CIDRMask(48, 128),
+func TestParseUptime(t *testing.T) {
+	cases := map[string]time.Duration{
+		"02:09:02": 2*time.Hour + 9*time.Minute + 2*time.Second,
+		"00:00:30": 30 * time.Second,
+		"3d04h05m": 3*24*time.Hour + 4*time.Hour + 5*time.Minute,
+		"02w3d04h": 2*7*24*time.Hour + 3*24*time.Hour + 4*time.Hour,
+		"bogus":    0,
+		"":         0,
+	}
+	for in, want := range cases {
+		if got := parseUptime(in); got != want {
+			t.Errorf("parseUptime(%q) = %v, want %v", in, got, want)
+		}
+	}
+}
+
+// The user's exact bug: a static route is in the FIB (selected, distance
+// 120) while a stale OSPF entry (distance 110) is still listed but not
+// selected.  active must follow FRR's "selected" flag, not the lowest
+// admin distance.
+func TestActiveFollowsSelectedNotDistance(t *testing.T) {
+	const j = `{
+	  "192.168.20.0/24":[
+	    {"prefix":"192.168.20.0/24","protocol":"ospf","selected":false,"distance":110,"metric":100,"uptime":"02:11:49",
+	     "nexthops":[{"ip":"192.168.60.2","interfaceName":"e3","active":true}]},
+	    {"prefix":"192.168.20.0/24","protocol":"static","selected":true,"installed":true,"distance":120,"metric":0,"uptime":"02:09:02",
+	     "nexthops":[{"ip":"192.168.50.2","interfaceName":"e7","fib":true,"active":true}]}
+	  ]
+	}`
+
+	tr := tree.New()
+	w := New(tr, fakeQuerier{ipv4: j}, nil)
+	w.writeRibs(context.Background())
+
+	routes := ipv4Routes(t, tr)
+	if len(routes) != 2 {
+		t.Fatalf("expected 2 candidate routes, got %d", len(routes))
+	}
+
+	for _, r := range routes {
+		pref := toInt(r["route-preference"])
+		_, active := r["active"]
+		switch pref {
+		case 120:
+			if !active {
+				t.Error("static route (pref 120, selected) must be active")
+			}
+		case 110:
+			if active {
+				t.Error("ospf route (pref 110, not selected) must NOT be active")
+			}
+		default:
+			t.Errorf("unexpected route-preference %d", pref)
+		}
+	}
+}
+
+// A full snapshot read means a route removed from zebra disappears from
+// the cache without any ZAPI delete.
+func TestSnapshotPurgesRemovedRoutes(t *testing.T) {
+	const before = `{
+	  "192.168.20.0/24":[
+	    {"prefix":"192.168.20.0/24","protocol":"ospf","selected":true,"distance":110,"uptime":"00:05:00","nexthops":[{"ip":"192.168.60.2"}]}
+	  ]
+	}`
+	const after = `{
+	  "192.168.20.0/24":[
+	    {"prefix":"192.168.20.0/24","protocol":"static","selected":true,"installed":true,"distance":120,"uptime":"00:01:00","nexthops":[{"ip":"192.168.50.2","fib":true}]}
+	  ]
+	}`
+
+	tr := tree.New()
+	New(tr, fakeQuerier{ipv4: before}, nil).writeRibs(context.Background())
+	if got := len(ipv4Routes(t, tr)); got != 1 {
+		t.Fatalf("before: expected 1 route, got %d", got)
+	}
+
+	// zebra now has only the static route; the OSPF route is gone with no
+	// delete event.  A fresh snapshot must not carry the corpse forward.
+	New(tr, fakeQuerier{ipv4: after}, nil).writeRibs(context.Background())
+	routes := ipv4Routes(t, tr)
+	if len(routes) != 1 {
+		t.Fatalf("after: expected 1 route, got %d", len(routes))
+	}
+	if got := protocolName("static"); routes[0]["source-protocol"] != got {
+		t.Errorf("surviving route protocol = %v, want %v", routes[0]["source-protocol"], got)
+	}
+}
+
+func TestTransformRouteFields(t *testing.T) {
+	entry := map[string]any{
+		"prefix":    "10.0.0.0/24",
+		"protocol":  "ospf",
+		"selected":  true,
+		"installed": true,
+		"distance":  float64(110),
+		"metric":    float64(20),
+		"uptime":    "01:00:00",
+		"nexthops": []any{
+			map[string]any{"ip": "192.168.1.1", "interfaceName": "e1", "fib": true},
 		},
 	}
 
-	result := transformRoute(route, false, time.Now())
-
+	now := time.Date(2026, 6, 10, 12, 0, 0, 0, time.UTC)
 	var parsed map[string]any
-	if err := json.Unmarshal(result, &parsed); err != nil {
+	if err := json.Unmarshal(transformRoute("ipv4", "10.0.0.0/24", entry, now), &parsed); err != nil {
 		t.Fatalf("unmarshal: %v", err)
 	}
 
-	if got := parsed["ietf-ipv6-unicast-routing:destination-prefix"]; got != "2001:db8::/48" {
-		t.Errorf("destination-prefix = %v, want %q", got, "2001:db8::/48")
+	if parsed["ietf-ipv4-unicast-routing:destination-prefix"] != "10.0.0.0/24" {
+		t.Errorf("destination-prefix = %v", parsed["ietf-ipv4-unicast-routing:destination-prefix"])
 	}
-	if got := parsed["source-protocol"]; got != "ietf-ospf:ospfv2" {
-		t.Errorf("source-protocol = %v, want %q", got, "ietf-ospf:ospfv2")
+	if parsed["source-protocol"] != "ietf-ospf:ospfv2" {
+		t.Errorf("source-protocol = %v", parsed["source-protocol"])
+	}
+	if toInt(parsed["route-preference"]) != 110 {
+		t.Errorf("route-preference = %v", parsed["route-preference"])
+	}
+	if toInt(parsed["ietf-ospf:metric"]) != 20 {
+		t.Errorf("ietf-ospf:metric = %v", parsed["ietf-ospf:metric"])
+	}
+	if _, ok := parsed["active"]; !ok {
+		t.Error("selected route must have active leaf")
+	}
+	// last-updated = now - 1h
+	if parsed["last-updated"] != "2026-06-10T11:00:00Z" {
+		t.Errorf("last-updated = %v, want 2026-06-10T11:00:00Z", parsed["last-updated"])
+	}
+
+	hops := parsed["next-hop"].(map[string]any)["next-hop-list"].(map[string]any)["next-hop"].([]any)
+	if len(hops) != 1 {
+		t.Fatalf("expected 1 nexthop, got %d", len(hops))
+	}
+	hop := hops[0].(map[string]any)
+	if hop["ietf-ipv4-unicast-routing:address"] != "192.168.1.1" {
+		t.Errorf("nexthop address = %v", hop["ietf-ipv4-unicast-routing:address"])
+	}
+	if _, ok := hop["infix-routing:installed"]; !ok {
+		t.Error("fib nexthop must have infix-routing:installed")
 	}
 }
 
-func TestAddRouteSamePrefixProtoOverwrites(t *testing.T) {
+func TestTransformRouteBlackhole(t *testing.T) {
+	entry := map[string]any{
+		"prefix":   "10.1.0.0/24",
+		"protocol": "blackhole",
+		"distance": float64(0),
+		"uptime":   "00:00:10",
+	}
+	var parsed map[string]any
+	if err := json.Unmarshal(transformRoute("ipv4", "10.1.0.0/24", entry, time.Now()), &parsed); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+	nh := parsed["next-hop"].(map[string]any)
+	if nh["special-next-hop"] != "blackhole" {
+		t.Errorf("special-next-hop = %v, want blackhole", nh["special-next-hop"])
+	}
+}
+
+func TestWriteRibsQueryErrorKeepsData(t *testing.T) {
 	tr := tree.New()
-	w := New(tr, nil)
+	// Seed with good data.
+	New(tr, fakeQuerier{ipv4: `{"10.0.0.0/24":[{"prefix":"10.0.0.0/24","protocol":"static","selected":true,"distance":1,"uptime":"00:00:05","nexthops":[{"ip":"10.0.0.1"}]}]}`}, nil).
+		writeRibs(context.Background())
+	before := tr.Get(routingTreeKey)
 
-	gateway := net.ParseIP("192.168.50.2").To4()
-	prefix := net.IPNet{IP: net.IPv4(0, 0, 0, 0), Mask: net.CIDRMask(0, 32)}
+	// A failing query must not blank the table.
+	New(tr, fakeQuerier{err: context.DeadlineExceeded}, nil).writeRibs(context.Background())
+	after := tr.Get(routingTreeKey)
 
-	w.addRoute(&zapi.Route{
-		Type:     zapi.RouteStatic,
-		Distance: 120,
-		Prefix:   prefix,
-		Nexthops: []zapi.Nexthop{{Type: zapi.NHIPv4IFIndex, Gate: gateway}},
-	})
-
-	newGateway := net.ParseIP("10.0.0.1").To4()
-	w.addRoute(&zapi.Route{
-		Type:     zapi.RouteStatic,
-		Distance: 1,
-		Prefix:   prefix,
-		Nexthops: []zapi.Nexthop{{Type: zapi.NHIPv4IFIndex, Gate: newGateway}},
-	})
-
-	w.mu.Lock()
-	count := len(w.routes)
-	w.mu.Unlock()
-
-	if count != 1 {
-		t.Errorf("same prefix+proto should overwrite: got %d routes, want 1", count)
+	if string(before) != string(after) {
+		t.Error("query error overwrote existing rib data")
 	}
+}
 
-	data := tr.Get(routingTreeKey)
-	if data == nil {
-		t.Fatal("routing tree key not set")
-	}
+func TestWriteRibsBothFamilies(t *testing.T) {
+	tr := tree.New()
+	w := New(tr, fakeQuerier{
+		ipv4: `{"10.0.0.0/24":[{"prefix":"10.0.0.0/24","protocol":"static","selected":true,"distance":1,"uptime":"00:00:05","nexthops":[{"ip":"10.0.0.1"}]}]}`,
+		ipv6: `{"2001:db8::/64":[{"prefix":"2001:db8::/64","protocol":"connected","selected":true,"distance":0,"uptime":"00:00:05","nexthops":[{"interfaceName":"e1"}]}]}`,
+	}, nil)
+	w.writeRibs(context.Background())
 
 	var routing map[string]any
-	if err := json.Unmarshal(data, &routing); err != nil {
+	if err := json.Unmarshal(tr.Get(routingTreeKey), &routing); err != nil {
 		t.Fatalf("unmarshal: %v", err)
 	}
-
-	ribs := routing["ribs"].(map[string]any)
-	ribList := ribs["rib"].([]any)
-	var ipv4Routes []any
-	for _, rib := range ribList {
-		ribMap := rib.(map[string]any)
-		if ribMap["name"] == "ipv4" {
-			ipv4Routes = ribMap["routes"].(map[string]any)["route"].([]any)
-			break
-		}
-	}
-
-	if len(ipv4Routes) != 1 {
-		t.Fatalf("expected 1 route, got %d", len(ipv4Routes))
-	}
-
-	rm := ipv4Routes[0].(map[string]any)
-	if pref := int(rm["route-preference"].(float64)); pref != 1 {
-		t.Errorf("route-preference = %d, want 1 (latest add wins)", pref)
-	}
-}
-
-func TestDeleteRouteWithoutNexthops(t *testing.T) {
-	tr := tree.New()
-	w := New(tr, nil)
-
-	gateway := net.ParseIP("192.168.10.3").To4()
-	prefix := net.IPNet{IP: net.IPv4(0, 0, 0, 0), Mask: net.CIDRMask(0, 32)}
-
-	w.addRoute(&zapi.Route{
-		Type:     zapi.RouteStatic,
-		Distance: 5,
-		Prefix:   prefix,
-		Nexthops: []zapi.Nexthop{{Type: zapi.NHIPv4IFIndex, Gate: gateway}},
-	})
-
-	w.mu.Lock()
-	count := len(w.routes)
-	w.mu.Unlock()
-	if count != 1 {
-		t.Fatalf("expected 1 route after add, got %d", count)
-	}
-
-	w.deleteRoute(&zapi.Route{
-		Type:   zapi.RouteStatic,
-		Prefix: prefix,
-	})
-
-	w.mu.Lock()
-	count = len(w.routes)
-	w.mu.Unlock()
-	if count != 0 {
-		t.Errorf("expected 0 routes after delete without nexthops, got %d", count)
+	ribs := routing["ribs"].(map[string]any)["rib"].([]any)
+	if len(ribs) != 2 {
+		t.Fatalf("expected 2 ribs, got %d", len(ribs))
 	}
 }