feat(node): push USB changes and preserve sensor setup across sessions

2026-09-06 00:24:48 +03:00
parent a8647c4d87
commit 22d69107ba
24 changed files with 636 additions and 60 deletions
@@ -238,6 +238,12 @@ func (p *Pairing) send(ctx context.Context, b CoreBinding, path string, payload
 }
 func (p *Pairing) channel(ctx context.Context) {
 	instance := "agent_" + token()
+	var changed <-chan struct{}
+	if p.Sensors != nil {
+		var unsubscribe func()
+		changed, unsubscribe = p.Sensors.events.subscribe()
+		defer unsubscribe()
+	}
 	timer := time.NewTicker(5 * time.Second)
 	defer timer.Stop()
 	defer func() {
@@ -335,6 +341,12 @@ func (p *Pairing) channel(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case <-timer.C:
+		case <-changed:
+			select {
+			case <-ctx.Done():
+				return
+			case <-time.After(250 * time.Millisecond):
+			}
 		}
 	}
 }
@@ -0,0 +1,137 @@
+package node
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"os/exec"
+	"strings"
+	"sync"
+	"time"
+)
+
+// Events are hints to reconcile OS/SDK state, never commands or trusted inventory.
+// A bounded latest-state signal prevents a slow viewer from blocking discovery.
+type sensorEvents struct {
+	mu        sync.Mutex
+	listeners map[chan struct{}]bool
+}
+
+func (e *sensorEvents) subscribe() (<-chan struct{}, func()) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	if e.listeners == nil {
+		e.listeners = map[chan struct{}]bool{}
+	}
+	c := make(chan struct{}, 1)
+	e.listeners[c] = true
+	return c, func() { e.mu.Lock(); delete(e.listeners, c); e.mu.Unlock() }
+}
+func (e *sensorEvents) notify() {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	for c := range e.listeners {
+		select {
+		case c <- struct{}{}:
+		default:
+		}
+	}
+}
+
+func usbEvents(input io.Reader, changed func()) {
+	scanner := bufio.NewScanner(input)
+	fields := map[string]string{}
+	for scanner.Scan() {
+		line := scanner.Text()
+		if line == "" {
+			if fields["SUBSYSTEM"] == "usb" && fields["DEVTYPE"] == "usb_device" &&
+				(fields["ACTION"] == "add" || fields["ACTION"] == "remove" || fields["ACTION"] == "change") {
+				changed()
+			}
+			fields = map[string]string{}
+		} else if key, value, ok := strings.Cut(line, "="); ok && len(fields) < 128 {
+			fields[key] = value
+		}
+	}
+}
+
+func (s *Sensors) WatchUSB(ctx context.Context) {
+	for ctx.Err() == nil {
+		cmd := exec.CommandContext(ctx, "/usr/bin/udevadm", "monitor", "--udev", "--subsystem-match=usb", "--property")
+		cmd.Env = []string{"PATH=/usr/bin:/bin", "LANG=C"}
+		pipe, err := cmd.StdoutPipe()
+		if err == nil && cmd.Start() == nil {
+			usbEvents(pipe, s.events.notify)
+			_ = cmd.Wait()
+		}
+		// A failed monitor cannot disable the existing heartbeat reconciliation.
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(30 * time.Second):
+		}
+	}
+}
+
+func (s *Sensors) stream(w http.ResponseWriter, r *http.Request, server *Server) {
+	if !server.authorized(w, r) {
+		return
+	}
+	if _, ok := w.(http.Flusher); !ok {
+		http.Error(w, "Streaming unavailable", 503)
+		return
+	}
+	changed, unsubscribe := s.events.subscribe()
+	defer unsubscribe()
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("X-Accel-Buffering", "no")
+	controller := http.NewResponseController(w)
+	timer := time.NewTimer(0)
+	defer timer.Stop()
+	for {
+		select {
+		case <-r.Context().Done():
+			return
+		case <-changed:
+			// USB devices expose several interfaces; coalesce the enumeration burst.
+			select {
+			case <-r.Context().Done():
+				return
+			case <-time.After(250 * time.Millisecond):
+			}
+		case <-timer.C:
+		}
+		if !server.authorized(w, r) {
+			return
+		}
+		value := s.Inventory()
+		data, err := json.Marshal(value)
+		if err != nil {
+			return
+		}
+		_ = controller.SetWriteDeadline(time.Now().Add(10 * time.Second))
+		if _, err = fmt.Fprintf(w, "retry: 3000\ndata: %s\n\n", data); err != nil {
+			return
+		}
+		if controller.Flush() != nil {
+			return
+		}
+		delay := 15 * time.Second
+		for _, raw := range value["operations"].([]any) {
+			if raw.(map[string]any)["state"] == "running" {
+				delay = 2 * time.Second
+				break
+			}
+		}
+		if !timer.Stop() {
+			select {
+			case <-timer.C:
+			default:
+			}
+		}
+		timer.Reset(delay)
+	}
+}
@@ -0,0 +1,36 @@
+package node
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestUSBEventsAreHintsAndDoNotMultiplyInterfaces(t *testing.T) {
+	input := "UDEV [1] add /devices/example (usb)\nACTION=add\nSUBSYSTEM=usb\nDEVTYPE=usb_device\n\n" +
+		"ACTION=add\nSUBSYSTEM=usb\nDEVTYPE=usb_interface\n\n" +
+		"ACTION=remove\nSUBSYSTEM=usb\nDEVTYPE=usb_device\n\n" +
+		"ACTION=add\nSUBSYSTEM=net\nDEVTYPE=usb_device\n\n"
+	e := sensorEvents{}
+	first, closeFirst := e.subscribe()
+	second, closeSecond := e.subscribe()
+	defer closeSecond()
+	count := 0
+	usbEvents(strings.NewReader(input), func() { count++; e.notify() })
+	if count != 2 || len(first) != 1 || len(second) != 1 {
+		t.Fatalf("events=%d first=%d second=%d", count, len(first), len(second))
+	}
+	<-first
+	closeFirst()
+	e.notify()
+	if len(first) != 0 {
+		t.Fatal("closed viewer still receives events")
+	}
+}
+
+func TestSensorEventStreamRequiresLocalSession(t *testing.T) {
+	s := newTestServer(t)
+	s.Sensors, _ = OpenSensors(t.TempDir(), "node_test")
+	if response := call(s, "GET", "/api/devices/events", "", nil); response.Code != 401 {
+		t.Fatal("unauthenticated device stream", response.Code)
+	}
+}
@@ -45,14 +45,16 @@ type SensorOperation struct {
 	Updated int64         `json:"updated_at"`
 }
 type Sensors struct {
-	mu         sync.Mutex
-	prepareMu  sync.Mutex
-	root       string
-	nodeID     string
-	instance   string
-	client     *http.Client
-	operations map[string]*SensorOperation
-	names      map[string]string
+	events      sensorEvents
+	mu          sync.Mutex
+	prepareMu   sync.Mutex
+	root        string
+	nodeID      string
+	instance    string
+	client      *http.Client
+	operations  map[string]*SensorOperation
+	names       map[string]string
+	initialized map[string]bool
 }

 var sensorID = regexp.MustCompile(`^rsd455_[0-9a-f]{32}$`)
@@ -63,7 +65,7 @@ func OpenSensors(root, nodeID string) (*Sensors, error) {
 	if e := os.MkdirAll(dir, 0700); e != nil {
 		return nil, e
 	}
-	s := &Sensors{root: dir, nodeID: nodeID, instance: "discovery_" + digest(token())[:24], operations: map[string]*SensorOperation{}, names: map[string]string{}}
+	s := &Sensors{root: dir, nodeID: nodeID, instance: "discovery_" + digest(token())[:24], operations: map[string]*SensorOperation{}, names: map[string]string{}, initialized: map[string]bool{}}
 	s.client = &http.Client{Timeout: 25 * time.Second, Transport: &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
 		return (&net.Dialer{}).DialContext(ctx, "unix", "/run/mission-core-sensors/driver.sock")
 	}}}
@@ -85,6 +87,19 @@ func OpenSensors(root, nodeID string) (*Sensors, error) {
 	}
 	data, _ := os.ReadFile(filepath.Join(dir, "names.json"))
 	_ = json.Unmarshal(data, &s.names)
+	data, _ = os.ReadFile(filepath.Join(dir, "initialized.json"))
+	_ = json.Unmarshal(data, &s.initialized)
+	if s.initialized == nil {
+		s.initialized = map[string]bool{}
+	}
+	for _, op := range s.operations {
+		if op.Command.Action == "prepare" && op.State == "complete" {
+			s.initialized[op.Command.Session.DeviceID] = true
+		}
+	}
+	if e := s.write("initialized.json", s.initialized); e != nil {
+		return nil, e
+	}
 	return s, nil
 }
 func (s *Sensors) write(name string, value any) error {
@@ -163,6 +178,7 @@ func (s *Sensors) Inventory() map[string]any {
 				id, _ := item["id"].(string)
 				seen[id] = true
 				s.mu.Lock()
+				item["configured"] = s.initialized[id]
 				if n := s.names[id]; n != "" {
 					item["name"] = n
 				}
@@ -189,15 +205,21 @@ func (s *Sensors) Inventory() map[string]any {
 		if seen[id] {
 			continue
 		}
-		now := time.Now().UTC().Format(time.RFC3339Nano)
-		name := "RealSense D455"
-		s.mu.Lock()
-		if n := s.names[id]; n != "" {
-			name = n
-		}
-		s.mu.Unlock()
-		items = append(items, map[string]any{"id": id, "name": name, "model": "RealSense D455", "prepared": false, "verified": false, "online": true, "usb": read("speed") + " Мбит/с", "layers": []any{}, "snapshot": map[string]any{"context": map[string]any{"session_id": s.instance + "_" + id, "device": map[string]any{"device_id": id, "model": map[string]string{"plugin_id": "missioncore.realsense", "plugin_version": "0.6.6", "model_id": "realsense.d455"}, "stability": "stable", "basis": "hardware-identifier"}, "execution": map[string]string{"node_id": s.nodeID, "agent_instance_id": s.instance, "platform": "linux"}, "opened_at": now}, "revision": 0, "enrollment": "empty", "connectivity": "connected", "acquisition": "idle", "observed_at": now}})
+		seen[id] = true
+		items = append(items, s.discovery(id, read("speed")+" Мбит/с", true))
 	}
+	s.mu.Lock()
+	configured := []string{}
+	for id, ready := range s.initialized {
+		if ready && sensorID.MatchString(id) && !seen[id] {
+			configured = append(configured, id)
+		}
+	}
+	s.mu.Unlock()
+	for _, id := range configured {
+		items = append(items, s.discovery(id, "—", false))
+	}
+
 	var preparation any
 	if data, e := os.ReadFile("/var/lib/mission-core-node-drivers/preparation.json"); e == nil && len(data) < 32768 {
 		_ = json.Unmarshal(data, &preparation)
@@ -212,6 +234,26 @@ func (s *Sensors) Inventory() map[string]any {
 	s.mu.Unlock()
 	return map[string]any{"schema": "missioncore.node.devices/v1", "items": items, "preparation": preparation, "operations": operations}
 }
+func (s *Sensors) discovery(id, speed string, online bool) map[string]any {
+	now := time.Now().UTC().Format(time.RFC3339Nano)
+	s.mu.Lock()
+	name, configured := s.names[id], s.initialized[id]
+	s.mu.Unlock()
+	if name == "" {
+		name = "RealSense D455"
+	}
+	connectivity, enrollment := "offline", "empty"
+	if online {
+		connectivity = "connected"
+	}
+	if configured {
+		enrollment = "enrolled"
+	}
+	return map[string]any{"id": id, "name": name, "model": "RealSense D455", "configured": configured, "prepared": false, "verified": false, "online": online, "usb": speed, "layers": []any{}, "snapshot": map[string]any{
+		"context":  map[string]any{"session_id": s.instance + "_" + id, "device": map[string]any{"device_id": id, "model": map[string]string{"plugin_id": "missioncore.realsense", "plugin_version": "0.6.6", "model_id": "realsense.d455"}, "stability": "stable", "basis": "hardware-identifier"}, "execution": map[string]string{"node_id": s.nodeID, "agent_instance_id": s.instance, "platform": "linux"}, "opened_at": now},
+		"revision": 0, "enrollment": enrollment, "connectivity": connectivity, "acquisition": "idle", "observed_at": now}}
+}
+
 func sensorViewAction(action string) bool {
 	return action == "details" || action == "offer" || action == "close-peer"
 }
@@ -267,10 +309,12 @@ func (s *Sensors) Submit(c SensorCommand, remote bool) (*SensorOperation, error)
 	}
 	s.operations[c.ID] = value
 	copy := *value
+	s.events.notify()
 	go s.execute(c)
 	return &copy, nil
 }
 func (s *Sensors) execute(c SensorCommand) {
+	defer s.events.notify()
 	var result any
 	var err error
 	uncertain := false
@@ -313,6 +357,13 @@ func (s *Sensors) execute(c SensorCommand) {
 	}
 	s.mu.Lock()
 	defer s.mu.Unlock()
+	if err == nil && c.Action == "prepare" {
+		s.initialized[c.Session.DeviceID] = true
+		if e := s.write("initialized.json", s.initialized); e != nil {
+			err = e
+			uncertain = true
+		}
+	}
 	v := s.operations[c.ID]
 	v.Updated = time.Now().Unix()
 	if err != nil {
@@ -394,6 +445,7 @@ func (s *Sensors) RemoteResults() []any {
 	return out
 }
 func (s *Sensors) Routes(mux *http.ServeMux, server *Server) {
+	mux.HandleFunc("GET /api/devices/events", func(w http.ResponseWriter, r *http.Request) { s.stream(w, r, server) })
 	mux.HandleFunc("GET /api/devices", func(w http.ResponseWriter, r *http.Request) {
 		if server.authorized(w, r) {
 			reply(w, 200, s.Inventory())
@@ -84,3 +84,28 @@ func TestSensorPreservesUncertainDriverOutcome(t *testing.T) {
 		})
 	}
 }
+
+func TestSensorConfiguredIdentitySurvivesRestartAndDisconnect(t *testing.T) {
+	root := t.TempDir()
+	s, _ := OpenSensors(root, "node_test")
+	c := sensorTestCommand()
+	s.initialized[c.Session.DeviceID] = true
+	if e := s.write("initialized.json", s.initialized); e != nil {
+		t.Fatal(e)
+	}
+	s, e := OpenSensors(root, "node_test")
+	if e != nil {
+		t.Fatal(e)
+	}
+	s.client = &http.Client{Transport: sensorRoundTrip(func(r *http.Request) (*http.Response, error) {
+		return &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(`{"items":[]}`)), Header: http.Header{}}, nil
+	})}
+	items := s.Inventory()["items"].([]any)
+	if len(items) != 1 {
+		t.Fatalf("configured camera disappeared: %d", len(items))
+	}
+	item := items[0].(map[string]any)
+	if item["id"] != c.Session.DeviceID || item["online"] != false || item["configured"] != true {
+		t.Fatalf("incorrect offline identity: %+v", item)
+	}
+}