yangerd: Fix ntp and lldp status

src/yangerd/internal/collector/ntp.go

@@ -48,21 +48,36 @@ func (c *NTPCollector) Collect(ctx context.Context, t *tree.Tree) error {
 	c.addServerStatus(ctx, ntp)
 	c.addServerStats(ctx, ntp)
 
-	if len(ntp) == 0 {
-		return nil
-	}
-
-	if data, err := json.Marshal(ntp); err == nil {
-		t.Set("ietf-ntp:ntp", data)
-	}
-
-	// Populate the Infix NTP sources augmentation under system-state.
-	if sources := c.addSources(sourcesOut); sources != nil {
-		if data, err := json.Marshal(map[string]interface{}{
-			"infix-system:ntp": sources,
-		}); err == nil {
-			t.Merge("ietf-system:system-state", data)
+	if len(ntp) > 0 {
+		if data, err := json.Marshal(ntp); err == nil {
+			t.Set("ietf-ntp:ntp", data)
 		}
+	} else {
+		// chronyd is not running (NTP unconfigured, or disabled by a
+		// config change).  Drop the key so data from a previous run does
+		// not linger -- yangerd outlives config resets, so stale state
+		// would otherwise survive until restart.
+		t.Delete("ietf-ntp:ntp")
+	}
+
+	// Always refresh the Infix NTP sources under system-state, even when
+	// empty.  Otherwise a source that disappears from chrony (e.g. a DHCP
+	// lease without option 42, or NTP turned off) lingers as stale
+	// operational data -- a phantom "selected" server chronyc no longer
+	// reports.  Merge only overwrites the keys it is given, so we must
+	// hand it an empty source list to clear a previously-populated one.
+	sources := c.addSources(sourcesOut)
+	if sources == nil {
+		sources = map[string]interface{}{
+			"sources": map[string]interface{}{
+				"source": []interface{}{},
+			},
+		}
+	}
+	if data, err := json.Marshal(map[string]interface{}{
+		"infix-system:ntp": sources,
+	}); err == nil {
+		t.Merge("ietf-system:system-state", data)
 	}
 
 	return nil

src/yangerd/internal/collector/ntp_test.go

@@ -225,6 +225,36 @@ func TestNTPSources(t *testing.T) {
 	}
 }
 
+// ntpSourceCount returns the number of infix-system:ntp sources in the
+// system-state tree key, failing the test if the subtree is missing.
+func ntpSourceCount(t *testing.T, tr *tree.Tree) int {
+	t.Helper()
+
+	raw := tr.Get("ietf-system:system-state")
+	if raw == nil {
+		t.Fatal("system-state not set")
+	}
+
+	var data map[string]json.RawMessage
+	if err := json.Unmarshal(raw, &data); err != nil {
+		t.Fatalf("unmarshal system-state: %v", err)
+	}
+	ntpRaw, ok := data["infix-system:ntp"]
+	if !ok {
+		t.Fatal("infix-system:ntp not present")
+	}
+
+	var ntp struct {
+		Sources struct {
+			Source []json.RawMessage `json:"source"`
+		} `json:"sources"`
+	}
+	if err := json.Unmarshal(ntpRaw, &ntp); err != nil {
+		t.Fatalf("unmarshal infix-system:ntp: %v", err)
+	}
+	return len(ntp.Sources.Source)
+}
+
 func TestNTPSourcesEmpty(t *testing.T) {
 	runner := &testutil.MockRunner{
 		Results: map[string][]byte{
@@ -237,9 +267,69 @@ func TestNTPSourcesEmpty(t *testing.T) {
 	tr := tree.New()
 	c.Collect(context.Background(), tr)
 
-	raw := tr.Get("ietf-system:system-state")
-	if raw != nil {
-		t.Fatal("should not set system-state when no sources available")
+	// With no chrony sources the collector must still write an empty
+	// source list, so a previously-reported source cannot linger as
+	// stale operational data.
+	if n := ntpSourceCount(t, tr); n != 0 {
+		t.Fatalf("expected empty NTP source list, got %d", n)
+	}
+}
+
+// When chronyd stops (NTP disabled via config reset), the whole
+// ietf-ntp:ntp key must disappear -- yangerd outlives config resets, so
+// a key that is only ever Set when non-empty would keep stale data from
+// a previous run forever.
+func TestNTPTreeKeyRemovedWhenChronydStops(t *testing.T) {
+	tr := tree.New()
+
+	running := &testutil.MockRunner{
+		Results: map[string][]byte{
+			"chronyc -c sources":  []byte(testChronycSources),
+			"chronyc -c tracking": []byte(testChronycTracking),
+		},
+		Errors: map[string]error{},
+	}
+	newNTPCollector(running).Collect(context.Background(), tr)
+	if tr.Get("ietf-ntp:ntp") == nil {
+		t.Fatal("expected ietf-ntp:ntp after first poll")
+	}
+
+	stopped := &testutil.MockRunner{
+		Results: map[string][]byte{},
+		Errors:  map[string]error{},
+	}
+	newNTPCollector(stopped).Collect(context.Background(), tr)
+	if data := tr.Get("ietf-ntp:ntp"); data != nil {
+		t.Fatalf("stale ietf-ntp:ntp survived chronyd stop: %s", data)
+	}
+}
+
+// A source that disappears from chrony (e.g. a DHCP NTP server that is no
+// longer offered) must be cleared from operational, not left stale.
+func TestNTPSourcesClearedWhenGone(t *testing.T) {
+	tr := tree.New()
+
+	withSources := &testutil.MockRunner{
+		Results: map[string][]byte{
+			"chronyc -c sources":  []byte(testChronycSources),
+			"chronyc -c tracking": []byte(testChronycTracking),
+		},
+		Errors: map[string]error{},
+	}
+	newNTPCollector(withSources).Collect(context.Background(), tr)
+	if ntpSourceCount(t, tr) == 0 {
+		t.Fatal("expected NTP sources after first poll")
+	}
+
+	noSources := &testutil.MockRunner{
+		Results: map[string][]byte{
+			"chronyc -c tracking": []byte(testChronycTracking),
+		},
+		Errors: map[string]error{},
+	}
+	newNTPCollector(noSources).Collect(context.Background(), tr)
+	if n := ntpSourceCount(t, tr); n != 0 {
+		t.Fatalf("stale NTP sources not cleared: got %d, want 0", n)
 	}
 }
 

src/yangerd/internal/lldpmonitor/lldpmonitor.go

@@ -1,7 +1,12 @@
-// Package lldpmonitor manages a persistent `lldpcli -f json0 watch`
-// subprocess for reactive LLDP neighbor updates.  Events are framed
-// as pretty-printed JSON objects separated by blank lines.  Each
-// event triggers full LLDP subtree regeneration.
+// Package lldpmonitor keeps the LLDP neighbor table in the tree in sync
+// with lldpd.  A persistent `lldpcli -f json0 watch` subprocess is used
+// purely as a change trigger -- its events carry only the changed
+// neighbor, so they cannot be used to rebuild state (a delete event
+// would re-add the neighbor, and an event for one port would wipe the
+// others).  On every event the full table is re-read with
+// `lldpcli -f json0 show neighbors` and the subtree replaced, so removed
+// neighbors disappear and neighbors present before yangerd started are
+// picked up.
 package lldpmonitor
 
 import (
@@ -15,6 +20,7 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/kernelkit/infix/src/yangerd/internal/backoff"
 	"github.com/kernelkit/infix/src/yangerd/internal/tree"
@@ -23,22 +29,47 @@ import (
 const (
 	lldpMulticastMAC = "01:80:C2:00:00:0E"
 	treeKey          = "ieee802-dot1ab-lldp:lldp"
+
+	// debounceDelay coalesces bursts of watch events into one re-read.
+	debounceDelay = 200 * time.Millisecond
+	queryTimeout  = 5 * time.Second
 )
 
 // LLDPMonitor subscribes to LLDP neighbor events via a persistent
-// lldpcli watch subprocess and updates the tree on every event.
+// lldpcli watch subprocess and re-reads the full neighbor table on
+// every event.
 type LLDPMonitor struct {
-	tree *tree.Tree
-	log  *slog.Logger
+	tree    *tree.Tree
+	log     *slog.Logger
+	refresh chan struct{}
+
+	// query returns the current full neighbor table; overridable in tests.
+	query func(ctx context.Context) ([]byte, error)
 }
 
 // New creates an LLDPMonitor.
 func New(t *tree.Tree, log *slog.Logger) *LLDPMonitor {
-	return &LLDPMonitor{tree: t, log: log}
+	if log == nil {
+		log = slog.Default()
+	}
+	return &LLDPMonitor{
+		tree:    t,
+		log:     log,
+		refresh: make(chan struct{}, 1),
+		query:   queryNeighbors,
+	}
+}
+
+func queryNeighbors(ctx context.Context) ([]byte, error) {
+	ctx, cancel := context.WithTimeout(ctx, queryTimeout)
+	defer cancel()
+	return exec.CommandContext(ctx, "lldpcli", "-f", "json0", "show", "neighbors").Output()
 }
 
 // Run starts the LLDP monitor.  It blocks until ctx is cancelled.
 func (m *LLDPMonitor) Run(ctx context.Context) error {
+	go m.refreshLoop(ctx)
+
 	bo := backoff.Default()
 	delay := bo.Initial
 
@@ -68,6 +99,9 @@ func (m *LLDPMonitor) runOnce(ctx context.Context) error {
 	}
 	defer cmd.Wait()
 
+	// Pick up neighbors that existed before we attached.
+	m.triggerRefresh()
+
 	scanner := bufio.NewScanner(stdout)
 	scanner.Buffer(make([]byte, 0, 1*1024*1024), 1*1024*1024)
 
@@ -109,6 +143,8 @@ func (m *LLDPMonitor) runOnce(ctx context.Context) error {
 	return fmt.Errorf("lldpcli watch process exited")
 }
 
+// processEvent inspects a watch event and triggers a full table re-read.
+// The event payload itself is never used to build state.
 func (m *LLDPMonitor) processEvent(data []byte) {
 	var raw map[string]json.RawMessage
 	if err := json.Unmarshal(data, &raw); err != nil {
@@ -116,60 +152,210 @@ func (m *LLDPMonitor) processEvent(data []byte) {
 		return
 	}
 
-	// Dispatch by root key: lldp-added, lldp-updated, lldp-deleted.
-	// All event types trigger full subtree rebuild from payload.
 	for key := range raw {
 		switch key {
 		case "lldp-added", "lldp-updated", "lldp-deleted":
-			m.rebuildTree(data)
+			m.log.Debug("lldp monitor: neighbor change", "event", key)
+			m.triggerRefresh()
 			return
 		}
 	}
 	m.log.Debug("lldp monitor: unknown event keys", "keys", keysOf(raw))
 }
 
-func (m *LLDPMonitor) rebuildTree(data []byte) {
-	result := transformLLDPEvent(data)
-	m.tree.Set(treeKey, result)
+// triggerRefresh requests a table re-read; the buffered channel collapses
+// pending requests into one.
+func (m *LLDPMonitor) triggerRefresh() {
+	select {
+	case m.refresh <- struct{}{}:
+	default:
+	}
+}
+
+func (m *LLDPMonitor) refreshLoop(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-m.refresh:
+		}
+
+		// Let a burst of events settle before reading.
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(debounceDelay):
+		}
+		select {
+		case <-m.refresh:
+		default:
+		}
+
+		m.updateTree(ctx)
+	}
+}
+
+// updateTree reads the full neighbor table and replaces the subtree.
+// On a query error the previous data is left untouched.
+func (m *LLDPMonitor) updateTree(ctx context.Context) {
+	out, err := m.query(ctx)
+	if err != nil {
+		m.log.Warn("lldp monitor: show neighbors", "err", err)
+		return
+	}
+
+	m.tree.Set(treeKey, transformNeighbors(out))
 	m.log.Debug("lldp monitor: tree updated")
 }
 
-// transformLLDPEvent converts lldpcli json0 watch output to YANG
-// ieee802-dot1ab-lldp:lldp format matching the Python infix_lldp.py.
-func transformLLDPEvent(data []byte) json.RawMessage {
-	var event map[string]json.RawMessage
-	if err := json.Unmarshal(data, &event); err != nil {
-		return json.RawMessage(`{}`)
+// j0ID is a chassis/port id element: {"type": "mac", "value": "..."}.
+type j0ID struct {
+	Type  string `json:"type"`
+	Value string `json:"value"`
+}
+
+// j0Iface is one neighbor entry on an interface.  In json0 format the
+// interface name, rid and age are plain string fields and chassis/port
+// are arrays.  In the older keyed json format chassis/port are objects;
+// the custom unmarshallers accept both.
+type j0Iface struct {
+	Name    string      `json:"name"`
+	RID     interface{} `json:"rid"`
+	Age     string      `json:"age"`
+	Chassis j0IDHolder  `json:"chassis"`
+	Port    j0IDHolder  `json:"port"`
+}
+
+// j0IDHolder extracts the first id from a chassis/port node in either
+// json0 (array) or json (object) form.
+type j0IDHolder struct {
+	ID j0ID
+}
+
+func (h *j0IDHolder) UnmarshalJSON(data []byte) error {
+	var asArray []struct {
+		ID json.RawMessage `json:"id"`
+	}
+	if err := json.Unmarshal(data, &asArray); err == nil {
+		for _, e := range asArray {
+			if id, ok := parseID(e.ID); ok {
+				h.ID = id
+				return nil
+			}
+		}
+		return nil
 	}
 
-	// Extract the lldp payload from whichever event key is present.
-	var lldpPayload json.RawMessage
-	for _, key := range []string{"lldp-added", "lldp-updated", "lldp-deleted"} {
-		if p, ok := event[key]; ok {
-			lldpPayload = p
-			break
-		}
+	var asObject struct {
+		ID json.RawMessage `json:"id"`
 	}
-	if lldpPayload == nil {
-		return json.RawMessage(`{}`)
+	if err := json.Unmarshal(data, &asObject); err != nil {
+		return nil // tolerate unknown shapes
+	}
+	if id, ok := parseID(asObject.ID); ok {
+		h.ID = id
+	}
+	return nil
+}
+
+// parseID accepts an id as object {"type","value"} or array of such.
+func parseID(raw json.RawMessage) (j0ID, bool) {
+	if len(raw) == 0 {
+		return j0ID{}, false
+	}
+	var one j0ID
+	if err := json.Unmarshal(raw, &one); err == nil && (one.Type != "" || one.Value != "") {
+		return one, true
+	}
+	var many []j0ID
+	if err := json.Unmarshal(raw, &many); err == nil && len(many) > 0 {
+		return many[0], true
+	}
+	return j0ID{}, false
+}
+
+// collectIfaces extracts all neighbor interface entries from a show
+// neighbors document, accepting both json0 ("lldp" is an array, entries
+// carry a "name" field) and json ("lldp" is an object, entries are keyed
+// by interface name) output formats.
+func collectIfaces(data []byte) []j0Iface {
+	var ifaces []j0Iface
+
+	addRaw := func(name string, raw json.RawMessage) {
+		var iface j0Iface
+		if err := json.Unmarshal(raw, &iface); err != nil {
+			return
+		}
+		if iface.Name == "" {
+			iface.Name = name
+		}
+		if iface.Name != "" {
+			ifaces = append(ifaces, iface)
+		}
 	}
 
-	var lldpData struct {
-		LLDP struct {
-			Interface []map[string]json.RawMessage `json:"interface"`
-		} `json:"lldp"`
-	}
-	if err := json.Unmarshal(lldpPayload, &lldpData); err != nil {
-		// Try unwrapped format.
-		var unwrapped struct {
-			Interface []map[string]json.RawMessage `json:"interface"`
+	collectInterfaceNode := func(raw json.RawMessage) {
+		// json0: array of {"name": "eth0", ...}
+		var asArray []json.RawMessage
+		if err := json.Unmarshal(raw, &asArray); err == nil {
+			for _, e := range asArray {
+				// Either a direct entry with "name", or a keyed map
+				// {"eth0": {...}} from the older json format.
+				var iface j0Iface
+				if err := json.Unmarshal(e, &iface); err == nil && iface.Name != "" {
+					ifaces = append(ifaces, iface)
+					continue
+				}
+				var keyed map[string]json.RawMessage
+				if err := json.Unmarshal(e, &keyed); err == nil {
+					for name, v := range keyed {
+						addRaw(name, v)
+					}
+				}
+			}
+			return
 		}
-		if err := json.Unmarshal(lldpPayload, &unwrapped); err != nil {
-			return json.RawMessage(`{}`)
+		// json: single keyed map {"eth0": {...}}
+		var keyed map[string]json.RawMessage
+		if err := json.Unmarshal(raw, &keyed); err == nil {
+			for name, v := range keyed {
+				addRaw(name, v)
+			}
 		}
-		lldpData.LLDP.Interface = unwrapped.Interface
 	}
 
+	var doc map[string]json.RawMessage
+	if err := json.Unmarshal(data, &doc); err != nil {
+		return nil
+	}
+	lldpRaw, ok := doc["lldp"]
+	if !ok {
+		return nil
+	}
+
+	// json0: "lldp" is an array of {"interface": [...]}; json: an object.
+	var lldpArray []map[string]json.RawMessage
+	if err := json.Unmarshal(lldpRaw, &lldpArray); err == nil {
+		for _, entry := range lldpArray {
+			if ifRaw, ok := entry["interface"]; ok {
+				collectInterfaceNode(ifRaw)
+			}
+		}
+		return ifaces
+	}
+	var lldpObject map[string]json.RawMessage
+	if err := json.Unmarshal(lldpRaw, &lldpObject); err == nil {
+		if ifRaw, ok := lldpObject["interface"]; ok {
+			collectInterfaceNode(ifRaw)
+		}
+	}
+	return ifaces
+}
+
+// transformNeighbors converts `lldpcli show neighbors` output to the
+// YANG ieee802-dot1ab-lldp subtree (unwrapped -- the IPC layer adds the
+// module envelope when serving the key).
+func transformNeighbors(data []byte) json.RawMessage {
 	type remoteEntry struct {
 		TimeMark         int    `json:"time-mark"`
 		RemoteIndex      int    `json:"remote-index"`
@@ -178,7 +364,6 @@ func transformLLDPEvent(data []byte) json.RawMessage {
 		PortIDSubtype    string `json:"port-id-subtype"`
 		PortID           string `json:"port-id"`
 	}
-
 	type portEntry struct {
 		Name           string        `json:"name"`
 		DestMACAddress string        `json:"dest-mac-address"`
@@ -186,71 +371,47 @@ func transformLLDPEvent(data []byte) json.RawMessage {
 	}
 
 	portMap := make(map[string]*portEntry)
+	var order []string
 
-	for _, ifEntry := range lldpData.LLDP.Interface {
-		for ifName, ifData := range ifEntry {
-			var iface struct {
-				RID     interface{} `json:"rid"`
-				Age     string      `json:"age"`
-				Chassis struct {
-					ID struct {
-						Type  string `json:"type"`
-						Value string `json:"value"`
-					} `json:"id"`
-				} `json:"chassis"`
-				Port struct {
-					ID struct {
-						Type  string `json:"type"`
-						Value string `json:"value"`
-					} `json:"id"`
-				} `json:"port"`
+	for _, iface := range collectIfaces(data) {
+		port, ok := portMap[iface.Name]
+		if !ok {
+			port = &portEntry{
+				Name:           iface.Name,
+				DestMACAddress: lldpMulticastMAC,
 			}
-			if err := json.Unmarshal(ifData, &iface); err != nil {
-				continue
-			}
-
-			port, ok := portMap[ifName]
-			if !ok {
-				port = &portEntry{
-					Name:           ifName,
-					DestMACAddress: lldpMulticastMAC,
-				}
-				portMap[ifName] = port
-			}
-
-			rid := 0
-			switch v := iface.RID.(type) {
-			case float64:
-				rid = int(v)
-			case string:
-				rid, _ = strconv.Atoi(v)
-			}
-
-			remote := remoteEntry{
-				TimeMark:         parseAge(iface.Age),
-				RemoteIndex:      rid,
-				ChassisIDSubtype: chassisIDSubtype(iface.Chassis.ID.Type),
-				ChassisID:        iface.Chassis.ID.Value,
-				PortIDSubtype:    portIDSubtype(iface.Port.ID.Type),
-				PortID:           iface.Port.ID.Value,
-			}
-			port.RemoteSystems = append(port.RemoteSystems, remote)
+			portMap[iface.Name] = port
+			order = append(order, iface.Name)
 		}
-	}
 
-	var ports []portEntry
-	for _, p := range portMap {
-		if len(p.RemoteSystems) > 0 {
-			ports = append(ports, *p)
+		rid := 0
+		switch v := iface.RID.(type) {
+		case float64:
+			rid = int(v)
+		case string:
+			rid, _ = strconv.Atoi(v)
 		}
+
+		port.RemoteSystems = append(port.RemoteSystems, remoteEntry{
+			TimeMark:         parseAge(iface.Age),
+			RemoteIndex:      rid,
+			ChassisIDSubtype: chassisIDSubtype(iface.Chassis.ID.Type),
+			ChassisID:        iface.Chassis.ID.Value,
+			PortIDSubtype:    portIDSubtype(iface.Port.ID.Type),
+			PortID:           iface.Port.ID.Value,
+		})
 	}
 
-	result := map[string]interface{}{
-		"ieee802-dot1ab-lldp:lldp": map[string]interface{}{
-			"port": ports,
-		},
+	ports := make([]portEntry, 0, len(order))
+	for _, name := range order {
+		ports = append(ports, *portMap[name])
 	}
-	out, _ := json.Marshal(result)
+
+	if len(ports) == 0 {
+		return json.RawMessage(`{}`)
+	}
+
+	out, _ := json.Marshal(map[string]interface{}{"port": ports})
 	return json.RawMessage(out)
 }
 

src/yangerd/internal/lldpmonitor/lldpmonitor_test.go

@@ -1,167 +1,243 @@
 package lldpmonitor
 
 import (
+	"context"
 	"encoding/json"
+	"errors"
 	"reflect"
 	"testing"
+
+	"github.com/kernelkit/infix/src/yangerd/internal/tree"
 )
 
-func TestTransformLLDPEvent(t *testing.T) {
-	tests := []struct {
-		name         string
-		input        string
-		wantPort     int
-		wantByIfName map[string]int
-		wantFirst    map[string]string
-	}{
-		{
-			name: "lldp-added single neighbor",
-			input: `{
-				"lldp-added": {
-					"lldp": {
-						"interface": [
-							{
-								"eth0": {
-									"rid": 7,
-									"age": "0 day, 00:05:30",
-									"chassis": {"id": {"type": "mac", "value": "aa:bb:cc:dd:ee:ff"}},
-									"port": {"id": {"type": "ifname", "value": "swp1"}}
-								}
-							}
-						]
-					}
-				}
-			}`,
-			wantPort:     1,
-			wantByIfName: map[string]int{"eth0": 1},
-			wantFirst: map[string]string{
-				"chassis-id-subtype": "mac-address",
-				"port-id-subtype":    "interface-name",
-				"chassis-id":         "aa:bb:cc:dd:ee:ff",
-				"port-id":            "swp1",
-			},
-		},
-		{
-			name: "lldp-updated multiple neighbors same port",
-			input: `{
-				"lldp-updated": {
-					"lldp": {
-						"interface": [
-							{
-								"eth1": {
-									"rid": "1",
-									"age": "1 day, 02:30:15",
-									"chassis": {"id": {"type": "ifname", "value": "leaf1"}},
-									"port": {"id": {"type": "local", "value": "portA"}}
-								}
-							},
-							{
-								"eth1": {
-									"rid": 2,
-									"age": "10 days, 00:00:00",
-									"chassis": {"id": {"type": "ip", "value": "192.0.2.1"}},
-									"port": {"id": {"type": "mac", "value": "00:11:22:33:44:55"}}
-								}
-							}
-						]
-					}
-				}
-			}`,
-			wantPort:     1,
-			wantByIfName: map[string]int{"eth1": 2},
-		},
-		{
-			name: "lldp-deleted event",
-			input: `{
-				"lldp-deleted": {
-					"lldp": {
-						"interface": [
-							{
-								"eth2": {
-									"rid": 4,
-									"age": "0 day, 00:00:01",
-									"chassis": {"id": {"type": "local", "value": "chassis-local"}},
-									"port": {"id": {"type": "ip", "value": "198.51.100.3"}}
-								}
-							}
-						]
-					}
-				}
-			}`,
-			wantPort:     1,
-			wantByIfName: map[string]int{"eth2": 1},
-		},
-		{
-			name:         "empty object",
-			input:        `{}`,
-			wantPort:     0,
-			wantByIfName: map[string]int{},
-		},
-		{
-			name:         "malformed input",
-			input:        `{not-json`,
-			wantPort:     0,
-			wantByIfName: map[string]int{},
-		},
+type remote struct {
+	TimeMark         int    `json:"time-mark"`
+	RemoteIndex      int    `json:"remote-index"`
+	ChassisIDSubtype string `json:"chassis-id-subtype"`
+	ChassisID        string `json:"chassis-id"`
+	PortIDSubtype    string `json:"port-id-subtype"`
+	PortID           string `json:"port-id"`
+}
+type port struct {
+	Name          string   `json:"name"`
+	DestMAC       string   `json:"dest-mac-address"`
+	RemoteSystems []remote `json:"remote-systems-data"`
+}
+
+// outShape is the stored (unwrapped) subtree: the IPC layer adds the
+// module envelope when serving the key.
+type outShape struct {
+	Port []port `json:"port"`
+}
+
+// json0 format: "lldp" is an array, interface entries carry a "name"
+// field, rid/age are string attributes, chassis/port/id are arrays.
+const showNeighborsJSON0 = `{
+  "lldp": [{
+    "interface": [
+      {
+        "name": "eth0",
+        "via": "LLDP",
+        "rid": "7",
+        "age": "0 day, 00:05:30",
+        "chassis": [{
+          "id": [{"type": "mac", "value": "aa:bb:cc:dd:ee:ff"}],
+          "name": [{"value": "switch1"}]
+        }],
+        "port": [{
+          "id": [{"type": "ifname", "value": "swp1"}]
+        }]
+      },
+      {
+        "name": "eth1",
+        "via": "LLDP",
+        "rid": "9",
+        "age": "1 day, 02:30:15",
+        "chassis": [{
+          "id": [{"type": "local", "value": "Chassis ID 007"}]
+        }],
+        "port": [{
+          "id": [{"type": "mac", "value": "02:01:02:03:04:05"}]
+        }]
+      }
+    ]
+  }]
+}`
+
+// Older keyed json format: "lldp" is an object, interfaces are keyed by
+// name, chassis/port are objects.
+const showNeighborsJSON = `{
+  "lldp": {
+    "interface": [
+      {
+        "eth0": {
+          "rid": 7,
+          "age": "0 day, 00:05:30",
+          "chassis": {"id": {"type": "mac", "value": "aa:bb:cc:dd:ee:ff"}},
+          "port": {"id": {"type": "ifname", "value": "swp1"}}
+        }
+      }
+    ]
+  }
+}`
+
+func decode(t *testing.T, raw json.RawMessage) outShape {
+	t.Helper()
+	var out outShape
+	if err := json.Unmarshal(raw, &out); err != nil {
+		t.Fatalf("unmarshal output: %v", err)
+	}
+	return out
+}
+
+func TestTransformNeighborsJSON0(t *testing.T) {
+	out := decode(t, transformNeighbors([]byte(showNeighborsJSON0)))
+
+	if len(out.Port) != 2 {
+		t.Fatalf("port count = %d, want 2", len(out.Port))
 	}
 
-	type remote struct {
-		TimeMark         int    `json:"time-mark"`
-		RemoteIndex      int    `json:"remote-index"`
-		ChassisIDSubtype string `json:"chassis-id-subtype"`
-		ChassisID        string `json:"chassis-id"`
-		PortIDSubtype    string `json:"port-id-subtype"`
-		PortID           string `json:"port-id"`
-	}
-	type port struct {
-		Name          string   `json:"name"`
-		DestMAC       string   `json:"dest-mac-address"`
-		RemoteSystems []remote `json:"remote-systems-data"`
-	}
-	type outShape struct {
-		LLDP struct {
-			Port []port `json:"port"`
-		} `json:"ieee802-dot1ab-lldp:lldp"`
+	byIf := make(map[string]port)
+	for _, p := range out.Port {
+		if p.DestMAC != lldpMulticastMAC {
+			t.Fatalf("dest-mac-address = %q, want %q", p.DestMAC, lldpMulticastMAC)
+		}
+		byIf[p.Name] = p
 	}
 
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got := transformLLDPEvent([]byte(tt.input))
+	eth0, ok := byIf["eth0"]
+	if !ok || len(eth0.RemoteSystems) != 1 {
+		t.Fatalf("eth0 missing or wrong neighbor count: %#v", byIf)
+	}
+	want := remote{
+		TimeMark:         330,
+		RemoteIndex:      7,
+		ChassisIDSubtype: "mac-address",
+		ChassisID:        "aa:bb:cc:dd:ee:ff",
+		PortIDSubtype:    "interface-name",
+		PortID:           "swp1",
+	}
+	if !reflect.DeepEqual(eth0.RemoteSystems[0], want) {
+		t.Fatalf("eth0 remote\n got: %#v\nwant: %#v", eth0.RemoteSystems[0], want)
+	}
 
-			var decoded outShape
-			if err := json.Unmarshal(got, &decoded); err != nil {
-				t.Fatalf("unmarshal output: %v", err)
-			}
+	eth1 := byIf["eth1"]
+	if len(eth1.RemoteSystems) != 1 {
+		t.Fatalf("eth1 neighbor count = %d", len(eth1.RemoteSystems))
+	}
+	r := eth1.RemoteSystems[0]
+	if r.ChassisIDSubtype != "local" || r.ChassisID != "Chassis ID 007" {
+		t.Errorf("eth1 chassis = %s/%s", r.ChassisIDSubtype, r.ChassisID)
+	}
+	if r.PortIDSubtype != "mac-address" || r.PortID != "02:01:02:03:04:05" {
+		t.Errorf("eth1 port = %s/%s", r.PortIDSubtype, r.PortID)
+	}
+	if r.RemoteIndex != 9 || r.TimeMark != 95415 {
+		t.Errorf("eth1 rid/age = %d/%d", r.RemoteIndex, r.TimeMark)
+	}
+}
 
-			if len(decoded.LLDP.Port) != tt.wantPort {
-				t.Fatalf("port count = %d, want %d", len(decoded.LLDP.Port), tt.wantPort)
-			}
+func TestTransformNeighborsKeyedJSON(t *testing.T) {
+	out := decode(t, transformNeighbors([]byte(showNeighborsJSON)))
 
-			byIf := make(map[string]int)
-			for _, p := range decoded.LLDP.Port {
-				if p.DestMAC != lldpMulticastMAC {
-					t.Fatalf("dest-mac-address = %q, want %q", p.DestMAC, lldpMulticastMAC)
-				}
-				byIf[p.Name] = len(p.RemoteSystems)
-			}
+	if len(out.Port) != 1 {
+		t.Fatalf("port count = %d, want 1", len(out.Port))
+	}
+	p := out.Port[0]
+	if p.Name != "eth0" || len(p.RemoteSystems) != 1 {
+		t.Fatalf("unexpected port: %#v", p)
+	}
+	r := p.RemoteSystems[0]
+	if r.ChassisID != "aa:bb:cc:dd:ee:ff" || r.PortID != "swp1" || r.RemoteIndex != 7 {
+		t.Fatalf("unexpected remote: %#v", r)
+	}
+}
 
-			if !reflect.DeepEqual(byIf, tt.wantByIfName) {
-				t.Fatalf("neighbors by if mismatch\n got: %#v\nwant: %#v", byIf, tt.wantByIfName)
-			}
+func TestTransformNeighborsEmpty(t *testing.T) {
+	for name, in := range map[string]string{
+		"empty table json0": `{"lldp": [{}]}`,
+		"empty object":      `{}`,
+		"malformed":         `{not-json`,
+	} {
+		raw := transformNeighbors([]byte(in))
+		if string(raw) != "{}" {
+			t.Errorf("%s: got %s, want {}", name, raw)
+		}
+	}
+}
 
-			if len(tt.wantFirst) > 0 {
-				first := decoded.LLDP.Port[0].RemoteSystems[0]
-				gotFirst := map[string]string{
-					"chassis-id-subtype": first.ChassisIDSubtype,
-					"port-id-subtype":    first.PortIDSubtype,
-					"chassis-id":         first.ChassisID,
-					"port-id":            first.PortID,
-				}
-				if !reflect.DeepEqual(gotFirst, tt.wantFirst) {
-					t.Fatalf("first remote mismatch\n got: %#v\nwant: %#v", gotFirst, tt.wantFirst)
-				}
-			}
-		})
+// A neighbor that disappears between reads must vanish from the tree:
+// every update is a full-table replace.
+func TestUpdateTreeClearsRemovedNeighbors(t *testing.T) {
+	tr := tree.New()
+	m := New(tr, nil)
+
+	m.query = func(context.Context) ([]byte, error) {
+		return []byte(showNeighborsJSON0), nil
+	}
+	m.updateTree(context.Background())
+	if out := decode(t, tr.Get(treeKey)); len(out.Port) != 2 {
+		t.Fatalf("expected 2 ports after first read, got %d", len(out.Port))
+	}
+
+	m.query = func(context.Context) ([]byte, error) {
+		return []byte(`{"lldp": [{}]}`), nil
+	}
+	m.updateTree(context.Background())
+	if out := decode(t, tr.Get(treeKey)); len(out.Port) != 0 {
+		t.Fatalf("stale neighbors not cleared: %d ports remain", len(out.Port))
+	}
+}
+
+// A failing query must leave the previous data untouched.
+func TestUpdateTreeQueryErrorKeepsData(t *testing.T) {
+	tr := tree.New()
+	m := New(tr, nil)
+
+	m.query = func(context.Context) ([]byte, error) {
+		return []byte(showNeighborsJSON0), nil
+	}
+	m.updateTree(context.Background())
+	before := string(tr.Get(treeKey))
+
+	m.query = func(context.Context) ([]byte, error) {
+		return nil, errors.New("lldpcli gone")
+	}
+	m.updateTree(context.Background())
+
+	if after := string(tr.Get(treeKey)); after != before {
+		t.Fatal("query error overwrote existing lldp data")
+	}
+}
+
+// Watch events are triggers only: added/updated/deleted all request a
+// refresh, unknown events do not.
+func TestProcessEventTriggers(t *testing.T) {
+	m := New(tree.New(), nil)
+
+	drain := func() {
+		select {
+		case <-m.refresh:
+		default:
+		}
+	}
+
+	for _, ev := range []string{"lldp-added", "lldp-updated", "lldp-deleted"} {
+		drain()
+		m.processEvent([]byte(`{"` + ev + `": {"lldp": {}}}`))
+		select {
+		case <-m.refresh:
+		default:
+			t.Errorf("%s did not trigger refresh", ev)
+		}
+	}
+
+	drain()
+	m.processEvent([]byte(`{"lldp-unknown": {}}`))
+	select {
+	case <-m.refresh:
+		t.Error("unknown event triggered refresh")
+	default:
 	}
 }