add: processing controls — start/hold, pause/resume, retry

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
This commit is contained in:
Esa Kataja
2026-06-21 20:38:21 +03:00
parent 8fc00c96b9
commit 04d6c85712
8 changed files with 349 additions and 29 deletions
+27 -1
View File
@@ -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{