From 04d6c85712f3cc2b3e769552e40c1d19434874c8 Mon Sep 17 00:00:00 2001 From: Esa Kataja Date: Sun, 21 Jun 2026 20:38:21 +0300 Subject: [PATCH] =?UTF-8?q?add:=20processing=20controls=20=E2=80=94=20star?= =?UTF-8?q?t/hold,=20pause/resume,=20retry?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three lifecycle controls on the dashboard, sharing one /api surface and the status feed: - Start/hold gate (#11): the watcher tracks the queue but holds processing until POST /api/start. Held by default; AV1DAE_AUTOSTART=1 restores start- on-boot. NOTE: flips the previous auto-start default. - Pause/resume the active encode (#10): SIGSTOP/SIGCONT on the ffmpeg process (libsvtav1 is in-process, so one signal suspends all its threads). Resumes exactly where it left off; the tracker excludes paused time from elapsed. - Retry a failed file (#3): POST /api/retry?file=NAME moves it from failed/ back to input/, with a base-name guard against path traversal. /status now reports running + the failed list; snapshots carry a paused flag. Verified live: held queue, retry move, traversal -> 400, and a real encode suspending to process state T on pause and S on resume. Closes #3 Closes #10 Closes #11 --- cmd/av1dae/main.go | 33 +++++++++- docker-compose.yml | 5 +- internal/encoder/encoder.go | 59 ++++++++++++++++++ internal/server/index.html | 68 ++++++++++++++++++++- internal/server/server.go | 107 +++++++++++++++++++++++++++++---- internal/status/status.go | 57 ++++++++++++++---- internal/status/status_test.go | 21 +++++++ internal/watcher/watcher.go | 28 ++++++++- 8 files changed, 349 insertions(+), 29 deletions(-) diff --git a/cmd/av1dae/main.go b/cmd/av1dae/main.go index 520b8d6..33bb9f3 100644 --- a/cmd/av1dae/main.go +++ b/cmd/av1dae/main.go @@ -91,8 +91,29 @@ func main() { ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer cancel() + // Start/hold gate: held by default so the user clicks Start; AV1DAE_AUTOSTART + // (truthy) restores start-on-boot. + autostart := envTruthy(os.Getenv("AV1DAE_AUTOSTART")) + w := watcher.New(cfg.Paths.Input, 15, autostart) + if !autostart { + log.Info("Queue held on startup — click Start (set AV1DAE_AUTOSTART=1 to auto-start)") + } + + controls := server.Controls{ + Running: w.Running, + SetRunning: w.SetRunning, + Pause: enc.Pause, + Resume: enc.Resume, + RetryFailed: func(name string) error { + if name == "" || name != filepath.Base(name) { + return fmt.Errorf("invalid file name") + } + return mover.Rename(filepath.Join(cfg.Paths.Failed, name), filepath.Join(cfg.Paths.Input, name)) + }, + } + if addr := *cfg.HTTPAddr; addr != "" { - srv := &http.Server{Addr: addr, Handler: server.New(tracker, log, store, cfg.Paths.Input).Handler()} + srv := &http.Server{Addr: addr, Handler: server.New(tracker, log, store, controls, cfg.Paths.Input, cfg.Paths.Failed).Handler()} go func() { log.Info(fmt.Sprintf("Status server listening on %s", addr)) if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { @@ -107,7 +128,6 @@ func main() { }() } - w := watcher.New(cfg.Paths.Input, 15) w.Start(ctx, func(ctx context.Context, inputPath string) error { return processFile(ctx, inputPath, cfg, enc, metaClient, log, tracker, store) }) @@ -269,6 +289,15 @@ func failToFailed(path, failedDir string, log *logger.Logger) { } } +// envTruthy reports whether an env var is set to a truthy value. +func envTruthy(s string) bool { + switch strings.ToLower(strings.TrimSpace(s)) { + case "1", "true", "yes", "on": + return true + } + return false +} + // toStatusStreams maps probed source streams to the display shape, keeping only // audio and subtitle streams (the video stream isn't shown). func toStatusStreams(streams []encoder.StreamMetadata) []status.Stream { diff --git a/docker-compose.yml b/docker-compose.yml index c61e110..0ddd0bd 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -12,7 +12,10 @@ services: cpus: "8.0" memory: 4g ports: - - "8080:8080" # status server: GET http://host:8080/status (needs http_addr: ":8080" in config) + - "8080:8080" # status server + dashboard at http://host:8080 (needs http_addr: ":8080" in config) + # By default the queue is HELD on boot — click Start in the UI. Uncomment to auto-start: + # environment: + # - AV1DAE_AUTOSTART=1 volumes: - ./data/config.yaml:/config/config.yaml:ro # your config — paths inside must point at /data/* - ./data/media:/data # holds input/ output/ originals/ failed/ work/ + logs diff --git a/internal/encoder/encoder.go b/internal/encoder/encoder.go index 9ce04a7..17d1de4 100644 --- a/internal/encoder/encoder.go +++ b/internal/encoder/encoder.go @@ -5,12 +5,15 @@ import ( "context" "encoding/json" "fmt" + "os" "os/exec" "path/filepath" "regexp" "sort" "strconv" "strings" + "sync" + "syscall" "time" "av1dae/internal/logger" @@ -26,6 +29,59 @@ type Encoder struct { opusencPath string log *logger.Logger tracker *status.Tracker + + procMu sync.Mutex + cur *os.Process // the in-flight video encode, for pause/resume +} + +// Pause freezes the active video encode in place via SIGSTOP. libsvtav1 runs in +// the ffmpeg process (no forked children), so one signal suspends all its +// threads. No-op if nothing is encoding. The process keeps its memory and +// partial output and resumes exactly where it left off. +func (e *Encoder) Pause() error { + e.procMu.Lock() + defer e.procMu.Unlock() + if e.cur == nil { + return nil + } + if err := e.cur.Signal(syscall.SIGSTOP); err != nil { + return err + } + if e.tracker != nil { + e.tracker.SetPaused(true) + } + return nil +} + +// Resume thaws a paused encode via SIGCONT. No-op if nothing is encoding. +func (e *Encoder) Resume() error { + e.procMu.Lock() + defer e.procMu.Unlock() + if e.cur == nil { + return nil + } + if err := e.cur.Signal(syscall.SIGCONT); err != nil { + return err + } + if e.tracker != nil { + e.tracker.SetPaused(false) + } + return nil +} + +func (e *Encoder) setProc(p *os.Process) { + e.procMu.Lock() + e.cur = p + e.procMu.Unlock() +} + +func (e *Encoder) clearProc() { + e.procMu.Lock() + e.cur = nil + e.procMu.Unlock() + if e.tracker != nil { + e.tracker.SetPaused(false) + } } type StreamInfo struct { @@ -94,6 +150,9 @@ func (e *Encoder) runCmdProgress(ctx context.Context, label, file, name string, if err := cmd.Start(); err != nil { return nil, err } + // Publish the process so Pause/Resume can signal it; clear on exit. + e.setProc(cmd.Process) + defer e.clearProc() // Reads stdout to EOF (when ffmpeg exits), so Wait below is safe afterwards. var lastLog time.Time diff --git a/internal/server/index.html b/internal/server/index.html index f688440..302ff6c 100644 --- a/internal/server/index.html +++ b/internal/server/index.html @@ -114,6 +114,22 @@ .day:first-child { margin-top:0; } @media (prefers-reduced-motion:reduce){ .dot.ok{animation:none;} .bar.indet>i{animation:none; width:100%; opacity:.4;} } + + /* control bar */ + .controls { display:flex; align-items:center; gap:.6rem; flex-wrap:wrap; margin-bottom:1.25rem; min-height:1px; } + .ctl { font-family:var(--mono); font-size:.82rem; cursor:pointer; border-radius:7px; padding:.45rem .9rem; + background:var(--panel-2); color:var(--text); border:1px solid var(--line); } + .ctl:hover { border-color:var(--amber); color:var(--amber); } + .ctl.primary { background:var(--amber); color:#1a1200; border-color:var(--amber); font-weight:600; } + .ctl.primary:hover { background:var(--amber-soft); color:#1a1200; } + .ctl.small { font-size:.72rem; padding:.25rem .65rem; } + .ctl-state { font-family:var(--mono); font-size:.72rem; color:var(--muted); margin-left:.3rem; text-transform:uppercase; letter-spacing:.08em; } + .ctl-state.paused { color:var(--amber); } + .count { font-family:var(--mono); color:var(--red); } + /* failed list */ + #failed .frow { display:flex; align-items:center; gap:.6rem; padding:.35rem 0; border-bottom:1px solid var(--line); font-family:var(--mono); font-size:.82rem; } + #failed .frow:last-child { border-bottom:0; } + #failed .frow .fn { flex:1; color:var(--text); word-break:break-word; } @@ -124,10 +140,17 @@ connecting +
+
+ +

Queue

@@ -356,17 +379,60 @@ txt.textContent = state === "ok" ? "live" : state === "idle" ? "idle" : state === "err" ? "offline" : "connecting"; } + // ---- runtime controls: start/hold, pause/resume, retry ---- + async function post(path) { + try { await fetch(path, { method: "POST" }); } catch (e) {} + poll(); // reflect the new state immediately + } + + function renderControls(d) { + const c = d.current || {}; + const encoding = c.file && c.phase !== "idle"; + let html = d.running + ? `` + : ``; + if (encoding) { + html += c.paused + ? `` + : ``; + } + const state = !d.running ? "held" : c.paused ? "running · encode paused" : "running"; + html += `${state}`; + $("controls").innerHTML = html; + } + + function renderFailed(failed) { + const card = $("failedCard"); + if (!failed || !failed.length) { card.hidden = true; $("failed").innerHTML = ""; return; } + card.hidden = false; + $("fcount").textContent = "(" + failed.length + ")"; + $("failed").innerHTML = failed.map(f => + `
${esc(f)}
` + ).join(""); + } + + $("controls").addEventListener("click", e => { + const b = e.target.closest("[data-act]"); + if (b) post(b.getAttribute("data-act")); + }); + $("failed").addEventListener("click", e => { + const b = e.target.closest("[data-retry]"); + if (b) post("/api/retry?file=" + encodeURIComponent(b.getAttribute("data-retry"))); + }); + async function poll() { try { const r = await fetch("status", { cache: "no-store" }); if (!r.ok) throw new Error(r.status); const d = await r.json(); pushSpeed(d.current); + renderControls(d); renderJob(d.current); + renderFailed(d.failed); renderQueue(d.queue, d.current); renderEvents(d.recent); const active = d.current && d.current.file && d.current.phase !== "idle"; - setLive(active ? "ok" : "idle"); + setLive(d.current && d.current.paused ? "idle" : active ? "ok" : "idle"); } catch (e) { setLive("err"); } finally { diff --git a/internal/server/server.go b/internal/server/server.go index 30a9963..58db645 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -20,15 +20,27 @@ var indexHTML []byte //go:embed settings.html var settingsHTML []byte -type Server struct { - tracker *status.Tracker - log *logger.Logger - store *settings.Store - inputDir string +// Controls is the runtime-control surface the UI drives, wired in main from the +// watcher (start/hold gate), encoder (pause/resume), and mover (retry). +type Controls struct { + Running func() bool + SetRunning func(bool) + Pause func() error + Resume func() error + RetryFailed func(name string) error } -func New(tracker *status.Tracker, log *logger.Logger, store *settings.Store, inputDir string) *Server { - return &Server{tracker: tracker, log: log, store: store, inputDir: inputDir} +type Server struct { + tracker *status.Tracker + log *logger.Logger + store *settings.Store + controls Controls + inputDir string + failedDir string +} + +func New(tracker *status.Tracker, log *logger.Logger, store *settings.Store, controls Controls, inputDir, failedDir string) *Server { + return &Server{tracker: tracker, log: log, store: store, controls: controls, inputDir: inputDir, failedDir: failedDir} } // Handler returns the mux for all status routes. Phase 3 adds "/" (the HTML @@ -38,10 +50,72 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("/status", s.handleStatus) mux.HandleFunc("/settings", s.handleSettingsPage) mux.HandleFunc("/api/settings", s.handleAPISettings) + mux.HandleFunc("/api/start", s.gateHandler(true)) + mux.HandleFunc("/api/hold", s.gateHandler(false)) + mux.HandleFunc("/api/pause", s.actionHandler(func() error { return s.controls.Pause() }, "Encode paused")) + mux.HandleFunc("/api/resume", s.actionHandler(func() error { return s.controls.Resume() }, "Encode resumed")) + mux.HandleFunc("/api/retry", s.handleRetry) mux.HandleFunc("/", s.handleIndex) return mux } +// gateHandler flips the start/hold gate. POST only. +func (s *Server) gateHandler(run bool) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + methodNotAllowed(w) + return + } + s.controls.SetRunning(run) + if run { + s.log.Info("Queue started via web UI") + } else { + s.log.Info("Queue held via web UI") + } + w.WriteHeader(http.StatusNoContent) + } +} + +// actionHandler wraps a no-arg control action (pause/resume). POST only. +func (s *Server) actionHandler(fn func() error, logMsg string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + methodNotAllowed(w) + return + } + if err := fn(); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + s.log.Info(logMsg + " via web UI") + w.WriteHeader(http.StatusNoContent) + } +} + +// handleRetry moves a named file from failed/ back to input/. POST ?file=NAME. +func (s *Server) handleRetry(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + methodNotAllowed(w) + return + } + name := r.URL.Query().Get("file") + if name == "" { + http.Error(w, "missing file", http.StatusBadRequest) + return + } + if err := s.controls.RetryFailed(name); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + s.log.Info("Retry requested via web UI: " + name) + w.WriteHeader(http.StatusNoContent) +} + +func methodNotAllowed(w http.ResponseWriter) { + w.Header().Set("Allow", "POST") + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) +} + func (s *Server) handleSettingsPage(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/html; charset=utf-8") _, _ = w.Write(settingsHTML) @@ -83,17 +157,14 @@ func (s *Server) handleIndex(w http.ResponseWriter, r *http.Request) { } type statusResponse struct { + Running bool `json:"running"` Current status.Snapshot `json:"current"` Queue []string `json:"queue"` + Failed []string `json:"failed"` Recent []logger.LogEntry `json:"recent"` } func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) { - queue := []string{} - for _, f := range watcher.InputFiles(s.inputDir) { - queue = append(queue, filepath.Base(f)) - } - recent, err := s.log.RecentLogs(50) if err != nil { http.Error(w, "reading logs", http.StatusInternalServerError) @@ -102,8 +173,18 @@ func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(statusResponse{ + Running: s.controls.Running(), Current: s.tracker.Snapshot(), - Queue: queue, + Queue: baseNames(watcher.InputFiles(s.inputDir)), + Failed: baseNames(watcher.InputFiles(s.failedDir)), Recent: recent, }) } + +func baseNames(paths []string) []string { + out := []string{} + for _, p := range paths { + out = append(out, filepath.Base(p)) + } + return out +} diff --git a/internal/status/status.go b/internal/status/status.go index fd579eb..10177cf 100644 --- a/internal/status/status.go +++ b/internal/status/status.go @@ -43,16 +43,19 @@ type Stream struct { // Tracker holds live progress for the active job. Safe for concurrent use: // the encode goroutine writes, HTTP/log readers call Snapshot. type Tracker struct { - mu sync.RWMutex - file string - phase string - meta *JobMeta - streams []Stream - totalSec float64 // source duration; 0 until known - outTime float64 // encoded position in seconds - fps float64 - speed float64 - startedAt time.Time + mu sync.RWMutex + file string + phase string + meta *JobMeta + streams []Stream + totalSec float64 // source duration; 0 until known + outTime float64 // encoded position in seconds + fps float64 + speed float64 + startedAt time.Time + paused bool + pausedAt time.Time // when the current pause began + pausedTotal time.Duration // accumulated paused time this job } // Snapshot is an immutable view of the tracker for readers. @@ -64,6 +67,7 @@ type Snapshot struct { Percent float64 `json:"percent"` FPS float64 `json:"fps"` Speed float64 `json:"speed"` + Paused bool `json:"paused"` ElapsedSec int `json:"elapsed_sec"` ETASec int `json:"eta_sec"` StartedAt time.Time `json:"started_at"` @@ -83,6 +87,26 @@ func (t *Tracker) Begin(file string) { t.streams = nil t.totalSec, t.outTime, t.fps, t.speed = 0, 0, 0, 0 t.startedAt = time.Now() + t.paused = false + t.pausedAt = time.Time{} + t.pausedTotal = 0 +} + +// SetPaused records pause/resume transitions so elapsed time excludes the +// paused span. Idempotent on repeated same-state calls. +func (t *Tracker) SetPaused(p bool) { + t.mu.Lock() + defer t.mu.Unlock() + if p == t.paused { + return + } + if p { + t.pausedAt = time.Now() + } else if !t.pausedAt.IsZero() { + t.pausedTotal += time.Since(t.pausedAt) + t.pausedAt = time.Time{} + } + t.paused = p } func (t *Tracker) SetPhase(p string) { @@ -129,6 +153,9 @@ func (t *Tracker) Idle() { t.streams = nil t.totalSec, t.outTime, t.fps, t.speed = 0, 0, 0, 0 t.startedAt = time.Time{} + t.paused = false + t.pausedAt = time.Time{} + t.pausedTotal = 0 } func (t *Tracker) Snapshot() Snapshot { @@ -141,10 +168,18 @@ func (t *Tracker) Snapshot() Snapshot { Streams: t.streams, FPS: t.fps, Speed: t.speed, + Paused: t.paused, StartedAt: t.startedAt, } if !t.startedAt.IsZero() { - s.ElapsedSec = int(time.Since(t.startedAt).Seconds()) + elapsed := time.Since(t.startedAt) - t.pausedTotal + if t.paused && !t.pausedAt.IsZero() { + elapsed -= time.Since(t.pausedAt) + } + if elapsed < 0 { + elapsed = 0 + } + s.ElapsedSec = int(elapsed.Seconds()) } if t.totalSec > 0 { s.Percent = t.outTime / t.totalSec * 100 diff --git a/internal/status/status_test.go b/internal/status/status_test.go index fbcc4cd..db89dbe 100644 --- a/internal/status/status_test.go +++ b/internal/status/status_test.go @@ -81,6 +81,27 @@ func TestSnapshotPercentAndETA(t *testing.T) { } } +func TestSetPausedFlag(t *testing.T) { + tr := New() + tr.Begin("x.mkv") + if tr.Snapshot().Paused { + t.Fatal("should not start paused") + } + tr.SetPaused(true) + if !tr.Snapshot().Paused { + t.Error("expected paused after SetPaused(true)") + } + tr.SetPaused(true) // idempotent — must not double-count + tr.SetPaused(false) + if tr.Snapshot().Paused { + t.Error("expected not paused after SetPaused(false)") + } + tr.Idle() + if tr.Snapshot().Paused { + t.Error("Idle should clear paused") + } +} + func TestSnapshotIdleNoDivByZero(t *testing.T) { s := New().Snapshot() // no Begin, totalSec 0 if s.Percent != 0 || s.ETASec != 0 || s.Phase != PhaseIdle { diff --git a/internal/watcher/watcher.go b/internal/watcher/watcher.go index 472e439..96dd1c8 100644 --- a/internal/watcher/watcher.go +++ b/internal/watcher/watcher.go @@ -6,6 +6,7 @@ import ( "os" "path/filepath" "sort" + "sync" "time" "av1dae/pkg/types" @@ -59,6 +60,10 @@ type failureRecord struct { type Watcher struct { inputDir string interval time.Duration + // runMu guards running: the start/hold gate. While held, settled files are + // tracked (so the queue is reported) but not handed to processFn. + runMu sync.RWMutex + running bool // seen tracks the last-observed (mtime, size) for every file currently in // the input dir. A file is only handed to processFn once two consecutive // ticks agree on both fields (partial-write protection). @@ -68,7 +73,7 @@ type Watcher struct { failed map[string]failureRecord } -func New(inputDir string, intervalSeconds int) *Watcher { +func New(inputDir string, intervalSeconds int, autostart bool) *Watcher { interval := time.Duration(intervalSeconds) * time.Second if interval < 10*time.Second { interval = 10 * time.Second @@ -79,11 +84,26 @@ func New(inputDir string, intervalSeconds int) *Watcher { return &Watcher{ inputDir: inputDir, interval: interval, + running: autostart, seen: make(map[string]fileStat), failed: make(map[string]failureRecord), } } +// SetRunning toggles the start/hold gate. +func (w *Watcher) SetRunning(v bool) { + w.runMu.Lock() + w.running = v + w.runMu.Unlock() +} + +// Running reports whether the queue is being processed. +func (w *Watcher) Running() bool { + w.runMu.RLock() + defer w.runMu.RUnlock() + return w.running +} + func (w *Watcher) Start(ctx context.Context, processFn func(context.Context, string) error) { ticker := time.NewTicker(w.interval) defer ticker.Stop() @@ -146,6 +166,12 @@ func (w *Watcher) scanAndProcess(ctx context.Context, processFn func(context.Con continue } + if !w.Running() { + // Held: the file stays in the queue (already in nextSeen) and is + // picked up once the user starts processing. + continue + } + if err := processFn(ctx, file); err != nil { fmt.Printf("Error processing %s: %v\n", file, err) nextFailed[file] = failureRecord{