From 450a9d33273282446b034a4b20f609570ac3f7ab Mon Sep 17 00:00:00 2001 From: Mitchell-Johnson Date: Tue, 20 Oct 2026 13:08:52 -0700 Subject: [PATCH] files: seek anywhere in a re-encode; it restarts from there A re-encode's playlist is now the whole film on a fixed 4 s grid. A segment that isn't on disk and isn't coming soon moves the encode there; runs stop short of segments already made, and gaps are filled after. Every run is cut the same way (-ss, -output_ts_offset start+10, frag_discont), so segments from different runs join. Repackaging keeps its growing playlist. Co-Authored-By: Claude Opus 5.5 --- PLAN.md | 35 ++- internal/api/play.go | 30 ++- internal/media/encode.go | 63 ++++++ internal/media/encode_test.go | 29 +++ internal/media/hls.go | 387 +++++++++++++++++++++++++++++++--- internal/media/hls_test.go | 143 ++++++++++++- web/src/Player.tsx | 12 +- web/src/api.ts | 6 +- 8 files changed, 660 insertions(+), 45 deletions(-) diff --git a/PLAN.md b/PLAN.md index 8aab125..a3acba4 100644 --- a/PLAN.md +++ b/PLAN.md @@ -772,9 +772,38 @@ something long in one, switch away and back, refresh — it's all still there. 11.7×, AV1 720p 13×, all on the Arc. `core` and `hamtv` re-encode on the CPU until *Allow GPU access* is pressed (libx264 at 9.4× for a 720p→480p clip on `core`); the Mac and `gate` need ffmpeg first, which the previews note offers to install. - - *Not done, on purpose:* no restart-from-a-seek beyond what is encoded (the encode is - faster than watching; a seek far ahead waits), no bit-rate ladder or adaptive - switching, no subtitles, no keeping a finished re-encode for next time. + - *Not done, on purpose:* no bit-rate ladder or adaptive switching, no subtitles, no + keeping a finished re-encode for next time. (Restart-from-a-seek was left out here at + first; it came 2026-10-20, the next entry.) +- **Seek anywhere in a re-encode** — **2026-10-20**. Asked for as "there's no way to seek in + the video and have the service just start transcoding from that point onward". + - **The playlist is the whole film, up front:** a VOD playlist on a fixed 4 s grid + (`segCount`, the last 0.3 s left out so sound running past the picture never lists a + segment no frame starts), so the seek bar is the film's from the first frame. + - **A segment not on disk** is waited for when the run going now will write it within 3 + segments; otherwise it is a seek: that run is ended and one starts at the segment asked + for. A run stops half a frame short of the next segment already on disk, never over it. + When a run ends by itself the next gap is filled, wrapping round, until every segment is + there. The newest request moves the run; an older one still waiting never moves it back. + A run we end closes its half-done segment short on the way out, so whatever it finished + after being asked to stop is deleted. + - **Runs are interchangeable** because every one, the first included, is made the same way + (`segArgs`): accurate `-ss` to the segment, timestamps put back with + `-output_ts_offset start+10` (the +10 keeps every run clear of the muxer's shift for the + B-frame delay, which a run from the top would otherwise get and a seek wouldn't), kept + absolute in each fragment (`movflags=+frag_discont`), `-start_number`. Tested on MKV + and on MPEG-TS with a 2.8 s start time (where `-copyts` was a second out): the init + segments come out byte-identical, and a seeked run's segments join a from-the-top run's + within a frame. Only the first run's init is served; a later run writes `init-run.mp4` + and the log notes if it ever differs. + - Repackaged ("copy") streams keep the growing event playlist: the source's keyframes + don't sit on a grid, and repackaging is near-instant anyway. + - A segment waits at most 45 s, under the hub proxy's 60 s for a node to answer; hls.js + is told a segment's first byte may take 50 s. The bar reads "7.3× from 12:00 · 43% + ready — seek anywhere". + - Tested (`TestSeekRestartsTheEncode`): a 64 s 1080p clip, a request for segment 12 right + after the start runs from 48 s, the gap is filled, all 16 segments joined behind the one + init decode with no hole. - **Files edge sweep** — **2026-10-11** (v0.1.0-214), asked for as "a full bug sweep of the Files section, every example and edge case". Every handler and both UI files read through with odd inputs in mind, then a battery of ~60 requests against the live hub diff --git a/internal/api/play.go b/internal/api/play.go index 0dbaf2e..c5730bf 100644 --- a/internal/api/play.go +++ b/internal/api/play.go @@ -146,24 +146,48 @@ func (s *Server) playStop(w http.ResponseWriter, r *http.Request) { } // playFile: GET /api/files/play/{id}/{file} — the playlist, the init segment, a segment. +// A seekable stream's playlist is written whole, up front; a segment of it that isn't on disk +// yet is waited for, or re-encoded from there if the encode isn't heading that way (a seek). func (s *Server) playFile(w http.ResponseWriter, r *http.Request) { st, ok := s.plays.Find(r.PathValue("id")) if !ok { jsonErr(w, 404, "no such stream") return } - p, ok := st.File(r.PathValue("file")) + name := r.PathValue("file") + p, ok := st.File(name) if !ok { jsonErr(w, 404, "not a stream file") return } + w.Header().Set("Cache-Control", "no-store") + w.Header().Set("X-Content-Type-Options", "nosniff") + if st.Seekable { + if name == "index.m3u8" { + w.Header().Set("Content-Type", "application/vnd.apple.mpegurl") + _, _ = w.Write(st.Playlist()) + return + } + if i, isSeg := media.SegmentIndex(name); isSeg { + sp, err := st.Segment(r.Context(), i) + switch { + case errors.Is(err, media.ErrNotInFilm): + jsonErr(w, 404, err.Error()) + return + case r.Context().Err() != nil: + return // the player moved on + case err != nil: + jsonErr(w, 503, err.Error()) + return + } + p = sp + } + } switch filepath.Ext(p) { case ".m3u8": w.Header().Set("Content-Type", "application/vnd.apple.mpegurl") default: w.Header().Set("Content-Type", "video/mp4") } - w.Header().Set("Cache-Control", "no-store") - w.Header().Set("X-Content-Type-Options", "nosniff") http.ServeFile(w, r, p) } diff --git a/internal/media/encode.go b/internal/media/encode.go index 7637840..5883abd 100644 --- a/internal/media/encode.go +++ b/internal/media/encode.go @@ -212,6 +212,69 @@ func hlsArgs(dir, in string, src Info, r Recipe, copyCodec bool) []string { const hlsSeconds = 4 +// tsBase is added to every run's timestamps. A run from the top would otherwise start just +// below zero (the B-frame delay) and the muxer would shift it to make room, while a run from a +// seek point would not: with the base, no run is ever shifted, and a segment means the same +// stretch of the film whichever run made it. +const tsBase = 10 + +// segArgs is one run of a seekable re-encode: segments from..to-1 of the fixed 4 s grid. +// Every run is made the same way, the first included, so their segments are interchangeable: +// an accurate seek to from×4 s, timestamps put back where that is in the film (+ tsBase), +// kept as they are in each fragment (frag_discont), numbered from `from`, and an end just +// short of segment `to`, which is already on disk, so the run never writes over it. Segments +// land in dir as seg%05d.m4s; init is the init segment's name (only the first run's is used). +func segArgs(dir, in string, src Info, r Recipe, from, to, n int, init string) []string { + fps := 30.0 + if src.Video != nil && src.Video.FPS > 0 { + fps = src.Video.FPS + } + start := from * hlsSeconds + a := []string{"-nostdin", "-hide_banner", "-loglevel", "error", "-y"} + a = append(a, r.Pre...) + a = append(a, "-ss", strconv.Itoa(start), "-i", in, "-map", "0:v:0", "-map", "0:a:0?", "-sn", "-dn") + if r.VF != "" { + a = append(a, "-vf", r.VF) + } + a = append(a, r.Video...) + switch { + case src.Audio == nil: + a = append(a, "-an") + case src.Audio.Codec == "aac": + a = append(a, "-c:a", "copy") + default: + a = append(a, "-c:a", "aac", "-b:a", "160k", "-ac", "2") + } + gop := int(math.Round(fps * hlsSeconds)) + if gop < 24 { + gop = 24 + } + if gop > 300 { + gop = 300 + } + a = append(a, "-g", strconv.Itoa(gop), "-force_key_frames", fmt.Sprintf("expr:gte(t,n_forced*%d)", hlsSeconds)) + if to < n { + a = append(a, "-t", strconv.FormatFloat(float64((to-from)*hlsSeconds)-0.5/fps, 'f', 3, 64)) + } + a = append(a, "-output_ts_offset", strconv.Itoa(start+tsBase), + "-f", "hls", "-hls_time", strconv.Itoa(hlsSeconds), "-hls_playlist_type", "event", "-hls_segment_type", "fmp4", + "-hls_segment_options", "movflags=+frag_discont", "-start_number", strconv.Itoa(from), + "-hls_fmp4_init_filename", init, "-hls_segment_filename", dir+"/seg%05d.m4s", "-hls_flags", "independent_segments+temp_file", + "-progress", dir+"/progress", "-stats_period", "1", dir+"/run.m3u8") + return a +} + +// segCount is how many 4 s segments a seekable re-encode of a file this long lists. The last +// 0.3 s are left out of the count, so a container whose sound runs on past the picture never +// lists a segment no frame starts. +func segCount(duration float64) int { + d := duration - 0.3 + if d < 0 { + d = 0 + } + return int(d/hlsSeconds) + 1 +} + func has(list []string, x string) bool { for _, v := range list { if v == x { diff --git a/internal/media/encode_test.go b/internal/media/encode_test.go index e53423c..408ecd4 100644 --- a/internal/media/encode_test.go +++ b/internal/media/encode_test.go @@ -110,6 +110,35 @@ func TestHLSArgs(t *testing.T) { } } +func TestSegArgs(t *testing.T) { + in := src("h264", 1920, 1080, 25) + r := recipes(gridCaps(), "h264", in, 1920, 1080, 10000)[0] + first := strings.Join(segArgs("/d", "/v.mkv", in, r, 0, 900, 900, "init.mp4"), " ") + for _, want := range []string{"-ss 0 -i /v.mkv", "-output_ts_offset 10", "-start_number 0", "movflags=+frag_discont", "-hls_fmp4_init_filename init.mp4", "-hls_segment_filename /d/seg%05d.m4s", "/d/run.m3u8", "-force_key_frames expr:gte(t,n_forced*4)"} { + if !strings.Contains(first, want) { + t.Errorf("first run: missing %q in\n%s", want, first) + } + } + if strings.Contains(first, " -t ") { + t.Error("a run to the end has no -t") + } + // a seek: segments 30..44, stopping half a frame short of 45, which is on disk + seek := strings.Join(segArgs("/d", "/v.mkv", in, r, 30, 45, 900, "init-run.mp4"), " ") + for _, want := range []string{"-ss 120 -i /v.mkv", "-t 59.980", "-output_ts_offset 130", "-start_number 30", "-hls_fmp4_init_filename init-run.mp4"} { + if !strings.Contains(seek, want) { + t.Errorf("seek run: missing %q in\n%s", want, seek) + } + } + if strings.Index(seek, "-ss 120") > strings.Index(seek, "-i /v.mkv") || strings.Index(seek, "-hwaccel") > strings.Index(seek, "-i /v.mkv") { + t.Error("-ss and the decoder's options go before -i") + } + for d, n := range map[float64]int{9: 3, 8.2: 2, 8.31: 3, 64: 16, 0.1: 1, 3600: 900} { + if got := segCount(d); got != n { + t.Errorf("segCount(%v) = %d, want %d", d, got, n) + } + } +} + func TestParseProgress(t *testing.T) { b := []byte("frame=10\nout_time_us=1000000\nspeed=2.5x\nprogress=continue\nframe=20\nout_time_us=2500000\nspeed=3.1x\nprogress=continue\n") s, sp := parseProgress(b) diff --git a/internal/media/hls.go b/internal/media/hls.go index 53d7eb4..b88651c 100644 --- a/internal/media/hls.go +++ b/internal/media/hls.go @@ -3,14 +3,26 @@ package media // Playing what the browser cannot play as it is: an HLS stream, re-encoded on demand by // ffmpeg on the machine's own GPU where it has one it can use, on its CPU otherwise. A // stream is one file at one codec and size, in a folder of its own under -// /transcode/: ffmpeg writes an event playlist and fragmented-MP4 segments there -// as fast as it can, and the player reads them as they land, so the range it can seek in -// grows until the whole file is through — on a GPU, several times faster than watching. -// Nothing is kept: a stream nobody has touched for a while is ended and its folder -// removed, and the folder is emptied when the daemon starts. +// /transcode/. +// +// A re-encode is seekable anywhere from the start. Its playlist is written up front, for +// the whole film: a fixed grid of 4 s segments, so the player's timeline is the film's. +// ffmpeg runs ahead from the top as fast as it can; a request for a segment it hasn't +// reached and won't reach soon — a seek ahead, or back into a stretch never made — ends that +// run and starts another at the segment asked for, which stops on its own where segments +// already exist. When a run finishes, the next gap is filled, until the whole file is on +// disk. Every run is made the same way (segArgs), so segments from different runs join. +// +// "Repackaged" (copy) can't cut on a fixed grid — its keyframes are the source's — so it is +// an event playlist that grows as ffmpeg writes it, seekable within what's done. Repackaging +// runs at disk speed, so that is all of it within seconds. +// +// Nothing is kept: a stream nobody has touched for a while is ended and its folder removed, +// and the folder is emptied when the daemon starts. import ( "bufio" + "bytes" "context" "crypto/rand" "encoding/hex" @@ -28,26 +40,46 @@ import ( // Stream is one re-encode, in progress or finished. type Stream struct { - ID string `json:"id"` - Path string `json:"path"` - Codec string `json:"codec"` - Quality string `json:"quality"` - Engine string `json:"engine"` - Width int `json:"width"` - Height int `json:"height"` - Kbps int `json:"kbps"` - Source Info `json:"source"` - Started time.Time `json:"started"` - Root bool `json:"root"` + ID string `json:"id"` + Path string `json:"path"` + Codec string `json:"codec"` + Quality string `json:"quality"` + Engine string `json:"engine"` + Width int `json:"width"` + Height int `json:"height"` + Kbps int `json:"kbps"` + Source Info `json:"source"` + Started time.Time `json:"started"` + Root bool `json:"root"` + Seekable bool `json:"seekable"` // seeking anywhere restarts the re-encode there key string dir string mu sync.Mutex used time.Time - proc *ffproc + proc *ffproc // the run going now (or the only one, for copy) finished bool failure string gone bool // forgotten by the Streamer + + // seekable streams only + caps Caps + self string + recipe Recipe + n int // segments in the playlist + end int // segments past this are not in the file after all (n until a run finds out) + run *segRun + runMu sync.Mutex // one run change at a time; held across ending one run and starting the next + asked time.Time // when the request that last moved the run arrived: an older one never moves it back + initOK bool // a later run's init segment was checked against the first's + initBad bool +} + +// segRun is one ffmpeg over segments from..to-1. +type segRun struct { + from, to int + proc *ffproc + stopped bool // ended by us: what it wrote after that is not to be trusted } // StreamStatus is what the player asks for while the file is still going through. @@ -55,9 +87,10 @@ type StreamStatus struct { Running bool `json:"running"` Done bool `json:"done"` Error string `json:"error,omitempty"` - Progress float64 `json:"progress"` // seconds of output written + Progress float64 `json:"progress"` // seconds of output written (copy), or on disk (seekable) Speed float64 `json:"speed"` // × realtime Segments int `json:"segments"` + From float64 `json:"from"` // seekable: where the run going now started, seconds } // Streamer keeps one machine's streams. @@ -72,7 +105,15 @@ type Streamer struct { const streamIdle = 10 * time.Minute const firstSegmentWait = 40 * time.Second +// segmentWait is how long a request for a segment waits for a run to write it. +const segmentWait = 45 * time.Second // under the hub proxy's 60 s wait for a node to answer + +// lookahead: a segment this many past what the run has written is waited for, not +// restarted at. A few seconds of film, which any engine here writes faster than that. +const lookahead = 3 + var streamFile = regexp.MustCompile(`^(index\.m3u8|init\.mp4|seg\d{5}\.m4s)$`) +var segName = regexp.MustCompile(`^seg(\d{5})\.m4s$`) // NewStreamer empties the folder — a daemon restart orphans every stream — and starts the // sweeper. @@ -128,8 +169,15 @@ func (m *Streamer) Open(ctx context.Context, c Caps, root bool, path, codec, qua m.mu.Unlock() return nil, fmt.Errorf("%d videos are already being re-encoded on this machine; one must finish or be closed first", running) } - s := &Stream{ID: newID(), Path: path, Codec: codec, Quality: quality, Source: src, Started: time.Now(), Root: root, key: key, used: time.Now()} + s := &Stream{ID: newID(), Path: path, Codec: codec, Quality: quality, Source: src, Started: time.Now(), Root: root, key: key, used: time.Now(), + caps: c, self: m.self} s.dir = filepath.Join(m.dir, s.ID) + // A fixed grid needs the film's length; without one it can only grow as it goes. + s.Seekable = codec != "copy" && src.Duration > 0 + if s.Seekable { + s.n = segCount(src.Duration) + s.end = s.n + } m.streams[s.ID] = s m.mu.Unlock() @@ -149,18 +197,30 @@ func (m *Streamer) Open(ctx context.Context, c Caps, root bool, path, codec, qua return nil, err } _ = clearDir(s.dir) - args := hlsArgs(s.dir, path, src, r, codec == "copy") + var args []string + if s.Seekable { + args = segArgs(s.dir, path, src, r, 0, s.n, s.n, "init.mp4") + } else { + args = hlsArgs(s.dir, path, src, r, codec == "copy") + } p, err := startFF(c, m.self, root, args) if err != nil { last = r.Engine + ": " + err.Error() continue } s.mu.Lock() - s.proc, s.Engine = p, r.Engine + s.proc, s.Engine, s.recipe = p, r.Engine, r + if s.Seekable { + s.run = &segRun{from: 0, to: s.n, proc: p} + } s.mu.Unlock() msg, ok := s.waitFirstSegment(ctx) if ok { - go s.watch() + if s.Seekable { + go s.watchRun(s.run) + } else { + go s.watch() + } log.Printf("media: stream %s: %s → %s %s %dx%d %dk on %s", s.ID, filepath.Base(path), codec, quality, s.Width, s.Height, s.Kbps, r.Engine) return s, nil } @@ -176,12 +236,12 @@ func (m *Streamer) Open(ctx context.Context, c Caps, root bool, path, codec, qua return nil, &DecodeError{Msg: last} } -// waitFirstSegment watches for the playlist's first segment, or ffmpeg giving up. +// waitFirstSegment watches for the first segment, or ffmpeg giving up. func (s *Stream) waitFirstSegment(ctx context.Context) (string, bool) { deadline := time.Now().Add(firstSegmentWait) playlist := filepath.Join(s.dir, "index.m3u8") for { - if hasSegments(playlist) { + if s.Seekable && exists(s.segPath(0)) || !s.Seekable && hasSegments(playlist) { return "", true } if done, err := s.proc.exited(); done { @@ -205,7 +265,7 @@ func (s *Stream) waitFirstSegment(ctx context.Context) (string, bool) { } } -// watch records how the encode ended. +// watch records how the encode ended (copy streams: one run, start to end). func (s *Stream) watch() { <-s.proc.done s.mu.Lock() @@ -222,6 +282,250 @@ func (s *Stream) watch() { } } +// ---- seekable streams ---- + +func (s *Stream) segPath(i int) string { return filepath.Join(s.dir, fmt.Sprintf("seg%05d.m4s", i)) } + +// written is the last segment of a run on disk, counting on from where it started. +func (s *Stream) written(r *segRun) int { + i := r.from + for i < r.to && exists(s.segPath(i)) { + i++ + } + return i - 1 +} + +// watchRun follows one run to its end. A run that ends by itself — the end of the film, or +// the segments it was to stop short of — is followed by the next gap. One that failed fails +// the stream; one we ended is the business of whoever ended it. +func (s *Stream) watchRun(r *segRun) { + <-r.proc.done + s.mu.Lock() + if s.gone || r.stopped || s.run != r { + s.mu.Unlock() + return + } + if r.proc.err != nil { + s.failure = r.proc.lastWords() + if s.failure == "" { + s.failure = r.proc.err.Error() + } + s.finished = true + s.mu.Unlock() + log.Printf("media: stream %s: run from segment %d failed: %s", s.ID, r.from, s.failure) + return + } + if w := s.written(r); w+1 < r.to && r.to == s.n { + // ran to the end of the file before the end of the grid: the picture is shorter than + // the container says, and those last segments will never exist + s.end = w + 1 + log.Printf("media: stream %s: the film ends at segment %d of %d", s.ID, s.end, s.n) + } + s.checkInit(r) + s.mu.Unlock() + s.fillNext(r.to) +} + +// checkInit compares a later run's init segment with the first run's, once. They come out +// identical (same encoder, same settings); if one day they don't, the log says so. +func (s *Stream) checkInit(r *segRun) { + if r.from == 0 || s.initOK { + return + } + a, err1 := os.ReadFile(filepath.Join(s.dir, "init.mp4")) + b, err2 := os.ReadFile(filepath.Join(s.dir, "init-run.mp4")) + if err1 != nil || err2 != nil { + return + } + s.initOK = true + if !bytes.Equal(a, b) && !s.initBad { + s.initBad = true + log.Printf("media: stream %s: a restarted run's init segment differs from the first (%s); its segments may not join", s.ID, s.Engine) + } +} + +// missingFrom is the first segment not on disk at or after i (wrapping round to the start), +// or -1 when the whole film is there. +func (s *Stream) missingFrom(i int) int { + end := s.end + for k := 0; k < end; k++ { + j := (i + k) % end + if !exists(s.segPath(j)) { + return j + } + } + return -1 +} + +// fillNext starts a run at the next gap after `after`; when there is none, the stream is done. +func (s *Stream) fillNext(after int) { + s.runMu.Lock() + defer s.runMu.Unlock() + s.mu.Lock() + if s.gone || s.failure != "" || (s.run != nil && !s.run.done()) { + s.mu.Unlock() + return + } + end := s.end + s.mu.Unlock() + if after >= end { + after = 0 + } + next := s.missingFrom(after) + if next < 0 { + s.mu.Lock() + s.finished, s.proc = true, nil + s.mu.Unlock() + log.Printf("media: stream %s: all %d segments on disk", s.ID, end) + return + } + s.startRun(next) +} + +func (r *segRun) done() bool { d, _ := r.proc.exited(); return d } + +// startRun ends the run going now and starts one at segment from, stopping short of the next +// segment already on disk. The run it ends is asked to stop and waited for, and whatever it +// finished writing after being asked is removed: ffmpeg closes the segment it was in the +// middle of on its way out, short. Called with runMu held, mu not. +func (s *Stream) startRun(from int) { + s.mu.Lock() + old := s.run + end := s.end + s.mu.Unlock() + if old != nil && !old.done() { + had := map[int]bool{} + for i := old.from; i < old.to; i++ { + had[i] = exists(s.segPath(i)) + } + s.mu.Lock() + old.stopped = true + s.mu.Unlock() + old.proc.end() + for i := old.from; i < old.to; i++ { + if !had[i] && exists(s.segPath(i)) { + _ = os.Remove(s.segPath(i)) + } + } + } + to := from + 1 + for to < end && !exists(s.segPath(to)) { + to++ + } + s.mu.Lock() + if s.gone { + s.mu.Unlock() + return + } + s.mu.Unlock() + init := "init-run.mp4" + if !exists(filepath.Join(s.dir, "init.mp4")) { + init = "init.mp4" + } + p, err := startFF(s.caps, s.self, s.Root, segArgs(s.dir, s.Path, s.Source, s.recipe, from, to, s.n, init)) + s.mu.Lock() + defer s.mu.Unlock() + if err != nil { + s.failure, s.finished = "restarting at "+strconv.Itoa(from*hlsSeconds)+" s: "+err.Error(), true + return + } + r := &segRun{from: from, to: to, proc: p} + s.run, s.proc, s.finished = r, p, false + go s.watchRun(r) + log.Printf("media: stream %s: run from %d s to %d s", s.ID, from*hlsSeconds, to*hlsSeconds) +} + +// Playlist is the whole film's, written before most of it exists. +func (s *Stream) Playlist() []byte { + var b strings.Builder + b.WriteString("#EXTM3U\n#EXT-X-VERSION:7\n") + fmt.Fprintf(&b, "#EXT-X-TARGETDURATION:%d\n", hlsSeconds) + b.WriteString("#EXT-X-MEDIA-SEQUENCE:0\n#EXT-X-PLAYLIST-TYPE:VOD\n#EXT-X-INDEPENDENT-SEGMENTS\n#EXT-X-MAP:URI=\"init.mp4\"\n") + for i := 0; i < s.n; i++ { + d := float64(hlsSeconds) + if i == s.n-1 { + d = s.Source.Duration - float64(i*hlsSeconds) + } + fmt.Fprintf(&b, "#EXTINF:%.6f,\nseg%05d.m4s\n", d, i) + } + b.WriteString("#EXT-X-ENDLIST\n") + return []byte(b.String()) +} + +// ErrNotInFilm is a segment the playlist lists but the film turned out not to reach. +var ErrNotInFilm = errors.New("past the end of the film") + +// Segment is the path of segment i once it is on disk. When the run going now will write it +// soon, it waits; otherwise this is a seek, and the run moves there — unless a newer request +// has moved it since this one arrived. +func (s *Stream) Segment(ctx context.Context, i int) (string, error) { + asked := time.Now() + deadline := asked.Add(segmentWait) + p := s.segPath(i) + for { + s.touch() + if exists(p) { + return p, nil + } + s.mu.Lock() + switch { + case s.gone: + s.mu.Unlock() + return "", errors.New("the stream was closed") + case i >= s.end: + s.mu.Unlock() + return "", ErrNotInFilm + case s.failure != "": + f := s.failure + s.mu.Unlock() + return "", errors.New(f) + } + r := s.run + coming := r != nil && !r.done() && r.from <= i && i < r.to && i <= s.written(r)+lookahead + move := !coming && !asked.Before(s.asked) + if move { + s.asked = asked + } + s.mu.Unlock() + if move { + s.runMu.Lock() + // still the newest, and still not on its way? (a run may have moved for a + // newer request while this one waited for the lock) + s.mu.Lock() + r = s.run + coming = r != nil && !r.done() && r.from <= i && i < r.to && i <= s.written(r)+lookahead + newest := !asked.Before(s.asked) + s.mu.Unlock() + if !coming && newest && !exists(p) { + log.Printf("media: stream %s: seek to %d s", s.ID, i*hlsSeconds) + s.startRun(i) + } + s.runMu.Unlock() + continue + } + if time.Now().After(deadline) { + return "", fmt.Errorf("segment %d didn't arrive in %s", i, segmentWait) + } + select { + case <-ctx.Done(): + return "", ctx.Err() + case <-time.After(150 * time.Millisecond): + } + } +} + +// SegmentIndex reads the number out of a segment's name. +func SegmentIndex(name string) (int, bool) { + m := segName.FindStringSubmatch(name) + if m == nil { + return 0, false + } + i, err := strconv.Atoi(m[1]) + return i, err == nil +} + +// ---- both kinds ---- + func (m *Streamer) Find(id string) (*Stream, bool) { m.mu.Lock() defer m.mu.Unlock() @@ -238,7 +542,11 @@ func (m *Streamer) Close(id string) { return } s.mu.Lock() + s.gone = true // a run ending now is ours, not a gap to fill p := s.proc + if s.run != nil { + s.run.stopped = true + } s.mu.Unlock() if p != nil { p.end() @@ -282,8 +590,8 @@ func (m *Streamer) Sweep(idle time.Duration) { } } -// File is the on-disk path of one of the stream's files — the playlist, the init segment -// or a media segment — and nothing else. +// File is the on-disk path of one of the stream's files — the playlist (copy streams; a +// seekable one's is Playlist), the init segment or a media segment — and nothing else. func (s *Stream) File(name string) (string, bool) { if !streamFile.MatchString(name) { return "", false @@ -300,16 +608,33 @@ func (s *Stream) running() bool { return s.proc != nil && !s.finished } -// Status is read from ffmpeg's own progress file and the playlist. +// Status is read from ffmpeg's own progress file and what is on disk. func (s *Stream) Status() StreamStatus { s.mu.Lock() st := StreamStatus{Running: s.proc != nil && !s.finished, Done: s.finished && s.failure == "", Error: s.failure} + seekable, end, run := s.Seekable, s.end, s.run s.mu.Unlock() - if b, err := os.ReadFile(filepath.Join(s.dir, "progress")); err == nil { + b, err := os.ReadFile(filepath.Join(s.dir, "progress")) + if err == nil { st.Progress, st.Speed = parseProgress(b) } - if b, err := os.ReadFile(filepath.Join(s.dir, "index.m3u8")); err == nil { - st.Segments = strings.Count(string(b), "#EXTINF") + if !seekable { + if b, err := os.ReadFile(filepath.Join(s.dir, "index.m3u8")); err == nil { + st.Segments = strings.Count(string(b), "#EXTINF") + } + return st + } + for i := 0; i < end; i++ { + if exists(s.segPath(i)) { + st.Segments++ + } + } + st.Progress = float64(st.Segments * hlsSeconds) + if st.Progress > s.Source.Duration { + st.Progress = s.Source.Duration + } + if run != nil && st.Running { + st.From = float64(run.from * hlsSeconds) } return st } diff --git a/internal/media/hls_test.go b/internal/media/hls_test.go index a4c67ed..d2c803f 100644 --- a/internal/media/hls_test.go +++ b/internal/media/hls_test.go @@ -2,9 +2,14 @@ package media import ( "context" + "errors" + "math" "os" "os/exec" "path/filepath" + "sort" + "strconv" + "strings" "testing" "time" ) @@ -38,12 +43,16 @@ func TestStreamWithFFmpeg(t *testing.T) { if err != nil { t.Fatal(err) } - if s.Width != 640 || s.Height != 360 || s.Engine == "" { + if s.Width != 640 || s.Height != 360 || s.Engine == "" || !s.Seekable { t.Fatalf("stream: %+v", s) } - if p, ok := s.File("index.m3u8"); !ok { + if _, ok := s.File("index.m3u8"); !ok { t.Fatal("no playlist") - } else if !hasSegments(p) { + } + if pl := string(s.Playlist()); !strings.Contains(pl, "#EXT-X-PLAYLIST-TYPE:VOD") || !strings.HasSuffix(pl, "#EXT-X-ENDLIST\n") || strings.Count(pl, "#EXTINF") != 3 { + t.Fatalf("playlist for 9 s:\n%s", pl) + } + if !exists(s.segPath(0)) { t.Fatal("Open returned before the first segment") } for _, bad := range []string{"../x", "progress", "seg1.m4s", "index.m3u8/", "init.mp4.tmp"} { @@ -63,7 +72,7 @@ func TestStreamWithFFmpeg(t *testing.T) { } time.Sleep(250 * time.Millisecond) } - if !st.Done || st.Error != "" || st.Segments < 2 || st.Progress < 8 { + if !st.Done || st.Error != "" || st.Segments != 3 || st.Progress < 8 { t.Fatalf("status at the end: %+v", st) } if init, _ := s.File("init.mp4"); !exists(init) { @@ -83,3 +92,129 @@ func TestStreamWithFFmpeg(t *testing.T) { t.Fatalf("left behind: %v", left) } } + +// A seek ahead of the encode: the run moves there, the segment comes from a run that started +// at it, the gaps are filled afterwards, and the segments of the different runs, joined, are +// one film with no hole and nothing twice beyond a frame at a join. +func TestSeekRestartsTheEncode(t *testing.T) { + if _, err := exec.LookPath("ffmpeg"); err != nil { + t.Skip("no ffmpeg") + } + c := Probe(context.Background()) + if c.Error != "" || EngineFor(c, "h264") == "" { + t.Skip("nothing here encodes h264") + } + dir := t.TempDir() + clip := filepath.Join(dir, "clip.mkv") + // big and busy enough that the first run is still near the start when the seek comes + out, err := exec.Command("ffmpeg", "-nostdin", "-v", "error", "-f", "lavfi", "-i", "testsrc2=size=1920x1080:rate=30000/1001:duration=64", "-f", "lavfi", "-i", "sine=frequency=440:duration=64", + "-c:v", "libx264", "-preset", "ultrafast", "-g", "250", "-c:a", "libopus", "-shortest", clip).CombinedOutput() + if err != nil { + t.Skipf("could not make a clip: %s", out) + } + src, err := Inspect(context.Background(), c, clip, false) + if err != nil { + t.Fatal(err) + } + m := NewStreamer(dir, "hamlab", 2) + s, err := m.Open(context.Background(), c, false, clip, "h264", "1080p", src) + if err != nil { + t.Fatal(err) + } + defer m.Close(s.ID) + if s.n != 16 { + t.Fatalf("%d segments for 64 s", s.n) + } + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + p, err := s.Segment(ctx, 12) + if err != nil { + t.Fatal(err) + } + s.mu.Lock() + from := s.run.from + s.mu.Unlock() + if from != 12 { + t.Fatalf("the run after the seek starts at segment %d, not 12 (the first run was already there?)", from) + } + if got := firstVideoPTS(t, filepath.Join(s.dir, "init.mp4"), p); math.Abs(got-(48+tsBase)) > 0.05 { + t.Fatalf("segment 12 starts at %.3f, want %d", got, 48+tsBase) + } + if _, err := s.Segment(ctx, 16); !errors.Is(err, ErrNotInFilm) { + t.Fatalf("past the end: %v", err) + } + var st StreamStatus + for i := 0; i < 480; i++ { + if st = s.Status(); st.Done || st.Error != "" { + break + } + time.Sleep(250 * time.Millisecond) + } + if !st.Done || st.Segments != 16 { + t.Fatalf("not filled in: %+v", st) + } + // the joined film: every segment in order behind the one init segment + joined := filepath.Join(dir, "joined.mp4") + f, _ := os.Create(joined) + for _, n := range append([]string{"init.mp4"}, segNames(16)...) { + b, err := os.ReadFile(filepath.Join(s.dir, n)) + if err != nil { + t.Fatal(err) + } + _, _ = f.Write(b) + } + f.Close() + if out, err := exec.Command("ffmpeg", "-v", "error", "-i", joined, "-f", "null", "-").CombinedOutput(); err != nil || strings.Contains(string(out), "error") { + t.Fatalf("the joined film doesn't decode cleanly: %v %s", err, out) + } + pts := videoPTS(t, joined) + if len(pts) < 1900 || pts[0] > tsBase+0.1 || pts[len(pts)-1] < tsBase+63.5 { + t.Fatalf("%d frames, %.3f .. %.3f", len(pts), pts[0], pts[len(pts)-1]) + } + for i := 1; i < len(pts); i++ { + if d := pts[i] - pts[i-1]; d > 0.07 { // a frame is 0.033; a join may lose up to one + t.Fatalf("a hole of %.3f s at %.3f", d, pts[i-1]) + } + } +} + +func segNames(n int) []string { + var out []string + for i := 0; i < n; i++ { + out = append(out, "seg"+strconv.FormatInt(int64(100000+i), 10)[1:]+".m4s") + } + return out +} + +func videoPTS(t *testing.T, file string) []float64 { + out, err := exec.Command("ffprobe", "-v", "error", "-select_streams", "v", "-show_entries", "packet=pts_time", "-of", "csv=p=0", file).Output() + if err != nil { + t.Fatal(err) + } + var pts []float64 + seen := map[string]bool{} + for _, l := range strings.Fields(string(out)) { + if seen[l] { // the same frame from both runs at a join + continue + } + seen[l] = true + if v, err := strconv.ParseFloat(l, 64); err == nil { + pts = append(pts, v) + } + } + sort.Float64s(pts) + return pts +} + +func firstVideoPTS(t *testing.T, init, seg string) float64 { + dir := t.TempDir() + j := filepath.Join(dir, "one.mp4") + a, _ := os.ReadFile(init) + b, _ := os.ReadFile(seg) + _ = os.WriteFile(j, append(a, b...), 0o644) + pts := videoPTS(t, j) + if len(pts) == 0 { + t.Fatal("no frames in the segment") + } + return pts[0] +} diff --git a/web/src/Player.tsx b/web/src/Player.tsx index a8607e9..cd17671 100644 --- a/web/src/Player.tsx +++ b/web/src/Player.tsx @@ -3,6 +3,8 @@ // picker: codec and size, and the machine re-encodes it to HLS on its own GPU (or CPU), // which plays natively on Safari and through hls.js everywhere else. The same picker is // one tap away even when the file plays, for a 36 Mbps capture over a slow link. +// A re-encode's timeline is the whole film from the start: seek anywhere, and the machine +// starts encoding from there (repackaging, which is near-instant, grows as it goes instead). import { useCallback, useEffect, useRef, useState } from 'react' import { api, apiPath, PlayCaps, PlayStatus, PlayStream } from './api' import { Btn, Spinner } from './ui' @@ -13,6 +15,7 @@ const QUALITY_KEY = 'hamlab.play.quality' const remember = (k: string, v: string) => { try { localStorage.setItem(k, v) } catch { /* private mode */ } } const recall = (k: string, d: string) => { try { return localStorage.getItem(k) || d } catch { return d } } +const clock = (sec: number) => { const t = Math.floor(sec), h = Math.floor(t / 3600), m = Math.floor((t % 3600) / 60), x = t % 60; return `${h ? `${h}:${String(m).padStart(2, '0')}` : m}:${String(x).padStart(2, '0')}` } const fmtMbps = (bps?: number) => (bps ? `${(bps / 1e6).toFixed(bps >= 10e6 ? 0 : 1)} Mbps` : '') const describe = (s: PlayStream) => { const v = s.source.video @@ -72,7 +75,10 @@ export default function Player({ path, sudo, name }: { path: string; sudo: boole const { default: Hls } = await import('hls.js') if (gone) return if (!Hls.isSupported()) { setErr('this browser has neither HLS nor MediaSource — nothing can play here'); return } - const h = new Hls({ maxBufferLength: 60, backBufferLength: 30 }) + // a segment the machine hasn't made yet is made when asked for (a seek): its first + // byte can take as long as the re-encode takes to start there and write it + const slow = { maxTimeToFirstByteMs: 50000, maxLoadTimeMs: 120000, timeoutRetry: { maxNumRetry: 2, retryDelayMs: 0, maxRetryDelayMs: 0 }, errorRetry: { maxNumRetry: 3, retryDelayMs: 1000, maxRetryDelayMs: 8000 } } + const h = new Hls({ maxBufferLength: 60, backBufferLength: 30, fragLoadPolicy: { default: slow } }) hls = h h.on(Hls.Events.ERROR, (_e, d) => { if (d.fatal) setErr(`playback failed: ${d.details}`) }) h.loadSource(url) @@ -131,7 +137,9 @@ export default function Player({ path, sudo, name }: { path: string; sudo: boole <>
{describe(stream.s)} - {status && status.running && · {status.speed ? `${status.speed.toFixed(1)}×` : ''} {stream.s.source.duration ? `${Math.min(100, Math.round(status.progress / stream.s.source.duration * 100))}% through` : `${Math.round(status.progress)} s done`} — seek within what's done} + {status && status.running && (stream.s.seekable + ? · {status.speed ? `${status.speed.toFixed(1)}× ` : ''}from {clock(status.from ?? 0)} · {Math.min(100, Math.round(status.progress / stream.s.source.duration * 100))}% ready — seek anywhere + : · {status.speed ? `${status.speed.toFixed(1)}×` : ''} {stream.s.source.duration ? `${Math.min(100, Math.round(status.progress / stream.s.source.duration * 100))}% through` : `${Math.round(status.progress)} s done`} — seek within what's done)} {status?.done && · all of it is ready} {status?.error && · stopped early: {status.error}} diff --git a/web/src/api.ts b/web/src/api.ts index 5d5c496..9783858 100644 --- a/web/src/api.ts +++ b/web/src/api.ts @@ -252,8 +252,10 @@ export interface ARCView { detail: ARCDetail; live: ARCLive | null } export interface FileEntry { name: string; type: 'dir' | 'file' | 'link' | 'other'; size: number; mode: string; perm: string; mtime: number; user: string; group: string; target?: string; broken?: boolean; preview?: 'image' | 'video' | 'audio' } export interface PlayCaps { codecs: { id: string; label: string; engine?: string }[]; qualities: { id: string; label: string; height: number }[]; error?: string } export interface PlayTrack { codec: string; profile?: string; pix_fmt?: string; width?: number; height?: number; fps?: number; channels?: number } -export interface PlayStream { id: string; codec: string; quality: string; engine: string; width: number; height: number; kbps: number; source: { container: string; duration: number; bit_rate?: number; video?: PlayTrack; audio?: PlayTrack } } -export interface PlayStatus { running: boolean; done: boolean; error?: string; progress: number; speed: number; segments: number } +export interface PlayStream { id: string; codec: string; quality: string; engine: string; width: number; height: number; kbps: number; seekable?: boolean; source: { container: string; duration: number; bit_rate?: number; video?: PlayTrack; audio?: PlayTrack } } +// progress: seconds written (repackaged), or seconds of the film on disk (a seekable re-encode, +// whose run now going started at `from`) +export interface PlayStatus { running: boolean; done: boolean; error?: string; progress: number; speed: number; segments: number; from?: number } export interface ThumbGPU { api: string; device: string; name?: string; codecs?: string[]; reason?: string; permission?: boolean } // How previews get made on the machine: ok with an engine, or why not and what would fix it. export interface ThumbStatus { ok: boolean; engine: 'gpu' | 'cpu' | ''; summary: string; ffmpeg?: string; gpus: ThumbGPU[]; unusable?: ThumbGPU[]; error?: string; install?: { pm: string; package: string }; gpu_fix?: boolean; user: string; cache: { path: string; entries: number; bytes: number } }