From d5c16540a1fa98d1fa9821dbac25dc6e679e64d9 Mon Sep 17 00:00:00 2001 From: Mitchell-Johnson Date: Fri, 23 Oct 2026 16:07:00 -0700 Subject: [PATCH] files: video previews are made on the Transcode device too The setting's device makes every video preview: on the file's own machine when it is there (the hub tells it by header), else on the device's machine, which reads the file through the hub by a one-file token, and the preview is kept in the file's machine's own cache (PUT /api/files/thumb, refused if the file changed). Images and covers stay on the file's machine's CPU. If the device's machine can't, the file's makes it as before. Co-Authored-By: Claude Opus 5.5 --- internal/api/play.go | 3 + internal/api/play_test.go | 42 ++++++ internal/api/server.go | 3 + internal/api/thumbs.go | 246 +++++++++++++++++++++++++++++++++- internal/media/encode_test.go | 14 ++ internal/media/thumbs.go | 25 +++- internal/nodes/nodes.go | 14 +- 7 files changed, 335 insertions(+), 12 deletions(-) diff --git a/internal/api/play.go b/internal/api/play.go index 83fbfa7..94329ae 100644 --- a/internal/api/play.go +++ b/internal/api/play.go @@ -225,6 +225,9 @@ func (s *Server) playPiece(w http.ResponseWriter, r *http.Request) { // transcodeChoice is the setting; nil is the machine the file is on. func (s *Server) transcodeChoice() *store.Transcode { var t *store.Transcode + if s.store == nil { + return nil + } s.store.View(func(st *store.State) { if st.Transcode != nil { c := *st.Transcode diff --git a/internal/api/play_test.go b/internal/api/play_test.go index ad5e740..7a6bd68 100644 --- a/internal/api/play_test.go +++ b/internal/api/play_test.go @@ -1,7 +1,9 @@ package api import ( + "fmt" "net/http/httptest" + "net/url" "os" "path/filepath" "strings" @@ -49,3 +51,43 @@ func TestPlaySrc(t *testing.T) { t.Fatalf("unknown token: %d", w.Code) } } + +// A preview made on another machine is kept by the file's own: only while the file is still +// the one it was made of, and a failure is kept as a failure. +func TestThumbPut(t *testing.T) { + dir := t.TempDir() + p := filepath.Join(dir, "film.mp4") + if err := os.WriteFile(p, []byte("0123456789"), 0o644); err != nil { + t.Fatal(err) + } + fi, _ := os.Stat(p) + s := &Server{fsLocal: files.Local{}, thumbs: newThumbSvc(dir, "")} + v := fmt.Sprintf("%d-%d", fi.ModTime().UnixMilli(), fi.Size()) + put := func(v string, body string, hdr map[string]string) int { + r := httptest.NewRequest("PUT", "/api/files/thumb?path="+url.QueryEscape(p)+"&v="+v, strings.NewReader(body)) + for k, x := range hdr { + r.Header.Set(k, x) + } + w := httptest.NewRecorder() + s.thumbPut(w, r) + return w.Code + } + get := func() *httptest.ResponseRecorder { + r := httptest.NewRequest("GET", "/api/files/thumb?cached=1&path="+url.QueryEscape(p)+"&v="+v, nil) + w := httptest.NewRecorder() + s.filesThumb(w, r) + return w + } + if w := get(); w.Code != 404 { + t.Fatalf("cached=1 before anything: %d", w.Code) + } + if c := put("1-10", "jpeg", nil); c != 409 { + t.Fatalf("a changed file: %d", c) + } + if c := put(v, "jpeg-bytes", map[string]string{"X-Hamlab-Thumb": "grid vaapi:/dev/dri/renderD129"}); c != 204 { + t.Fatalf("put: %d", c) + } + if w := get(); w.Code != 200 || w.Body.String() != "jpeg-bytes" || !strings.Contains(w.Header().Get("X-Hamlab-Thumb"), "grid") { + t.Fatalf("kept: %d %q %q", w.Code, w.Body, w.Header().Get("X-Hamlab-Thumb")) + } +} diff --git a/internal/api/server.go b/internal/api/server.go index dab1a94..8f44110 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -276,6 +276,9 @@ func (s *Server) routes() *http.ServeMux { m.HandleFunc("POST /api/files/delete", s.auth(s.filesDelete)) m.HandleFunc("POST /api/files/chmod", s.auth(s.filesChmod)) m.HandleFunc("GET /api/files/thumb", s.auth(s.filesThumb)) + m.HandleFunc("PUT /api/files/thumb", s.auth(s.thumbPut)) + m.HandleFunc("POST /api/files/thumb-from", s.auth(s.thumbFrom)) + m.HandleFunc("GET /api/nodes/{name}/files/thumb", s.auth(s.hubThumb)) m.HandleFunc("GET /api/files/thumbs", s.auth(s.thumbsStatus)) m.HandleFunc("POST /api/files/thumbs/install", s.auth(s.thumbsInstall)) m.HandleFunc("POST /api/files/thumbs/gpu-access", s.auth(s.thumbsGPUAccess)) diff --git a/internal/api/thumbs.go b/internal/api/thumbs.go index 3c9ebde..44c2519 100644 --- a/internal/api/thumbs.go +++ b/internal/api/thumbs.go @@ -10,10 +10,16 @@ package api // opens every render node, whatever groups the daemon's user has. import ( + "bytes" "context" + "encoding/json" "errors" + "fmt" + "io" "log" "net/http" + "net/http/httptest" + "net/url" "os" "os/exec" "path/filepath" @@ -25,6 +31,7 @@ import ( "hamlab/internal/media" "hamlab/internal/priv" + "hamlab/internal/store" ) const thumbSize = 256 @@ -117,7 +124,9 @@ func describeCaps(c media.Caps) string { } // make produces a preview, sharing the work when the same file is asked for twice at once. -func (t *thumbSvc) make(ctx context.Context, root bool, key, path string) (media.Result, error) { +// in is a path, or a URL the hub serves another machine's file at (kind then says what it +// is); device is the one to decode on ("" any, "cpu" none). +func (t *thumbSvc) make(ctx context.Context, root bool, key, in, kind, device string) (media.Result, error) { t.mu.Lock() if c, ok := t.calls[key]; ok { t.mu.Unlock() @@ -153,19 +162,29 @@ func (t *thumbSvc) make(ctx context.Context, root bool, key, path string) (media c.err = err return c.res, c.err } - o := media.Opts{Size: thumbSize, Timeout: thumbTimeout} + o := media.Opts{Size: thumbSize, Timeout: thumbTimeout, Kind: kind} + caps = caps.Only(device) if root { - c.res, c.err = media.MakeRoot(wctx, t.self, caps, path, o) + c.res, c.err = media.MakeRoot(wctx, t.self, caps, in, o) } else { - c.res, c.err = media.Make(wctx, caps, path, o) + c.res, c.err = media.Make(wctx, caps, in, o) } return c.res, c.err } // ---- handlers ---- -// filesThumb: GET /api/files/thumb?path=…&v=-[&sudo=1]. v is the client's idea of -// the file, from its listing; it makes a cache hit free of a stat, and a changed file a new URL. +// thumbDeviceHeader carries the device a hub wants a video preview made on: the Transcode +// device setting, when it is on the machine the file is on. +const thumbDeviceHeader = "X-Hamlab-Device" + +// filesThumb: GET /api/files/thumb?path=…&v=-[&sudo=1][&cached=1]. v is the +// client's idea of the file, from its listing; it makes a cache hit free of a stat, and a +// changed file a new URL. cached=1 answers from the cache or 404, and makes nothing: the hub +// asks that before having the preview made on another machine. +// +// A video preview is made on the Transcode device: on the hub, by the setting (another +// machine's device reads the file through the hub); on a node, by the hub's header. func (s *Server) filesThumb(w http.ResponseWriter, r *http.Request) { fsys, ok := s.fsFor(w, r) if !ok { @@ -203,7 +222,28 @@ func (s *Server) filesThumb(w http.ResponseWriter, r *http.Request) { if s.serveCachedThumb(w, key) { return } - res, err := s.thumbs.make(r.Context(), root, key, p) + if r.URL.Query().Get("cached") == "1" { + jsonErr(w, 404, "not made yet") + return + } + device := r.Header.Get(thumbDeviceHeader) + if c := s.transcodeChoice(); c != nil && device == "" && media.Kind(p) == "video" { + if c.Node != "" { + // The hub's own file, previewed on another machine. + if res, ok := s.thumbElsewhere(r.Context(), c, "", p, root, key); ok { + if res.Failure != "" { + _ = s.thumbs.cache.Put(key, media.Entry{Path: p, Failure: res.Failure}) + jsonErr(w, 422, res.Failure) + return + } + s.keepThumb(w, key, p, res) + return + } + } else { + device = c.Device + } + } + res, err := s.thumbs.make(r.Context(), root, key, p, "", device) if err != nil { var de *media.DecodeError switch { @@ -223,12 +263,204 @@ func (s *Server) filesThumb(w http.ResponseWriter, r *http.Request) { } return } + s.keepThumb(w, key, p, res) +} + +// keepThumb puts a preview in the cache and answers with it. +func (s *Server) keepThumb(w http.ResponseWriter, key, p string, res media.Result) { if err := s.thumbs.cache.Put(key, media.Entry{Path: p, Engine: res.Engine, Codec: res.Codec, Width: res.Width, Height: res.Height, JPEG: res.JPEG}); err != nil { log.Printf("thumbs: cache: %v", err) } writeThumb(w, key, res.JPEG, res.Engine) } +// ---- previews made on another machine (the Transcode device) ---- + +// thumbFrom: POST /api/files/thumb-from {url, kind, key, device} — a preview of a file on +// another machine, read through the hub at url, made here on device. Answers the JPEG (the +// engine in X-Hamlab-Thumb), or 422 with why the file can't be previewed. +func (s *Server) thumbFrom(w http.ResponseWriter, r *http.Request) { + var body struct{ URL, Kind, Key, Device string } + if err := json.NewDecoder(io.LimitReader(r.Body, 8192)).Decode(&body); err != nil || !strings.HasPrefix(body.URL, "http") || body.Key == "" { + jsonErr(w, 400, "expected {url, kind, key, device}") + return + } + res, err := s.thumbs.make(r.Context(), false, "from:"+body.Key, body.URL, body.Kind, body.Device) + if err != nil { + var de *media.DecodeError + if errors.As(err, &de) { + jsonErr(w, 422, de.Msg) + } else { + jsonErr(w, 503, err.Error()) + } + return + } + w.Header().Set("X-Hamlab-Thumb-Codec", res.Codec) + w.Header().Set("X-Hamlab-Thumb-Size", fmt.Sprintf("%dx%d", res.Width, res.Height)) + writeThumb(w, body.Key, res.JPEG, res.Engine) +} + +// thumbPut: PUT /api/files/thumb?path=…&v=-[&sudo=1] — a preview another machine +// made of this file, kept here as if made here. The body is the JPEG, or nothing with +// X-Hamlab-Thumb-Failure saying why the file can't be previewed. Refused if the file has +// changed since v. +func (s *Server) thumbPut(w http.ResponseWriter, r *http.Request) { + fsys, ok := s.fsFor(w, r) + if !ok { + return + } + p, ok := cleanQuery(w, r, "path") + if !ok { + return + } + mt, sz, ok := parseV(r.URL.Query().Get("v")) + if !ok { + jsonErr(w, 400, "v=- is required") + return + } + in, err := fsys.Stat(r.Context(), p) + if err != nil { + fsErr(w, err) + return + } + if in.Size != sz || in.Mtime != mt { + jsonErr(w, 409, "the file has changed") + return + } + key := media.Key(p, sz, mt) + e := media.Entry{Path: p, Engine: r.Header.Get("X-Hamlab-Thumb"), Codec: r.Header.Get("X-Hamlab-Thumb-Codec"), Failure: r.Header.Get("X-Hamlab-Thumb-Failure")} + fmt.Sscanf(r.Header.Get("X-Hamlab-Thumb-Size"), "%dx%d", &e.Width, &e.Height) + if e.Failure == "" { + if e.JPEG, err = io.ReadAll(io.LimitReader(r.Body, 4<<20)); err != nil || len(e.JPEG) == 0 { + jsonErr(w, 400, "no picture") + return + } + } + if err := s.thumbs.cache.Put(key, e); err != nil { + jsonErr(w, 500, err.Error()) + return + } + w.WriteHeader(204) +} + +// thumbElsewhere has a video preview made on the Transcode device's machine c.Node, of a +// file on machine node ("" the hub), and answers it. ok is false when that machine can't +// (unreachable, an older Ham Lab): the caller makes it where the file is instead. A file +// that can't be previewed is a result too, with the failure in it. +func (s *Server) thumbElsewhere(ctx context.Context, c *store.Transcode, node, p string, root bool, key string) (media.Result, bool) { + src, err := s.srcURL(c.Node, node, p, root) + if err != nil { + return media.Result{}, false + } + body, _ := json.Marshal(map[string]string{"url": src, "kind": "video", "key": key, "device": c.Device}) + ctx, cancel := context.WithTimeout(ctx, thumbTimeout+10*time.Second) + defer cancel() + var code int + var b []byte + var hdr http.Header + if c.Node == "" { + rec := httptest.NewRecorder() + req := httptest.NewRequest("POST", "/api/files/thumb-from", bytes.NewReader(body)).WithContext(ctx) + s.thumbFrom(rec, req) + code, b, hdr = rec.Code, rec.Body.Bytes(), rec.Header() + } else { + code, b, hdr, err = s.nodes.DoHeaders(ctx, c.Node, "POST", "/api/files/thumb-from", body, http.Header{"Content-Type": []string{"application/json"}}) + if err != nil { + log.Printf("thumbs: %s on %s: %v; made where it is", p, c.Node, err) + return media.Result{}, false + } + } + switch code { + case 200: + res := media.Result{JPEG: b, Engine: c.Node + " " + hdr.Get("X-Hamlab-Thumb"), Codec: hdr.Get("X-Hamlab-Thumb-Codec")} + fmt.Sscanf(hdr.Get("X-Hamlab-Thumb-Size"), "%dx%d", &res.Width, &res.Height) + if c.Node == "" { + res.Engine = s.hostName() + " " + hdr.Get("X-Hamlab-Thumb") + } + return res, true + case 422: + var e struct{ Error string } + _ = json.Unmarshal(b, &e) + return media.Result{Failure: e.Error}, true + } + log.Printf("thumbs: %s on %s answered %d; made where it is", p, nodeLabel(c.Node), code) + return media.Result{}, false +} + +// hubThumb: GET /api/nodes/{name}/files/thumb — a node's preview, through the hub. Without a +// Transcode device setting, or for an image or a sound file's cover, it goes to the node as +// any request does. A video is made on the setting's device: the node's own (told by header), +// or another machine's, which reads the file through the hub; the node keeps the result in +// its own cache either way. +func (s *Server) hubThumb(w http.ResponseWriter, r *http.Request) { + node := r.PathValue("name") + proxy, ok := s.nodes.Proxy(node) + if !ok { + jsonErr(w, 404, "no such machine") + return + } + c := s.transcodeChoice() + q := r.URL.Query() + p := q.Get("path") + if c == nil || media.Kind(p) != "video" { + proxy.ServeHTTP(w, r) + return + } + if c.Node == node { + r.Header.Set(thumbDeviceHeader, c.Device) + proxy.ServeHTTP(w, r) + return + } + mt, sz, okV := parseV(q.Get("v")) + if !okV { + proxy.ServeHTTP(w, r) + return + } + // The node's own cache first. + cq := url.Values{"path": {p}, "v": {q.Get("v")}, "cached": {"1"}} + if q.Get("sudo") == "1" { + cq.Set("sudo", "1") + } + hdr := http.Header{} + if inm := r.Header.Get("If-None-Match"); inm != "" { + hdr.Set("If-None-Match", inm) + } + code, b, rh, err := s.nodes.DoHeaders(r.Context(), node, "GET", "/api/files/thumb?"+cq.Encode(), nil, hdr) + if err == nil && code != 404 { + for _, k := range []string{"Content-Type", "Cache-Control", "ETag", "X-Hamlab-Thumb"} { + if v := rh.Get(k); v != "" { + w.Header().Set(k, v) + } + } + w.WriteHeader(code) + _, _ = w.Write(b) + return + } + key := media.Key(p, sz, mt) + res, made := s.thumbElsewhere(r.Context(), c, node, p, q.Get("sudo") == "1", key) + if !made { + proxy.ServeHTTP(w, r) // the node makes it itself + return + } + // Kept where the file is, so it is made once. + ph := http.Header{"X-Hamlab-Thumb": {res.Engine}, "X-Hamlab-Thumb-Codec": {res.Codec}, "X-Hamlab-Thumb-Size": {fmt.Sprintf("%dx%d", res.Width, res.Height)}} + if res.Failure != "" { + ph.Set("X-Hamlab-Thumb-Failure", res.Failure) + } + pq := url.Values{"path": {p}, "v": {q.Get("v")}} + if q.Get("sudo") == "1" { + pq.Set("sudo", "1") + } + if code, _, _, err := s.nodes.DoHeaders(r.Context(), node, "PUT", "/api/files/thumb?"+pq.Encode(), res.JPEG, ph); err != nil || code != 204 { + log.Printf("thumbs: keeping %s's preview of %s: %d %v", node, p, code, err) + } + if res.Failure != "" { + jsonErr(w, 422, res.Failure) + return + } + writeThumb(w, key, res.JPEG, res.Engine) +} + // serveCachedThumb answers from the cache when it can. A cache that cannot be read is a // miss: the preview is made again rather than refused. func (s *Server) serveCachedThumb(w http.ResponseWriter, key string) bool { diff --git a/internal/media/encode_test.go b/internal/media/encode_test.go index 8ce3761..3693c2d 100644 --- a/internal/media/encode_test.go +++ b/internal/media/encode_test.go @@ -155,3 +155,17 @@ func TestPieceArgs(t *testing.T) { t.Error("no sound means -an") } } + +func TestCapsOnly(t *testing.T) { + c := gridCaps() + if got := c.Only("/dev/dri/renderD129"); len(got.GPUs) != 1 || got.GPUs[0].Name != "Intel Arc A380" || len(c.GPUs) != 2 { + t.Fatalf("one card: %+v", got.GPUs) + } + if len(c.Only(CPU).GPUs) != 0 || len(c.Only("/dev/dri/renderD200").GPUs) != 0 || len(c.Only("").GPUs) != 2 { + t.Fatal("cpu, a device not here, any") + } + // previews then go to that card, or the CPU + if g := c.Only("/dev/dri/renderD129").pick("hevc"); g == nil || g.Device != "/dev/dri/renderD129" { + t.Fatalf("pick: %+v", g) + } +} diff --git a/internal/media/thumbs.go b/internal/media/thumbs.go index e9b9da8..72949db 100644 --- a/internal/media/thumbs.go +++ b/internal/media/thumbs.go @@ -62,6 +62,24 @@ type Caps struct { type Opts struct { Size int // the long edge in pixels; 256 when zero Timeout time.Duration // for the whole job, ffprobe and every ffmpeg attempt; 30 s when zero + Kind string // image | video | audio; from the name when empty (a URL has none) +} + +// Only is the caps with one device left to decode on: a render node, CPU for none, or "" +// for all of them — the Transcode device setting, which previews follow too. A device this +// machine doesn't have leaves the CPU. +func (c Caps) Only(device string) Caps { + if device == "" { + return c + } + out := c + out.GPUs = nil + for _, g := range c.GPUs { + if g.Device == device { + out.GPUs = append(out.GPUs, g) + } + } + return out } func (o Opts) defaults() Opts { @@ -81,6 +99,8 @@ type Result struct { Codec string `json:"codec"` Width int `json:"width"` Height int `json:"height"` + // Failure is why the file can't be previewed, when another machine tried (no JPEG then). + Failure string `json:"-"` } // ErrUnsupported: not an image or a video by its name. @@ -356,7 +376,10 @@ func (a attempt) engine() string { // Make produces the preview of one file. func Make(ctx context.Context, c Caps, path string, o Opts) (Result, error) { - kind := Kind(path) + kind := o.Kind + if kind == "" { + kind = Kind(path) + } if kind == "" { return Result{}, ErrUnsupported } diff --git a/internal/nodes/nodes.go b/internal/nodes/nodes.go index c054b86..76d479a 100644 --- a/internal/nodes/nodes.go +++ b/internal/nodes/nodes.go @@ -383,13 +383,19 @@ func (m *Manager) PushAll(ctx context.Context, binPath, ourArch string, force bo // WebSocket origin check (meant for browsers) doesn't reject the hub. // Do makes one request to a managed machine as this hub and returns the whole response. func (m *Manager) Do(ctx context.Context, name, method, path string, body []byte, hdr http.Header) (int, []byte, error) { + code, b, _, err := m.DoHeaders(ctx, name, method, path, body, hdr) + return code, b, err +} + +// DoHeaders is Do that also answers the response's headers. +func (m *Manager) DoHeaders(ctx context.Context, name, method, path string, body []byte, hdr http.Header) (int, []byte, http.Header, error) { n, ok := m.node(name) if !ok { - return 0, nil, errors.New("no such machine") + return 0, nil, nil, errors.New("no such machine") } req, err := http.NewRequestWithContext(ctx, method, strings.TrimRight(n.URL, "/")+path, bytes.NewReader(body)) if err != nil { - return 0, nil, err + return 0, nil, nil, err } for k, v := range hdr { req.Header[k] = v @@ -397,11 +403,11 @@ func (m *Manager) Do(ctx context.Context, name, method, path string, body []byte req.Header.Set("Authorization", "Bearer "+n.Token) resp, err := m.client(n).Do(req) if err != nil { - return 0, nil, errors.New(shortErr(err)) + return 0, nil, nil, errors.New(shortErr(err)) } defer resp.Body.Close() b, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20)) - return resp.StatusCode, b, err + return resp.StatusCode, b, resp.Header, err } // URL is a managed machine's address, "" for one not managed here.