From dfbd74b0d2a91e5cb71aeb67a289336dd9a272b9 Mon Sep 17 00:00:00 2001 From: Esa Kataja Date: Sun, 21 Jun 2026 20:48:37 +0300 Subject: [PATCH] add: queue total ETA from current encode speed The queue card now shows an estimated time to clear: current job remaining + sum(pending source durations) / current speed. Durations come from a path+mtime-keyed cache that probes each file once in the background (ffprobe format=duration, header-only), so the 1s poll never re-probes; entries are pruned to the live queue. No speed (idle/held) -> no estimate; files still being probed mark it partial (shown as '~Xh+'). Closes #9 --- cmd/av1dae/main.go | 7 +- internal/server/index.html | 11 ++- internal/server/server.go | 134 ++++++++++++++++++++++++++++++--- internal/server/server_test.go | 62 +++++++++++++++ 4 files changed, 198 insertions(+), 16 deletions(-) create mode 100644 internal/server/server_test.go diff --git a/cmd/av1dae/main.go b/cmd/av1dae/main.go index 33bb9f3..dd06dc8 100644 --- a/cmd/av1dae/main.go +++ b/cmd/av1dae/main.go @@ -113,7 +113,12 @@ func main() { } if addr := *cfg.HTTPAddr; addr != "" { - srv := &http.Server{Addr: addr, Handler: server.New(tracker, log, store, controls, cfg.Paths.Input, cfg.Paths.Failed).Handler()} + probeDuration := func(p string) (float64, error) { + pctx, pcancel := context.WithTimeout(context.Background(), 10*time.Second) + defer pcancel() + return enc.GetDuration(pctx, p) + } + srv := &http.Server{Addr: addr, Handler: server.New(tracker, log, store, controls, probeDuration, 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 { diff --git a/internal/server/index.html b/internal/server/index.html index 302ff6c..a05a997 100644 --- a/internal/server/index.html +++ b/internal/server/index.html @@ -97,6 +97,7 @@ .qsum { display:flex; align-items:baseline; gap:.5rem; flex-wrap:wrap; margin-bottom:.8rem; } .qsum .qn { font-family:var(--mono); font-size:1.4rem; color:var(--amber); line-height:1; } .qsum .ql { color:var(--muted); font-size:.85rem; } + .qsum .qeta { font-family:var(--mono); font-size:.8rem; color:var(--amber-soft); margin-left:auto; } ul.list { list-style:none; margin:0; padding:0; font-family:var(--mono); font-size:.82rem; } ul.list li { padding:.35rem 0; border-bottom:1px solid var(--line); color:var(--muted); word-break:break-word; } ul.list li:last-child { border-bottom:0; } @@ -325,7 +326,8 @@ for (const s of arr) { while (!s.startsWith(p)) p = p.slice(0, -1); if (!p) break; } return p; } - function renderQueue(queue, current) { + function renderQueue(d) { + const queue = d.queue, current = d.current; const cur = base(current && current.file); const pending = (queue || []).filter(f => f !== cur); const el = $("queue"); @@ -334,7 +336,10 @@ return; } const pre = pending.length > 1 ? commonPrefix(pending) : ""; - const head = `
${pending.length}${pending.length === 1 ? "file" : "files"} waiting
`; + const eta = d.queue_eta_sec > 0 + ? `~${fmtDurLong(d.queue_eta_sec)}${d.queue_eta_partial ? "+" : ""} to clear` + : ""; + const head = `
${pending.length}${pending.length === 1 ? "file" : "files"} waiting${eta}
`; const rows = pending.slice(0, 3).map(f => { const tail = pre ? f.slice(pre.length) : f; return `
  • ${pre ? `${esc(pre)}` : ""}${esc(tail)}
    • `; @@ -429,7 +434,7 @@ renderControls(d); renderJob(d.current); renderFailed(d.failed); - renderQueue(d.queue, d.current); + renderQueue(d); renderEvents(d.recent); const active = d.current && d.current.file && d.current.phase !== "idle"; setLive(d.current && d.current.paused ? "idle" : active ? "ok" : "idle"); diff --git a/internal/server/server.go b/internal/server/server.go index 58db645..ebb4a7a 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -6,7 +6,10 @@ import ( _ "embed" "encoding/json" "net/http" + "os" "path/filepath" + "sync" + "time" "av1dae/internal/logger" "av1dae/internal/settings" @@ -35,12 +38,91 @@ type Server struct { log *logger.Logger store *settings.Store controls Controls + durCache *durationCache 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} +func New(tracker *status.Tracker, log *logger.Logger, store *settings.Store, controls Controls, probeDuration func(string) (float64, error), inputDir, failedDir string) *Server { + return &Server{ + tracker: tracker, + log: log, + store: store, + controls: controls, + durCache: newDurationCache(probeDuration), + inputDir: inputDir, + failedDir: failedDir, + } +} + +type durEntry struct { + mtime time.Time + sec float64 +} + +// durationCache memoizes source durations (keyed by path+mtime) so the queue +// ETA doesn't re-probe every file on every 1s poll. A miss kicks a background +// ffprobe and resolves on a later poll; the file shows as "estimating" until. +type durationCache struct { + probe func(string) (float64, error) + mu sync.Mutex + m map[string]durEntry + inflight map[string]bool +} + +func newDurationCache(probe func(string) (float64, error)) *durationCache { + return &durationCache{probe: probe, m: map[string]durEntry{}, inflight: map[string]bool{}} +} + +func (c *durationCache) Get(path string, mtime time.Time) (float64, bool) { + c.mu.Lock() + if e, ok := c.m[path]; ok && e.mtime.Equal(mtime) { + c.mu.Unlock() + return e.sec, true + } + if c.inflight[path] { + c.mu.Unlock() + return 0, false + } + c.inflight[path] = true + c.mu.Unlock() + + go func() { + sec, err := c.probe(path) + c.mu.Lock() + delete(c.inflight, path) + if err == nil { + c.m[path] = durEntry{mtime, sec} + } + c.mu.Unlock() + }() + return 0, false +} + +// Retain drops cached entries for paths no longer present, bounding the map on +// a long-running daemon. +func (c *durationCache) Retain(paths []string) { + keep := make(map[string]bool, len(paths)) + for _, p := range paths { + keep[p] = true + } + c.mu.Lock() + for p := range c.m { + if !keep[p] { + delete(c.m, p) + } + } + c.mu.Unlock() +} + +// queueETASeconds estimates wall-clock time to clear the queue: the current +// job's remaining time plus each pending file's duration / current speed. +// 0 when there's no speed to extrapolate from (idle/held). +func queueETASeconds(currentETASec int, sumQueuedSec, speed float64) int { + if speed <= 0 { + return 0 + } + return currentETASec + int(sumQueuedSec/speed) } // Handler returns the mux for all status routes. Phase 3 adds "/" (the HTML @@ -157,11 +239,13 @@ 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"` + Running bool `json:"running"` + Current status.Snapshot `json:"current"` + Queue []string `json:"queue"` + Failed []string `json:"failed"` + QueueETASec int `json:"queue_eta_sec"` + QueueETAPartial bool `json:"queue_eta_partial"` + Recent []logger.LogEntry `json:"recent"` } func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) { @@ -171,13 +255,39 @@ func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) { return } + snap := s.tracker.Snapshot() + files := watcher.InputFiles(s.inputDir) + + // Sum durations of pending files (excluding the current job, whose remaining + // time is already in snap.ETASec). Unprobed files mark the estimate partial. + var sumQueued float64 + partial := false + for _, f := range files { + if f == snap.File { + continue + } + info, statErr := os.Stat(f) + if statErr != nil { + partial = true + continue + } + if sec, ok := s.durCache.Get(f, info.ModTime()); ok { + sumQueued += sec + } else { + partial = true + } + } + s.durCache.Retain(files) + w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(statusResponse{ - Running: s.controls.Running(), - Current: s.tracker.Snapshot(), - Queue: baseNames(watcher.InputFiles(s.inputDir)), - Failed: baseNames(watcher.InputFiles(s.failedDir)), - Recent: recent, + Running: s.controls.Running(), + Current: snap, + Queue: baseNames(files), + Failed: baseNames(watcher.InputFiles(s.failedDir)), + QueueETASec: queueETASeconds(snap.ETASec, sumQueued, snap.Speed), + QueueETAPartial: partial, + Recent: recent, }) } diff --git a/internal/server/server_test.go b/internal/server/server_test.go new file mode 100644 index 0000000..646f4d3 --- /dev/null +++ b/internal/server/server_test.go @@ -0,0 +1,62 @@ +package server + +import ( + "testing" + "time" +) + +func TestQueueETASeconds(t *testing.T) { + // No speed → can't extrapolate. + if got := queueETASeconds(100, 3600, 0); got != 0 { + t.Errorf("speed 0: got %d, want 0", got) + } + // current remaining 600s + 1800s of queued source at 0.5x (=3600s) = 4200s. + if got := queueETASeconds(600, 1800, 0.5); got != 4200 { + t.Errorf("got %d, want 4200", got) + } + // At 2x, 1800s of source encodes in 900s; + 600 remaining = 1500. + if got := queueETASeconds(600, 1800, 2); got != 1500 { + t.Errorf("got %d, want 1500", got) + } + // Empty queue → just the current job's remaining. + if got := queueETASeconds(600, 0, 1); got != 600 { + t.Errorf("got %d, want 600", got) + } +} + +// waitCached polls Get (the probe resolves on a background goroutine). +func waitCached(c *durationCache, p string, mt time.Time) (float64, bool) { + for i := 0; i < 200; i++ { + if sec, ok := c.Get(p, mt); ok { + return sec, true + } + time.Sleep(2 * time.Millisecond) + } + return 0, false +} + +func TestDurationCache(t *testing.T) { + c := newDurationCache(func(p string) (float64, error) { return 42, nil }) + + mt := time.Unix(1000, 0) + if _, ok := c.Get("/a.mkv", mt); ok { + t.Fatal("first Get should miss") + } + if sec, ok := waitCached(c, "/a.mkv", mt); !ok || sec != 42 { + t.Fatalf("after probe: got %v ok=%v, want 42 true", sec, ok) + } + + // A changed mtime invalidates the entry. + if _, ok := c.Get("/a.mkv", time.Unix(2000, 0)); ok { + t.Error("changed mtime should miss") + } + if _, ok := waitCached(c, "/a.mkv", time.Unix(2000, 0)); !ok { + t.Fatal("re-probe under new mtime should cache") + } + + // Retain drops entries for paths not in the keep set. + c.Retain([]string{"/b.mkv"}) + if _, ok := c.Get("/a.mkv", time.Unix(2000, 0)); ok { + t.Error("Retain should have dropped /a.mkv") + } +}