diff --git a/Makefile b/Makefile index f7d2690..632c88d 100644 --- a/Makefile +++ b/Makefile @@ -19,9 +19,6 @@ DESTBIN := $(PREFIX)/bin/$(BINARY) # `make deploy` work whether or not you prefix them with sudo. SUDO := $(shell [ "$$(id -u)" -eq 0 ] || echo sudo) -# The kiosk session unit restarted by `make deploy`. -KIOSK_UNIT ?= getty@tty1 - .PHONY: all build pi pi32 test vet clean run-tui install deploy uninstall all: build @@ -44,10 +41,12 @@ install: build $(SUDO) install -m 0755 $(BINDIR)/$(BINARY) $(DESTBIN) @echo "installed -> $(DESTBIN)" -## deploy: install, then restart the kiosk session so the wall picks it up +## deploy: install, then restart the running wall in place (no reboot). Run as +## the kiosk user so it reaches the daemon's control socket. Falls back to a +## note if the daemon isn't up yet (it will start on next boot). deploy: install - $(SUDO) systemctl restart $(KIOSK_UNIT) - @echo "restarted $(KIOSK_UNIT); run 'rtsp-streamer status' to check" + @$(DESTBIN) restart || echo "daemon not running; it will start on next boot" + @echo "run 'rtsp-streamer status' to check" ## uninstall: remove the installed binary uninstall: diff --git a/README.md b/README.md index 5856457..c42624e 100644 --- a/README.md +++ b/README.md @@ -144,6 +144,9 @@ rtsp-streamer status # slot health from the running daemon rtsp-streamer layout ls # list layouts (* marks active) rtsp-streamer layout set quad # switch live (persists the choice) rtsp-streamer reload # apply hand-edits to the config live +rtsp-streamer restart # re-exec the daemon in place (picks up a new + # binary; no reboot) +rtsp-streamer logs -n 80 # recent daemon activity incl. mpv exit reasons rtsp-streamer version # baked-in git commit / build date ``` @@ -207,14 +210,21 @@ Then `sudo reboot` and the wall comes up on boot. ### Updating ```sh -git pull && make install && sudo reboot +git pull && make deploy ``` -`make install` builds natively and copies to `/usr/local/bin` (auto-sudo — run -it *without* `sudo` so `go build` keeps your `PATH`). Use **`sudo reboot`**, not -`systemctl restart getty@tty1`: a getty restart tends to leave the old sway + -daemon running alongside the new one (duplicate mpv, windows fighting). Confirm -what's live with `rtsp-streamer version`. +`make deploy` builds natively, copies to `/usr/local/bin` (auto-sudo — run it +*without* `sudo` so `go build` keeps your `PATH`), then runs +`rtsp-streamer restart`, which tells the running daemon to **re-exec itself in +place**: same process, same sway session, now running the new binary — no +reboot, no duplicate sessions. Confirm what's live with `rtsp-streamer version`. + +`restart` works because the daemon is launched under a small relaunch loop in +the sway config, so it also recovers on its own if it ever crashes. (Older +installs that predate this launcher need their sway config refreshed — re-run +`deploy/install.sh` or copy `deploy/sway/config` to +`/etc/rtsp-streamer/sway/config` once, then reboot.) A full `sudo reboot` is +still fine and never wrong. ## Enabling RTSP in UniFi Protect @@ -262,6 +272,12 @@ pegged, work through: ## Troubleshooting +- **A stream keeps going unhealthy / flapping** — `rtsp-streamer logs` shows the + daemon's recent activity, including the *reason* each mpv exited (its stderr + tail: connection refused, unsupported codec, 401, etc.). This is the first + thing to check when one tile restarts repeatedly. A camera whose `video-codec` + is `hevc` can't hardware-decode on a Pi 4 and often stutters/exits under load — + switch it to H.264 in Protect or give the tile the `low` substream. - **`status` can't reach the daemon** (`control.sock: no such file`) — the wall is running as a *different user* than your SSH session (check `ps -o user= -p "$(pgrep -x sway)"` vs `id`). Re-run `install.sh ` @@ -306,7 +322,7 @@ pegged, work through: ## Layout ``` -cmd/rtsp-streamer/ CLI (cobra): daemon, discover, layout, reload, status, tui, config, version +cmd/rtsp-streamer/ CLI (cobra): daemon, discover, layout, reload, restart, logs, status, tui, config, version internal/config/ YAML load/save/validate, schema (tiles, streams), XDG paths internal/protect/ UniFi Protect client (login, bootstrap, enable RTSP, URLs) internal/player/ one supervised mpv per stream, JSON IPC, restart/backoff diff --git a/cmd/rtsp-streamer/commands.go b/cmd/rtsp-streamer/commands.go index 3da1eb9..ad49bb2 100644 --- a/cmd/rtsp-streamer/commands.go +++ b/cmd/rtsp-streamer/commands.go @@ -41,8 +41,8 @@ func daemonCmd() *cobra.Command { Use: "daemon", Short: "Run the video wall (launched by the kiosk session)", RunE: func(cmd *cobra.Command, _ []string) error { - log := newLogger() - d, err := daemon.New(cfgPath, log) + log, ring := newDaemonLogger() + d, err := daemon.New(cfgPath, log, ring) if err != nil { return err } @@ -375,6 +375,51 @@ func layoutCmd() *cobra.Command { return c } +// logsCmd retrieves recent daemon log lines over the control socket — the +// practical way to see why a stream is flapping on the headless kiosk. +func logsCmd() *cobra.Command { + var n int + c := &cobra.Command{ + Use: "logs", + Short: "Show recent daemon log lines (incl. mpv exit reasons)", + RunE: func(cmd *cobra.Command, _ []string) error { + resp, err := ipc.Send(ipc.Request{Cmd: "logs", Count: n}) + if err != nil { + return err + } + if !resp.OK { + return fmt.Errorf("%s", resp.Error) + } + for _, line := range resp.Logs { + fmt.Println(line) + } + return nil + }, + } + c.Flags().IntVarP(&n, "lines", "n", 50, "show at most the most recent N lines (0 = all buffered)") + return c +} + +// restartCmd asks the running daemon to re-exec itself in place, picking up a +// freshly installed binary without a reboot or a new sway session. +func restartCmd() *cobra.Command { + return &cobra.Command{ + Use: "restart", + Short: "Restart the running daemon in place (reloads the binary; no reboot)", + RunE: func(cmd *cobra.Command, _ []string) error { + resp, err := ipc.Send(ipc.Request{Cmd: "restart"}) + if err != nil { + return err + } + if !resp.OK { + return fmt.Errorf("%s", resp.Error) + } + fmt.Println("daemon is restarting in place (picking up the installed binary)…") + return nil + }, + } +} + // reloadCmd tells the running daemon to re-read its config and re-apply the // active layout. The daemon leaves streams untouched when nothing material // changed, so this is safe to run casually. diff --git a/cmd/rtsp-streamer/main.go b/cmd/rtsp-streamer/main.go index e4099a0..483f18b 100644 --- a/cmd/rtsp-streamer/main.go +++ b/cmd/rtsp-streamer/main.go @@ -5,10 +5,12 @@ package main import ( "fmt" + "io" "log/slog" "os" "github.com/lwoodard/rtsp-streamer/internal/config" + "github.com/lwoodard/rtsp-streamer/internal/logbuf" "github.com/spf13/cobra" ) @@ -39,7 +41,9 @@ func main() { daemonCmd(), discoverCmd(), layoutCmd(), + logsCmd(), reloadCmd(), + restartCmd(), statusCmd(), tuiCmd(), configCmd(), @@ -53,17 +57,25 @@ func main() { } } -func newLogger() *slog.Logger { - var lvl slog.Level +func logLevelValue() slog.Level { switch logLevel { case "debug": - lvl = slog.LevelDebug + return slog.LevelDebug case "warn": - lvl = slog.LevelWarn + return slog.LevelWarn case "error": - lvl = slog.LevelError + return slog.LevelError default: - lvl = slog.LevelInfo + return slog.LevelInfo } - return slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: lvl})) +} + +// newDaemonLogger builds the daemon's logger, teeing output to stderr and to +// an in-memory ring so `rtsp-streamer logs` can retrieve recent activity over +// the control socket (the daemon's stderr itself is not reachable on the +// headless kiosk — it goes to sway on tty1). +func newDaemonLogger() (*slog.Logger, *logbuf.Writer) { + ring := logbuf.New(500) + h := slog.NewTextHandler(io.MultiWriter(os.Stderr, ring), &slog.HandlerOptions{Level: logLevelValue()}) + return slog.New(h), ring } diff --git a/deploy/sway/config b/deploy/sway/config index c5384df..4b47af3 100644 --- a/deploy/sway/config +++ b/deploy/sway/config @@ -26,8 +26,12 @@ for_window [app_id="mpv"] floating enable, border none # We do NOT run swayidle, so sway will not blank the output on its own. If your # panel still sleeps, uncomment a keep-alive loop in a systemd timer instead. -### Launch the wall ### -exec rtsp-streamer daemon --log info +### Launch the wall (supervised) ### +# Run the daemon in a relaunch loop so it recovers on its own: if the daemon +# ever crashes it comes straight back, and `rtsp-streamer restart` (in-place +# re-exec) needs no help from here. The loop also means a clean daemon exit +# reappears within ~2s instead of leaving a black screen until reboot. +exec sh -c 'while :; do rtsp-streamer daemon --log info; sleep 2; done' ### Emergency escape hatches (handy while setting up over a keyboard) ### # Super+Return opens a terminal if one is installed; Super+Shift+Q exits sway. diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index df4fc15..8a28914 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -11,12 +11,15 @@ import ( "net" "os" "reflect" + "strings" "sync" + "syscall" "time" "github.com/lwoodard/rtsp-streamer/internal/compositor" "github.com/lwoodard/rtsp-streamer/internal/config" "github.com/lwoodard/rtsp-streamer/internal/ipc" + "github.com/lwoodard/rtsp-streamer/internal/logbuf" "github.com/lwoodard/rtsp-streamer/internal/player" "github.com/lwoodard/rtsp-streamer/internal/protect" "github.com/lwoodard/rtsp-streamer/internal/viewmap" @@ -26,6 +29,7 @@ import ( type Daemon struct { cfgPath string log *slog.Logger + logs *logbuf.Writer // ring of recent log lines, served by "logs" comp *compositor.Client runDir string @@ -39,8 +43,10 @@ type Daemon struct { pcl *protect.Client // cached Protect session for view sync } -// New constructs a daemon bound to a config path. -func New(cfgPath string, log *slog.Logger) (*Daemon, error) { +// New constructs a daemon bound to a config path. logs, when non-nil, is the +// ring the logger tees into so the "logs" control command can serve recent +// activity over the socket. +func New(cfgPath string, log *slog.Logger, logs *logbuf.Writer) (*Daemon, error) { comp, err := compositor.New() if err != nil { return nil, err @@ -49,7 +55,7 @@ func New(cfgPath string, log *slog.Logger) (*Daemon, error) { if err := os.MkdirAll(runDir, 0o700); err != nil { return nil, err } - return &Daemon{cfgPath: cfgPath, log: log, comp: comp, runDir: runDir}, nil + return &Daemon{cfgPath: cfgPath, log: log, logs: logs, comp: comp, runDir: runDir}, nil } // Run starts the wall and blocks until ctx is cancelled. @@ -509,14 +515,20 @@ func (d *Daemon) serveControl(ctx context.Context) { } func (d *Daemon) handleControl(ctx context.Context, conn net.Conn) { - defer conn.Close() _ = conn.SetDeadline(time.Now().Add(30 * time.Second)) var req ipc.Request if err := json.NewDecoder(conn).Decode(&req); err != nil { + conn.Close() return } resp := d.dispatch(ctx, req) _ = json.NewEncoder(conn).Encode(resp) + conn.Close() + // A restart replaces the process image; do it only after the reply has been + // flushed to the client, so `rtsp-streamer restart` sees the ack. + if req.Cmd == "restart" && resp.OK { + d.execRestart(ctx) + } } func (d *Daemon) dispatch(ctx context.Context, req ipc.Request) ipc.Response { @@ -537,11 +549,61 @@ func (d *Daemon) dispatch(ctx context.Context, req ipc.Request) ipc.Response { d.log.Warn("could not persist active layout", "err", err) } return d.status() + case "logs": + if d.logs == nil { + return ipc.Response{OK: false, Error: "log buffer not enabled"} + } + return ipc.Response{OK: true, ActiveLayout: d.layout, Logs: d.logs.Lines(req.Count)} + case "restart": + // Validate we can find the on-disk binary before acking; the actual + // exec happens in handleControl once the reply is sent. + if _, err := restartTarget(); err != nil { + return ipc.Response{OK: false, Error: err.Error()} + } + return ipc.Response{OK: true, ActiveLayout: d.layout} default: return ipc.Response{OK: false, Error: fmt.Sprintf("unknown command %q", req.Cmd)} } } +// restartTarget resolves the path of the binary to re-exec. os.Executable can +// return a "/path (deleted)" sentinel when the file was replaced (as `make +// install` does via rename); we trim that so we exec the freshly installed +// binary at the same path, and Stat confirms it is really there. +func restartTarget() (string, error) { + exe, err := os.Executable() + if err != nil { + return "", err + } + exe = strings.TrimSuffix(exe, " (deleted)") + if _, err := os.Stat(exe); err != nil { + return "", fmt.Errorf("cannot restart: executable %q not found: %w", exe, err) + } + return exe, nil +} + +// execRestart tears down the wall and re-execs the daemon in place: same PID, +// same parent (sway), same environment — so it keeps WAYLAND_DISPLAY/SWAYSOCK +// and comes back up with the newly installed binary, no reboot. On success it +// does not return. If the exec fails, it rebuilds the wall so the screen isn't +// left blank. +func (d *Daemon) execRestart(ctx context.Context) { + exe, err := restartTarget() + if err != nil { + d.log.Error("restart aborted", "err", err) + return + } + d.log.Info("restarting daemon in place", "exe", exe) + d.stopLayout() // kill mpv so processes don't leak across the exec + _ = os.Remove(ipc.SocketPath()) + err = syscall.Exec(exe, os.Args, os.Environ()) + // Only reached if exec failed. + d.log.Error("restart exec failed; rebuilding wall", "err", err) + if rerr := d.reload(ctx); rerr != nil { + d.log.Error("failed to rebuild wall after failed restart", "err", rerr) + } +} + func (d *Daemon) persistActiveLayout(name string) error { cfg, err := config.Load(d.cfgPath) if err != nil { diff --git a/internal/ipc/ipc.go b/internal/ipc/ipc.go index 33c9658..f58b277 100644 --- a/internal/ipc/ipc.go +++ b/internal/ipc/ipc.go @@ -17,8 +17,9 @@ import ( // Request is a command sent to the daemon. type Request struct { - Cmd string `json:"cmd"` // "status" | "reload" | "set-layout" - Name string `json:"name,omitempty"` // layout name for set-layout + Cmd string `json:"cmd"` // "status"|"reload"|"set-layout"|"logs"|"restart" + Name string `json:"name,omitempty"` // layout name for set-layout + Count int `json:"count,omitempty"` // for "logs": most-recent N lines (0 = all) } // SlotStatus reports one grid cell's state. @@ -36,6 +37,7 @@ type Response struct { Error string `json:"error,omitempty"` ActiveLayout string `json:"active_layout,omitempty"` Slots []SlotStatus `json:"slots,omitempty"` + Logs []string `json:"logs,omitempty"` // recent daemon log lines, for "logs" } // SocketPath returns the control socket path inside the runtime dir. diff --git a/internal/logbuf/logbuf.go b/internal/logbuf/logbuf.go new file mode 100644 index 0000000..89381b0 --- /dev/null +++ b/internal/logbuf/logbuf.go @@ -0,0 +1,60 @@ +// Package logbuf provides an in-memory ring of recent log lines. The daemon +// tees its slog output into one of these so `rtsp-streamer logs` can retrieve +// recent activity over the control socket — the headless Pi has no reachable +// place for the daemon's stderr to land (it goes to sway on tty1), so keeping +// a buffer in the process is the practical way to inspect it over SSH. +package logbuf + +import ( + "strings" + "sync" +) + +// Writer is an io.Writer that retains the last N newline-delimited lines +// written to it. It is safe for concurrent use, so it can sit behind an +// io.MultiWriter alongside os.Stderr in a slog handler. +type Writer struct { + mu sync.Mutex + lines []string + max int +} + +// New returns a ring holding at most max lines. +func New(max int) *Writer { + if max < 1 { + max = 1 + } + return &Writer{max: max, lines: make([]string, 0, max)} +} + +// Write appends each complete line in p to the ring, dropping the oldest when +// it overflows. It always reports the full length written so it never trips up +// the MultiWriter it sits behind. +func (w *Writer) Write(p []byte) (int, error) { + w.mu.Lock() + defer w.mu.Unlock() + for _, line := range strings.Split(strings.TrimRight(string(p), "\n"), "\n") { + if line == "" { + continue + } + w.lines = append(w.lines, line) + } + if len(w.lines) > w.max { + w.lines = append(w.lines[:0], w.lines[len(w.lines)-w.max:]...) + } + return len(p), nil +} + +// Lines returns a copy of the buffered lines, oldest first. When n > 0 only +// the most recent n are returned. +func (w *Writer) Lines(n int) []string { + w.mu.Lock() + defer w.mu.Unlock() + start := 0 + if n > 0 && n < len(w.lines) { + start = len(w.lines) - n + } + out := make([]string, len(w.lines)-start) + copy(out, w.lines[start:]) + return out +} diff --git a/internal/logbuf/logbuf_test.go b/internal/logbuf/logbuf_test.go new file mode 100644 index 0000000..8eda16e --- /dev/null +++ b/internal/logbuf/logbuf_test.go @@ -0,0 +1,48 @@ +package logbuf + +import ( + "fmt" + "testing" +) + +func TestRingKeepsMostRecent(t *testing.T) { + w := New(3) + for i := 0; i < 5; i++ { + fmt.Fprintf(w, "line %d\n", i) + } + got := w.Lines(0) + want := []string{"line 2", "line 3", "line 4"} + if len(got) != len(want) { + t.Fatalf("got %d lines, want %d: %v", len(got), len(want), got) + } + for i := range want { + if got[i] != want[i] { + t.Errorf("line %d = %q, want %q", i, got[i], want[i]) + } + } +} + +func TestLinesN(t *testing.T) { + w := New(10) + for i := 0; i < 5; i++ { + fmt.Fprintf(w, "l%d\n", i) + } + if got := w.Lines(2); len(got) != 2 || got[0] != "l3" || got[1] != "l4" { + t.Errorf("Lines(2) = %v, want [l3 l4]", got) + } + if got := w.Lines(100); len(got) != 5 { + t.Errorf("Lines(100) = %d lines, want all 5", len(got)) + } +} + +func TestMultilineAndBlankWrite(t *testing.T) { + w := New(10) + if _, err := w.Write([]byte("a\nb\n\nc\n")); err != nil { + t.Fatal(err) + } + got := w.Lines(0) + want := []string{"a", "b", "c"} // blank line dropped + if len(got) != len(want) { + t.Fatalf("got %v, want %v", got, want) + } +} diff --git a/internal/player/player.go b/internal/player/player.go index ceffed1..41b4b6d 100644 --- a/internal/player/player.go +++ b/internal/player/player.go @@ -13,11 +13,13 @@ import ( "os" "os/exec" "path/filepath" + "strings" "sync" "syscall" "time" "github.com/lwoodard/rtsp-streamer/internal/config" + "github.com/lwoodard/rtsp-streamer/internal/logbuf" ) // Player supervises a single mpv instance bound to one grid slot. @@ -38,6 +40,7 @@ type Player struct { mu sync.Mutex cmd *exec.Cmd started time.Time + stderr *logbuf.Writer // tail of the current process's stderr } // New creates a player. runDir is where the mpv IPC socket lives. @@ -143,15 +146,19 @@ func (p *Player) start(ctx context.Context) error { cmd.WaitDelay = 2 * time.Second // Inherit the caller's environment (WAYLAND_DISPLAY etc. must be set). cmd.Env = os.Environ() - // Discard mpv's chatty stdout/stderr; errors surface via exit code. + // Keep the tail of mpv's stderr so a crash's reason (bad codec, dropped + // connection, auth failure) is logged when the process exits, instead of + // vanishing. Cheap: a small ring, not the whole firehose. + stderr := logbuf.New(30) cmd.Stdout = nil - cmd.Stderr = nil + cmd.Stderr = stderr if err := cmd.Start(); err != nil { return fmt.Errorf("starting mpv: %w", err) } p.mu.Lock() p.cmd = cmd p.started = time.Now() + p.stderr = stderr p.mu.Unlock() p.log.Info("mpv started", "pid", cmd.Process.Pid) return nil @@ -177,6 +184,7 @@ func (p *Player) Supervise(ctx context.Context) { } p.mu.Lock() cmd := p.cmd + stderr := p.stderr p.mu.Unlock() err := cmd.Wait() @@ -187,7 +195,10 @@ func (p *Player) Supervise(ctx context.Context) { if ctx.Err() != nil { return } - p.log.Warn("mpv exited, will restart", "err", err, "after", backoff) + // Surface the last line(s) mpv printed — that's where the failure + // reason lives (connection refused, unsupported codec, 401, ...). + reason := lastLine(stderr) + p.log.Warn("mpv exited, will restart", "err", err, "after", backoff, "reason", reason) if !sleep(ctx, backoff) { return } @@ -271,6 +282,22 @@ func (p *Player) Healthy() bool { return reply["error"] == "success" } +// lastLine returns the most recent non-empty stderr line mpv emitted, trimmed, +// or "" if there was none. mpv prints its fatal error last, so the tail is the +// useful part. +func lastLine(w *logbuf.Writer) string { + if w == nil { + return "" + } + lines := w.Lines(0) + for i := len(lines) - 1; i >= 0; i-- { + if s := strings.TrimSpace(lines[i]); s != "" { + return s + } + } + return "" +} + func sleep(ctx context.Context, d time.Duration) bool { t := time.NewTimer(d) defer t.Stop()