diff --git a/cmd/soundtouch-service/main.go b/cmd/soundtouch-service/main.go index 25bc3ac..c857538 100644 --- a/cmd/soundtouch-service/main.go +++ b/cmd/soundtouch-service/main.go @@ -878,6 +878,12 @@ func setupRouter(server *handlers.Server) *chi.Mux { // because firmware may append path components (e.g. /index.xml). r.Get("/probe/{token}", server.HandleProbeInbound) r.Get("/probe/{token}/*", server.HandleProbeInbound) + // Passive peer-reachability probe. Registers a device IP with the + // in-process observer, nudges :8090/swUpdateCheck, and waits for + // any inbound from that IP. Used post-migration where the daemon + // caches its swUpdateUrl at boot and the active round-trip can't + // reach it without a reboot. + r.Post("/setup/peer-probe/{deviceId}", server.HandlePeerProbe) r.Get("/favicon.ico", func(w http.ResponseWriter, r *http.Request) { r.URL.Path = "/media/favicon-braille.svg" server.HandleMedia()(w, r) diff --git a/pkg/service/handlers/handlers_peer_probe.go b/pkg/service/handlers/handlers_peer_probe.go new file mode 100644 index 0000000..65d2135 --- /dev/null +++ b/pkg/service/handlers/handlers_peer_probe.go @@ -0,0 +1,66 @@ +package handlers + +import ( + "encoding/json" + "net/http" + "time" + + "github.com/go-chi/chi/v5" +) + +// peerProbeTimeout caps how long the passive observer waits for any +// inbound from the device IP after the :8090/swUpdateCheck nudge. 30s +// is comfortable for daemon wake-up latency on slow devices while still +// keeping the panel responsive; result.ElapsedMs surfaces the actual +// observed latency so the budget can be tuned from real data. +const peerProbeTimeout = 30 * time.Second + +// peerProbeResponse is the body of POST /setup/peer-probe/{deviceId}. +type peerProbeResponse struct { + OK bool `json:"ok"` + Result any `json:"result,omitempty"` + Error string `json:"error,omitempty"` +} + +// HandlePeerProbe runs the post-migration passive reachability check. +// Registers interest in the device's IP, nudges :8090/swUpdateCheck, +// and reports whether any inbound from that IP landed within +// peerProbeTimeout. Any inbound counts — on a migrated speaker, DNS +// interception routes the daemon's outbounds (update fan-out, marge, +// BMX) through this service regardless of which URL the daemon +// resolved internally, so the question reduces to "did the device +// dial us at all." +// +// Unlike the deprecated round-trip probe, this handler does not mutate +// device state. It presupposes the speaker is already migrated; the +// pre-flight orchestrator is responsible for only calling it in that +// state. +func (s *Server) HandlePeerProbe(w http.ResponseWriter, r *http.Request) { + deviceID := chi.URLParam(r, "deviceId") + if deviceID == "" { + writeJSONError(w, http.StatusBadRequest, "Device ID is required") + return + } + + deviceIP, err := s.resolveDeviceIDToIP(deviceID) + if err != nil { + writeJSONError(w, http.StatusNotFound, err.Error()) + return + } + + result, err := s.sm.RunPeerReachabilityProbe(deviceIP, s.peerObserver, peerProbeTimeout) + + w.Header().Set("Content-Type", "application/json") + + body := peerProbeResponse{ + OK: err == nil && result != nil && result.Reached, + Result: result, + } + if err != nil { + body.Error = err.Error() + } + + if err := json.NewEncoder(w).Encode(body); err != nil { + http.Error(w, "Failed to encode response", http.StatusInternalServerError) + } +} diff --git a/pkg/service/handlers/middleware_peer_observer.go b/pkg/service/handlers/middleware_peer_observer.go index 8df79a9..a4d8dc7 100644 --- a/pkg/service/handlers/middleware_peer_observer.go +++ b/pkg/service/handlers/middleware_peer_observer.go @@ -4,6 +4,8 @@ import ( "net" "net/http" "time" + + "github.com/gesellix/bose-soundtouch/pkg/service/setup" ) // PeerObserverMiddleware records every incoming request's source IP and @@ -21,9 +23,9 @@ func (s *Server) PeerObserverMiddleware(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { host, _, err := net.SplitHostPort(r.RemoteAddr) if err == nil && host != "" { - s.peerObserver.Signal(host, PeerHit{Path: r.URL.Path, At: time.Now()}) + s.peerObserver.Signal(host, setup.PeerHit{Path: r.URL.Path, At: time.Now()}) } next.ServeHTTP(w, r) }) -} \ No newline at end of file +} diff --git a/pkg/service/handlers/peer_observer.go b/pkg/service/handlers/peer_observer.go index 9fbb2a8..b9e2c22 100644 --- a/pkg/service/handlers/peer_observer.go +++ b/pkg/service/handlers/peer_observer.go @@ -2,16 +2,9 @@ package handlers import ( "sync" - "time" -) -// PeerHit is the payload delivered to a waiter when a request from a -// registered peer IP lands on the service. Path is the chi URL path of -// the inbound; At is the wall-clock time the middleware saw it. -type PeerHit struct { - Path string - At time.Time -} + "github.com/gesellix/bose-soundtouch/pkg/service/setup" +) // peerObserver is the rendezvous between the passive reachability probe // (which registers interest in a device IP and waits for any inbound) @@ -22,24 +15,27 @@ type PeerHit struct { // observer keys on the device's IP — the probe doesn't mutate device // state, so there's no token to thread through the request path. Any // inbound from the IP counts as proof of reachability. +// +// PeerHit and the abstract handle interface live in the setup package +// alongside the probe logic; this type implements that interface. type peerObserver struct { mu sync.Mutex - pending map[string]chan PeerHit + pending map[string]chan setup.PeerHit } func newPeerObserver() *peerObserver { - return &peerObserver{pending: make(map[string]chan PeerHit)} + return &peerObserver{pending: make(map[string]chan setup.PeerHit)} } // Register creates a one-shot buffered channel keyed by IP. The buffer // of 1 lets the middleware deliver the first hit and silently drop // subsequent hits during the wait window without blocking. Caller is // responsible for pairing every Register with Forget. -func (o *peerObserver) Register(ip string) <-chan PeerHit { +func (o *peerObserver) Register(ip string) <-chan setup.PeerHit { o.mu.Lock() defer o.mu.Unlock() - ch := make(chan PeerHit, 1) + ch := make(chan setup.PeerHit, 1) o.pending[ip] = ch return ch @@ -49,7 +45,7 @@ func (o *peerObserver) Register(ip string) <-chan PeerHit { // true when a matching registration existed AND the hit was delivered // (i.e. the channel had buffer space — first hit during the window). // Subsequent hits during the same window return false without blocking. -func (o *peerObserver) Signal(ip string, hit PeerHit) bool { +func (o *peerObserver) Signal(ip string, hit setup.PeerHit) bool { o.mu.Lock() defer o.mu.Unlock() @@ -73,4 +69,4 @@ func (o *peerObserver) Forget(ip string) { defer o.mu.Unlock() delete(o.pending, ip) -} \ No newline at end of file +} diff --git a/pkg/service/handlers/peer_observer_test.go b/pkg/service/handlers/peer_observer_test.go index 8cbdbde..8b394ab 100644 --- a/pkg/service/handlers/peer_observer_test.go +++ b/pkg/service/handlers/peer_observer_test.go @@ -3,6 +3,8 @@ package handlers import ( "testing" "time" + + "github.com/gesellix/bose-soundtouch/pkg/service/setup" ) func TestPeerObserver_RegisterSignalForget(t *testing.T) { @@ -13,14 +15,14 @@ func TestPeerObserver_RegisterSignalForget(t *testing.T) { t.Fatal("Register returned nil channel") } - first := PeerHit{Path: "/updates/soundtouch", At: time.Now()} + first := setup.PeerHit{Path: "/updates/soundtouch", At: time.Now()} if !o.Signal("192.168.1.42", first) { t.Error("Signal returned false for registered IP") } // Second signal while the buffer is still full (no reader yet) drops // silently and returns false — only the first hit per window matters. - if o.Signal("192.168.1.42", PeerHit{Path: "/streaming/x"}) { + if o.Signal("192.168.1.42", setup.PeerHit{Path: "/streaming/x"}) { t.Error("second Signal returned true; expected false (buffer full, undrained)") } @@ -43,7 +45,7 @@ func TestPeerObserver_RegisterSignalForget(t *testing.T) { func TestPeerObserver_UnknownIP(t *testing.T) { o := newPeerObserver() - if o.Signal("10.0.0.1", PeerHit{Path: "/anything"}) { + if o.Signal("10.0.0.1", setup.PeerHit{Path: "/anything"}) { t.Error("Signal returned true for unregistered IP") } } @@ -55,7 +57,7 @@ func TestPeerObserver_SignalIsNonBlocking(t *testing.T) { done := make(chan struct{}) go func() { for i := 0; i < 100; i++ { - o.Signal("192.168.1.42", PeerHit{Path: "/x"}) + o.Signal("192.168.1.42", setup.PeerHit{Path: "/x"}) } close(done) }() @@ -66,4 +68,4 @@ func TestPeerObserver_SignalIsNonBlocking(t *testing.T) { case <-time.After(500 * time.Millisecond): t.Fatal("Signal blocked when buffer was full — must drop silently") } -} \ No newline at end of file +} diff --git a/pkg/service/setup/peer_probe.go b/pkg/service/setup/peer_probe.go new file mode 100644 index 0000000..fc2f745 --- /dev/null +++ b/pkg/service/setup/peer_probe.go @@ -0,0 +1,94 @@ +package setup + +import ( + "errors" + "fmt" + "time" +) + +// PeerHit is the payload the observer middleware delivers to a probe +// waiter when a request from a registered peer IP lands on the service. +type PeerHit struct { + Path string + At time.Time +} + +// PeerObserverHandle is the abstract view of the peer-observer registry +// the probe needs: register interest in an IP, eventually forget it. +// The handlers package's peerObserver satisfies this implicitly. +type PeerObserverHandle interface { + Register(ip string) <-chan PeerHit + Forget(ip string) +} + +// PeerProbeResult is the JSON-serializable outcome of a passive +// reachability probe. Reached is the canonical success bit the UI keys +// off; ObservedPath and ElapsedMs are diagnostic. +type PeerProbeResult struct { + Reached bool `json:"reached"` + ObservedPath string `json:"observed_path,omitempty"` + ElapsedMs int64 `json:"elapsed_ms"` +} + +// RunPeerReachabilityProbe is the post-migration reachability check +// that replaces the active swUpdateUrl round-trip. The sequence: +// +// 1. Register the device IP with the observer. +// 2. Nudge :8090/swUpdateCheck on the device to make the swUpdate +// daemon fan out *something* sooner than its own ~5min timer. +// 3. Wait up to timeout for any inbound from that IP. +// +// Any inbound counts as proof of reachability — on a migrated speaker, +// DNS interception means the daemon's outbounds (update fan-out, marge +// polls, BMX registry calls) all funnel through this service regardless +// of which URL the daemon resolved internally. We don't need a specific +// URL to land; we just need *the device* to dial us. +// +// The nudge is fire-and-forget. If :8090 is unreachable, the request +// returns quickly and we still wait for the daemon's own next fan-out +// (or time out). No state on the device is mutated; the probe is safe +// to re-run. +func (m *Manager) RunPeerReachabilityProbe(deviceIP string, observer PeerObserverHandle, timeout time.Duration) (*PeerProbeResult, error) { + if observer == nil { + return nil, errors.New("peer probe not configured: observer is nil") + } + + if deviceIP == "" { + return nil, errors.New("peer probe: deviceIP is required") + } + + hitCh := observer.Register(deviceIP) + defer observer.Forget(deviceIP) + + // Nudge the device. Fire-and-forget — we don't gate on the response + // because the swUpdateCheck endpoint returns immediately after + // enqueuing, and the daemon's fan-out is what we actually want to + // observe. HTTPGet can be nil in test contexts. + if m.HTTPGet != nil { + swCheckURL := fmt.Sprintf("http://%s:8090/swUpdateCheck", deviceIP) + + go func() { + resp, err := m.HTTPGet(swCheckURL) + if err != nil { + return + } + + _ = resp.Body.Close() + }() + } + + start := time.Now() + result := &PeerProbeResult{} + + select { + case hit := <-hitCh: + result.Reached = true + result.ObservedPath = hit.Path + case <-time.After(timeout): + result.Reached = false + } + + result.ElapsedMs = time.Since(start).Milliseconds() + + return result, nil +} diff --git a/pkg/service/setup/peer_probe_test.go b/pkg/service/setup/peer_probe_test.go new file mode 100644 index 0000000..4ceb54f --- /dev/null +++ b/pkg/service/setup/peer_probe_test.go @@ -0,0 +1,142 @@ +package setup + +import ( + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" +) + +// fakePeerObserver is a deterministic PeerObserverHandle for unit +// tests. It exposes the channel returned from Register so the test can +// signal it manually to simulate a device inbound landing. +type fakePeerObserver struct { + mu sync.Mutex + channels map[string]chan PeerHit + forgotten []string +} + +func newFakePeerObserver() *fakePeerObserver { + return &fakePeerObserver{channels: map[string]chan PeerHit{}} +} + +func (o *fakePeerObserver) Register(ip string) <-chan PeerHit { + o.mu.Lock() + defer o.mu.Unlock() + ch := make(chan PeerHit, 1) + o.channels[ip] = ch + return ch +} + +func (o *fakePeerObserver) Forget(ip string) { + o.mu.Lock() + defer o.mu.Unlock() + delete(o.channels, ip) + o.forgotten = append(o.forgotten, ip) +} + +func (o *fakePeerObserver) signal(ip string, hit PeerHit) { + o.mu.Lock() + defer o.mu.Unlock() + ch, ok := o.channels[ip] + if !ok { + return + } + select { + case ch <- hit: + default: + } +} + +func peerProbeManager(onTrigger func()) *Manager { + return &Manager{ + HTTPGet: func(url string) (*http.Response, error) { + if onTrigger != nil { + onTrigger() + } + rr := httptest.NewRecorder() + rr.WriteHeader(200) + return rr.Result(), nil + }, + } +} + +func TestRunPeerReachabilityProbe_HappyPath(t *testing.T) { + obs := newFakePeerObserver() + + // On nudge, simulate the device fanning out to /updates/soundtouch + // which the middleware would signal as a hit on this IP. + m := peerProbeManager(func() { + obs.signal("192.168.1.42", PeerHit{Path: "/updates/soundtouch", At: time.Now()}) + }) + + result, err := m.RunPeerReachabilityProbe("192.168.1.42", obs, 2*time.Second) + if err != nil { + t.Fatalf("RunPeerReachabilityProbe error: %v", err) + } + if !result.Reached { + t.Error("Reached = false, want true") + } + if result.ObservedPath != "/updates/soundtouch" { + t.Errorf("ObservedPath = %q, want %q", result.ObservedPath, "/updates/soundtouch") + } + if len(obs.forgotten) != 1 || obs.forgotten[0] != "192.168.1.42" { + t.Errorf("Forget not called for IP: forgotten = %v", obs.forgotten) + } +} + +func TestRunPeerReachabilityProbe_Timeout(t *testing.T) { + obs := newFakePeerObserver() + m := peerProbeManager(nil) // nudge fires but device never responds + + start := time.Now() + result, err := m.RunPeerReachabilityProbe("192.168.1.42", obs, 200*time.Millisecond) + elapsed := time.Since(start) + + if err != nil { + t.Fatalf("RunPeerReachabilityProbe error: %v", err) + } + if result.Reached { + t.Error("Reached = true, want false (no hit)") + } + if elapsed < 200*time.Millisecond { + t.Errorf("returned early after %v; expected >= 200ms timeout", elapsed) + } + if len(obs.forgotten) != 1 { + t.Errorf("Forget not called after timeout: forgotten = %v", obs.forgotten) + } +} + +func TestRunPeerReachabilityProbe_NilObserver(t *testing.T) { + m := peerProbeManager(nil) + _, err := m.RunPeerReachabilityProbe("192.168.1.42", nil, time.Second) + if err == nil { + t.Error("expected error for nil observer, got nil") + } +} + +func TestRunPeerReachabilityProbe_EmptyIP(t *testing.T) { + m := peerProbeManager(nil) + obs := newFakePeerObserver() + _, err := m.RunPeerReachabilityProbe("", obs, time.Second) + if err == nil { + t.Error("expected error for empty deviceIP, got nil") + } +} + +func TestRunPeerReachabilityProbe_NilHTTPGetTimesOut(t *testing.T) { + // With nil HTTPGet the nudge is skipped entirely; the probe just + // waits for the device to dial in on its own. Useful in tests and + // in environments where the trigger isn't safe to fire. + m := &Manager{} // HTTPGet nil + obs := newFakePeerObserver() + + result, err := m.RunPeerReachabilityProbe("192.168.1.42", obs, 100*time.Millisecond) + if err != nil { + t.Fatalf("error: %v", err) + } + if result.Reached { + t.Error("Reached = true with no nudge and no signal") + } +}