Live inputs are probed for 0.5 s instead of ffmpeg's 5 s analyze time. Custom channels decode their first 6 s as fast as they can, then keep real time 6 s ahead of the schedule, so the first segments are ready at once; the next video is picked by the stream's own clock. YouTube channels find the current and next videos' addresses and captions ahead of time, so a tune never waits on yt-dlp. Each tune logs and returns how long each step took, the app can report its time to first frame (/api/metrics), and the admin page shows median and 90th percentile by channel kind. scripts/tunebench.mjs measures tunes against a server.
- commit
- 22dba1662cc8d3c13b3386f4160f060ddea9317f
- parent
- d94340ee0dbb9682bf48c2a0065c6f62ee78bf23
- author
- FelineStateMachine <226808055+FelineStateMachine@users.noreply.github.com>
- kind
- feat risk score 2.0 of 3, Jev's estimate
| M | internal/ | 9 | + |
| M | internal/ | 10 | + |
| M | internal/ | 29 | +++ |
| M | internal/ | 1 | + |
| M | internal/ | 24 | +++ |
| A | internal/ | 53 | ++++++ |
| A | internal/ | 120 | +++++++++++++ |
| A | internal/ | 72 | ++++++++ |
| A | internal/ | 88 | ++++++++++ |
| M | internal/ | 3 | + |
| M | internal/ | 120 | +++++++++++-- |
| A | internal/ | 60 | +++++++ |
| M | internal/ | 2 | + |
| M | internal/ | 2 | +- |
| M | internal/ | 6 | +- |
| M | internal/ | 4 | +- |
| M | internal/ | 14 | +- |
| M | internal/ | 96 | +++++++++- |
| M | internal/ | 7 | +- |
| M | internal/ | 134 | +++------------ |
| M | internal/ | 35 | ++-- |
| A | internal/ | 274 | ++++++++++++++++++++++++++++++ |
| A | internal/ | 131 | ++++++++++++++ |
| M | internal/ | 1 | + |
| M | internal/ | 27 | ++- |
| A | scripts/ | 224 | +++++++++++++++++++++++++ |
@@ -34,6 +34,7 @@ import (
3434 3535 "airwaves/internal/guide"3636 "airwaves/internal/jellyfin"37+ "airwaves/internal/phase"3738 "airwaves/internal/vchan"3839 )3940 @@ -123,6 +124,7 @@ func (s *Server) Handler() http.Handler {
123124 mux.HandleFunc("GET /admin/api/youtube/search", s.searchYouTube)124125 mux.HandleFunc("GET /admin/api/youtube/channel", s.lookupYouTube)125126 mux.HandleFunc("GET /admin/api/youtube/playlist", s.lookupPlaylist)127+ mux.HandleFunc("GET /admin/api/tunes", getTunes)126128 s.addBackupRoutes(mux)127129 // Changes through the API, the agent tools' included, are backed up.128130 api := s.watch(mux)@@ -1092,6 +1094,13 @@ func decode(w http.ResponseWriter, r *http.Request, v any) bool {
10921094 return true10931095 }10941096 1097+// getTunes reports how long channel changes took lately, by kind of1098+// channel: the server's time to a ready stream, and the app's from the key1099+// press to the picture, as apps report it (POST /api/metrics).1100+func getTunes(w http.ResponseWriter, _ *http.Request) {1101+ writeJSON(w, map[string]any{"tunes": phase.Tunes.Summaries()})1102+}1103+10951104 func writeJSON(w http.ResponseWriter, v any) {10961105 w.Header().Set("Content-Type", "application/json")10971106 _ = json.NewEncoder(w).Encode(v)@@ -550,6 +550,16 @@ code { font: 400 12px/1.5 var(--f-mono); }
550550 .bk-go { width: 6px; height: 6px; border: solid var(--faint); border-width: 1.5px 1.5px 0 0; transform: rotate(45deg); margin-right: 2px; }551551 .bk-line:hover .bk-go, .bk-line[aria-current="page"] .bk-go { border-color: var(--amber); }552552 553+/* ---------- Channel changes ---------- */554+.tune-side { margin: 0 12px 12px; padding: 8px 12px; border-radius: 3px; box-shadow: inset 0 0 0 1px var(--line); }555+.tune-head { display: flex; align-items: baseline; justify-content: space-between; gap: 10px; margin-bottom: 4px; }556+.tune-head b { font: 500 10px/1.2 var(--f-mono); letter-spacing: .16em; text-transform: uppercase; color: var(--paper); }557+.tune-head span { font: 400 10px var(--f-mono); color: var(--faint); }558+.tune-table { width: 100%; border-collapse: collapse; font: 400 11px/1.6 var(--f-mono); color: var(--muted); font-variant-numeric: tabular-nums; }559+.tune-table th { font-weight: 400; text-align: left; color: var(--faint); }560+.tune-table thead th, .tune-table td { text-align: right; }561+.tune-table tbody th { color: var(--muted); }562+553563 .bk { padding: 36px var(--pad) 48px; max-width: 1060px; display: grid; gap: 20px; align-content: start; height: 100%; overflow-y: auto; scrollbar-width: thin; }554564 .bk-head h1 { margin: 0 0 10px; font: 800 46px/.95 var(--f-display); letter-spacing: .005em; }555565 .bk-lead { margin: 0; max-width: 74ch; font-size: 14px; color: var(--muted); }@@ -3589,6 +3589,35 @@
35893589 35903590 loadBackups();35913591 3592+ // ---------- Channel changes ----------3593+ // How long tuning in took lately, by kind of channel: the server's time to a ready stream,3594+ // and the app's from the key press to the picture, as the apps report it.3595+ const TUNE_KINDS = { antenna: 'Antenna', folder: 'Folders', jellyfin: 'Jellyfin', youtube: 'YouTube', weather: 'Weather' };3596+ async function loadTunes() {3597+ let rows = null;3598+ try {3599+ const r = await api('GET', '/tunes');3600+ rows = Array.isArray(r && r.tunes) ? r.tunes : [];3601+ } catch { /* kept as it was */ }3602+ if (rows) renderTunes(rows);3603+ setTimeout(loadTunes, document.hidden ? 300000 : 60000);3604+ }3605+ function renderTunes(rows) {3606+ const box = $('#tune-side');3607+ if (!box) return;3608+ const by = {};3609+ for (const r of rows) (by[r.kind] = by[r.kind] || {})[r.measure] = r;3610+ const kinds = Object.keys(TUNE_KINDS).filter((k) => by[k]);3611+ box.hidden = !kinds.length;3612+ const cell = (r) => h('td', { title: r ? `${r.count} lately, the slowest ${r.max.toFixed(1)} s` : null }, r ? `${r.p50.toFixed(1)} / ${r.p90.toFixed(1)}` : '-');3613+ box.replaceChildren(3614+ h('div', { class: 'tune-head' }, h('b', null, 'Channel changes'), h('span', { text: 'median / 90%, s' })),3615+ h('table', { class: 'tune-table' },3616+ h('thead', null, h('tr', null, h('td', null), h('th', { scope: 'col' }, 'Server'), h('th', { scope: 'col' }, 'App'))),3617+ h('tbody', null, kinds.map((k) => h('tr', null, h('th', { scope: 'row' }, TUNE_KINDS[k]), cell(by[k].ready), cell(by[k]['first frame']))))));3618+ }3619+ loadTunes();3620+35923621 // ---------- Global keys ----------35933622 // The skip link would otherwise change the hash, which the router reads as a channel.35943623 $('.skip').addEventListener('click', (e) => {@@ -22,6 +22,7 @@
2222 </header>2323 <section id="account" class="acct" aria-label="Jellyfin account"></section>2424 <section id="bk-side" class="bk-side" aria-label="Backups"></section>25+ <section id="tune-side" class="tune-side" aria-label="Channel changes" hidden></section>2526 <nav id="channels" class="chlist" aria-label="Channels">2627 <p class="side-msg">Loading channels</p>2728 </nav>@@ -15,8 +15,10 @@ import (
1515 "log"1616 "net/http"1717 "strings"18+ "time"1819 1920 "airwaves/internal/dvr"21+ "airwaves/internal/phase"2022 "airwaves/internal/reception"2123 "airwaves/internal/service"2224 "airwaves/internal/stream"@@ -92,6 +94,15 @@ type clientReq struct {
9294 From float64 `json:"from,omitempty"`9395 }9496 97+// metricsReq is a channel change as the app timed it.98+type metricsReq struct {99+ Number string `json:"number"`100+ // Kind is the channel's: "antenna", "folder", "jellyfin", "youtube"101+ // or "weather".102+ Kind string `json:"kind"`103+ FirstFrameMS float64 `json:"firstFrameMs"`104+}105+95106 type progressReq struct {96107 ID string `json:"id"`97108 Position float64 `json:"position"`@@ -155,6 +166,19 @@ func (s *Server) Handler() http.Handler {
155166 }156167 reply(w)(struct{}{}, b.Stop(r.Context(), req.Client))157168 })169+ // The app reports how long its channel changes took, from the key170+ // press to the first frame playing, for the admin page.171+ mux.HandleFunc("POST /api/metrics", func(w http.ResponseWriter, r *http.Request) {172+ var req metricsReq173+ if !decode(w, r, &req) {174+ return175+ }176+ phase.Tunes.Add(req.Kind, phase.FirstFrame, time.Duration(req.FirstFrameMS*float64(time.Millisecond)))177+ reply(w)(struct{}{}, nil)178+ })179+ mux.HandleFunc("GET /api/metrics", func(w http.ResponseWriter, r *http.Request) {180+ reply(w)(phase.Tunes.Summaries(), nil)181+ })158182 mux.HandleFunc("GET /api/dvr", func(w http.ResponseWriter, r *http.Request) {159183 reply(w)(b.DVR(r.Context()))160184 })@@ -0,0 +1,53 @@
1+package api2+3+import (4+ "encoding/json"5+ "net/http"6+ "net/http/httptest"7+ "strings"8+ "testing"9+10+ "airwaves/internal/phase"11+)12+13+// TestMetrics: the app's channel change times are kept by kind, and read14+// back as each kind's recent times.15+func TestMetrics(t *testing.T) {16+ srv := httptest.NewServer((&Server{Backend: &fake{}}).Handler())17+ defer srv.Close()18+ for _, body := range []string{19+ `{"number":"1.4","kind":"jellyfin","firstFrameMs":1500}`,20+ `{"number":"1.4","kind":"jellyfin","firstFrameMs":2500}`,21+ `{"number":"9.9","kind":"made up","firstFrameMs":100}`,22+ } {23+ resp, err := http.Post(srv.URL+"/api/metrics", "application/json", strings.NewReader(body))24+ if err != nil {25+ t.Fatal(err)26+ }27+ resp.Body.Close()28+ if resp.StatusCode != http.StatusOK {29+ t.Fatalf("POST %s: HTTP %d", body, resp.StatusCode)30+ }31+ }32+ resp, err := http.Get(srv.URL + "/api/metrics")33+ if err != nil {34+ t.Fatal(err)35+ }36+ defer resp.Body.Close()37+ var got []phase.Summary38+ if err := json.NewDecoder(resp.Body).Decode(&got); err != nil {39+ t.Fatal(err)40+ }41+ var jf *phase.Summary42+ for i, s := range got {43+ if s.Kind == "made up" {44+ t.Errorf("kept an unknown kind: %+v", s)45+ }46+ if s.Kind == "jellyfin" && s.Measure == phase.FirstFrame {47+ jf = &got[i]48+ }49+ }50+ if jf == nil || jf.Count < 2 || jf.Max < 2.5 {51+ t.Errorf("jellyfin first frames: %+v in %+v", jf, got)52+ }53+}@@ -0,0 +1,120 @@
1+// Package phase times the steps of starting a stream, so a slow channel2+// change shows where its time went: "tune 9.1 (antenna): first data 1.1s,3+// first frame 1.9s, first segment 3.8s, ready 5.6s". It also keeps recent4+// tune times by kind of channel, as the server and the app measured them,5+// for the admin page.6+package phase7+8+import (9+ "context"10+ "fmt"11+ "strings"12+ "sync"13+ "time"14+)15+16+// Timer records when each step of one stream's start happened.17+type Timer struct {18+ // Label names the stream in logs: "9.1 (antenna)".19+ Label string20+ // Kind is the kind of channel for the stats: "antenna", "folder",21+ // "jellyfin", "youtube" or "weather"; empty for none.22+ Kind string23+ start time.Time24+25+ mu sync.Mutex26+ marks []Mark27+}28+29+// Mark is one step, at a time since the start.30+type Mark struct {31+ Name string `json:"name"`32+ At time.Duration `json:"-"`33+ MS int64 `json:"ms"`34+}35+36+// New starts timing a stream.37+func New(label, kind string) *Timer {38+ return &Timer{Label: label, Kind: kind, start: time.Now()}39+}40+41+// Mark records that a step happened now. Only a step's first time counts.42+// A nil Timer records nothing.43+func (t *Timer) Mark(name string) {44+ if t == nil {45+ return46+ }47+ at := time.Since(t.start)48+ t.mu.Lock()49+ defer t.mu.Unlock()50+ for _, m := range t.marks {51+ if m.Name == name {52+ return53+ }54+ }55+ t.marks = append(t.marks, Mark{Name: name, At: at, MS: at.Milliseconds()})56+}57+58+// Has reports whether a step was marked.59+func (t *Timer) Has(name string) bool {60+ if t == nil {61+ return false62+ }63+ t.mu.Lock()64+ defer t.mu.Unlock()65+ for _, m := range t.marks {66+ if m.Name == name {67+ return true68+ }69+ }70+ return false71+}72+73+// Since is how long ago the timer started.74+func (t *Timer) Since() time.Duration {75+ if t == nil {76+ return 077+ }78+ return time.Since(t.start)79+}80+81+// Marks lists the steps in the order they happened.82+func (t *Timer) Marks() []Mark {83+ if t == nil {84+ return nil85+ }86+ t.mu.Lock()87+ defer t.mu.Unlock()88+ out := make([]Mark, len(t.marks))89+ copy(out, t.marks)90+ return out91+}92+93+// String lists the steps: "first data 1.1s, ready 5.6s".94+func (t *Timer) String() string {95+ var parts []string96+ for _, m := range t.Marks() {97+ parts = append(parts, fmt.Sprintf("%s %.1fs", m.Name, m.At.Seconds()))98+ }99+ if len(parts) == 0 {100+ return "no steps"101+ }102+ return strings.Join(parts, ", ")103+}104+105+type key struct{}106+107+// NewContext returns ctx carrying t, for the code that starts the stream108+// to mark its steps on.109+func NewContext(ctx context.Context, t *Timer) context.Context {110+ return context.WithValue(ctx, key{}, t)111+}112+113+// FromContext returns the Timer ctx carries, or nil.114+func FromContext(ctx context.Context) *Timer {115+ t, _ := ctx.Value(key{}).(*Timer)116+ return t117+}118+119+// Step marks a step on ctx's Timer, if it has one.120+func Step(ctx context.Context, name string) { FromContext(ctx).Mark(name) }@@ -0,0 +1,72 @@
1+package phase2+3+import (4+ "context"5+ "strings"6+ "testing"7+ "time"8+)9+10+func TestTimerMarksEachStepOnce(t *testing.T) {11+ tm := New("tune 9.1 (antenna)", "antenna")12+ tm.Mark("first data")13+ time.Sleep(20 * time.Millisecond)14+ tm.Mark("ready")15+ tm.Mark("first data") // again: the first time counts16+ ms := tm.Marks()17+ if len(ms) != 2 || ms[0].Name != "first data" || ms[1].Name != "ready" || ms[1].At < 20*time.Millisecond || ms[1].MS != ms[1].At.Milliseconds() {18+ t.Fatalf("marks %+v", ms)19+ }20+ if !tm.Has("ready") || tm.Has("first frame") {21+ t.Error("Has")22+ }23+ if s := tm.String(); !strings.HasPrefix(s, "first data 0.0s, ready 0.") {24+ t.Errorf("String %q", s)25+ }26+27+ // A context without a timer marks nothing, and doesn't fail.28+ Step(context.Background(), "x")29+ var none *Timer30+ none.Mark("x")31+ if none.Has("x") || none.Marks() != nil {32+ t.Error("a nil timer recorded a step")33+ }34+ ctx := NewContext(context.Background(), tm)35+ Step(ctx, "opened")36+ if !tm.Has("opened") || FromContext(ctx) != tm {37+ t.Error("Step didn't reach the context's timer")38+ }39+}40+41+func TestStatsSummaries(t *testing.T) {42+ var s Stats43+ for i := 1; i <= 10; i++ {44+ s.Add("folder", Ready, time.Duration(i)*100*time.Millisecond)45+ }46+ s.Add("folder", FirstFrame, 2*time.Second)47+ s.Add("nonsense", Ready, time.Second) // not a kind48+ s.Add("folder", "other", time.Second) // not a measure49+ s.Add("antenna", Ready, -time.Second) // can't be50+ s.Add("antenna", Ready, 10*time.Minute) // can't be either51+ got := s.Summaries()52+ if len(got) != 2 {53+ t.Fatalf("summaries %+v", got)54+ }55+ r := got[0]56+ if r.Kind != "folder" || r.Measure != Ready || r.Count != 10 || r.P50 != 0.5 || r.P90 != 0.9 || r.Max != 1 {57+ t.Errorf("ready %+v", r)58+ }59+ if f := got[1]; f.Measure != FirstFrame || f.Count != 1 || f.P50 != 2 {60+ t.Errorf("first frame %+v", f)61+ }62+63+ // Only the last ones are kept.64+ for range 2 * keep {65+ s.Add("youtube", Ready, time.Second)66+ }67+ for _, x := range s.Summaries() {68+ if x.Kind == "youtube" && x.Count != keep {69+ t.Errorf("kept %d, want %d", x.Count, keep)70+ }71+ }72+}@@ -0,0 +1,88 @@
1+package phase2+3+import (4+ "slices"5+ "sync"6+ "time"7+)8+9+// Measures of a channel change.10+const (11+ // Ready is the server's: from the tune call to the stream being ready.12+ Ready = "ready"13+ // FirstFrame is the app's: from the key press to the first frame14+ // playing.15+ FirstFrame = "first frame"16+)17+18+// Kinds are the kinds of channel the stats keep.19+var Kinds = []string{"antenna", "folder", "jellyfin", "youtube", "weather"}20+21+// keep is how many recent tunes each kind and measure keeps.22+const keep = 10023+24+// Stats keeps recent tune times by kind of channel and measure.25+type Stats struct {26+ mu sync.Mutex27+ samples map[[2]string][]time.Duration // by kind and measure, oldest first28+}29+30+// Tunes are this server's.31+var Tunes = &Stats{}32+33+// Add records a tune's time. Kinds and measures not known, and times34+// that can't be right, are left out.35+func (s *Stats) Add(kind, measure string, d time.Duration) {36+ if !slices.Contains(Kinds, kind) || (measure != Ready && measure != FirstFrame) || d <= 0 || d > 2*time.Minute {37+ return38+ }39+ s.mu.Lock()40+ defer s.mu.Unlock()41+ if s.samples == nil {42+ s.samples = map[[2]string][]time.Duration{}43+ }44+ k := [2]string{kind, measure}45+ xs := append(s.samples[k], d)46+ if len(xs) > keep {47+ xs = xs[len(xs)-keep:]48+ }49+ s.samples[k] = xs50+}51+52+// Summary describes one kind and measure's recent tunes, in seconds.53+type Summary struct {54+ Kind string `json:"kind"`55+ Measure string `json:"measure"`56+ Count int `json:"count"`57+ P50 float64 `json:"p50"`58+ P90 float64 `json:"p90"`59+ Max float64 `json:"max"`60+}61+62+// Summaries lists the kinds and measures with tunes, by kind in the order63+// of Kinds, the server's measure first.64+func (s *Stats) Summaries() []Summary {65+ s.mu.Lock()66+ defer s.mu.Unlock()67+ var out []Summary68+ for _, kind := range Kinds {69+ for _, measure := range []string{Ready, FirstFrame} {70+ xs := slices.Clone(s.samples[[2]string{kind, measure}])71+ if len(xs) == 0 {72+ continue73+ }74+ slices.Sort(xs)75+ out = append(out, Summary{76+ Kind: kind, Measure: measure, Count: len(xs),77+ P50: quantile(xs, 0.5).Seconds(), P90: quantile(xs, 0.9).Seconds(), Max: xs[len(xs)-1].Seconds(),78+ })79+ }80+ }81+ return out82+}83+84+// quantile is the q quantile of sorted xs, nearest rank.85+func quantile(xs []time.Duration, q float64) time.Duration {86+ i := int(q*float64(len(xs))+0.999999) - 187+ return xs[min(max(i, 0), len(xs)-1)]88+}@@ -31,6 +31,7 @@ import (
3131 "airwaves/internal/geo"3232 "airwaves/internal/guide"3333 "airwaves/internal/lineup"34+ "airwaves/internal/phase"3435 "airwaves/internal/signal"3536 "airwaves/internal/store"3637 "airwaves/internal/stream"@@ -830,6 +831,7 @@ func (s *Service) Tune(ctx context.Context, client, number string) (*stream.Play
830831 return nil, s.streamErr831832 }832833 if v := s.customChannel(number); v != nil {834+ ctx = phase.NewContext(ctx, phase.New("tune "+number+" ("+vchan.KindOf(v)+")", vchan.KindOf(v)))833835 // The app gets subtitles as text, a WebVTT track it shows itself,834836 // turning it on for programs not in English, so none are burned in.835837 var script *cc.Script@@ -844,6 +846,7 @@ func (s *Service) Tune(ctx context.Context, client, number string) (*stream.Play
844846 return s.streams.Start(ctx, client, in)845847 }846848 if s.tvhTuner != nil {849+ ctx = phase.NewContext(ctx, phase.New("tune "+number+" (antenna)", "antenna"))847850 var pb *stream.Playback848851 err := s.tuneAntenna(ctx, client, number, func(ctx context.Context, in tuner.Input) error {849852 var err error@@ -4,6 +4,7 @@
44 package stream55 66 import (7+ "bufio"78 "bytes"89 "context"910 "crypto/rand"@@ -12,6 +13,7 @@ import (
1213 "errors"1314 "fmt"1415 "io"16+ "log"1517 "net"1618 "net/http"1719 "os"@@ -22,6 +24,7 @@ import (
2224 "sync"2325 "time"2426 27+ "airwaves/internal/phase"2528 "airwaves/internal/tuner"2629 )2730 @@ -40,6 +43,9 @@ type Playback struct {
4043 // Offset is where a recording's playback starts, in seconds, so the4144 // player can show positions in the whole recording.4245 Offset float64 `json:"offset,omitempty"`46+ // Timing lists when each step of starting the stream happened, from47+ // the tune call (see phase).48+ Timing []phase.Mark `json:"timing,omitempty"`4349 }4450 4551 // Server owns the ffmpeg sessions and serves their files under /live/.@@ -112,16 +118,32 @@ func (s *Server) FFprobe() string { return filepath.Join(filepath.Dir(s.ffmpeg),
112118 // the same client overlap (fast channel surfing) the last one wins and113119 // earlier ones fail; no ffmpeg process outlives its replacement.114120 func (s *Server) Start(ctx context.Context, client string, in tuner.Input) (*Playback, error) {121+ // The steps are timed from the tune call when its context has a timer.122+ t := phase.FromContext(ctx)123+ if t == nil {124+ t = phase.New("stream", "")125+ }115126 // Stop first so a network tuner frees its tuner before the next tune.116- s.Stop(client)127+ if s.stop(client) {128+ t.Mark("previous stopped")129+ }117130 118131 id := newID()119132 dir := filepath.Join(s.root, id)120133 if err := os.MkdirAll(dir, 0o755); err != nil {121134 return nil, err122135 }123- runCtx, cancel := context.WithCancel(context.Background())124- cmd := exec.CommandContext(runCtx, s.ffmpeg, ffmpegArgs(in, dir)...)136+ runCtx, cancel := context.WithCancel(phase.NewContext(context.Background(), t))137+ // ffmpeg reports its progress on fd 3, which tells when it has138+ // encoded its first frame.139+ progress, progressW, err := os.Pipe()140+ if err != nil {141+ cancel()142+ return nil, err143+ }144+ args := append([]string{"-progress", "pipe:3", "-stats_period", "0.1"}, ffmpegArgs(in, dir)...)145+ cmd := exec.CommandContext(runCtx, s.ffmpeg, args...)146+ cmd.ExtraFiles = []*os.File{progressW}125147 sess := &session{126148 id: id, client: client, dir: dir, started: time.Now(),127149 cancel: cancel, done: make(chan struct{}), stderr: &tail{max: 4096},@@ -129,20 +151,27 @@ func (s *Server) Start(ctx context.Context, client string, in tuner.Input) (*Pla
129151 cmd.Stderr = sess.stderr130152 var source io.WriteCloser131153 if in.Source != nil {132- var err error133154 if source, err = cmd.StdinPipe(); err != nil {155+ progress.Close()156+ progressW.Close()134157 cancel()135158 return nil, err136159 }137160 }138- if err := cmd.Start(); err != nil {161+ err = cmd.Start()162+ progressW.Close()163+ if err != nil {164+ progress.Close()139165 cancel()140166 return nil, fmt.Errorf("start ffmpeg: %w", err)141167 }168+ t.Mark("ffmpeg started")169+ go watchProgress(progress, func() { t.Mark("first frame") })142170 if source != nil {143171 go func() {144172 defer source.Close()145- if err := in.Source(runCtx, source); err != nil && runCtx.Err() == nil {173+ w := &firstWrite{w: source, mark: func() { t.Mark("first data") }}174+ if err := in.Source(runCtx, w); err != nil && runCtx.Err() == nil {146175 fmt.Fprintf(sess.stderr, "%v\n", err)147176 }148177 }()@@ -176,43 +205,64 @@ func (s *Server) Start(ctx context.Context, client string, in tuner.Input) (*Pla
176205 x.stop()177206 }178207 179- if err := waitReady(ctx, sess, playlists(in)); err != nil {208+ if err := waitReady(ctx, sess, playlists(in), t); err != nil {180209 s.mu.Lock()181210 if s.sessions[client] == sess {182211 delete(s.sessions, client)183212 }184213 s.mu.Unlock()185214 sess.stop()215+ log.Printf("%s failed after %.1fs: %s", t.Label, t.Since().Seconds(), t)186216 return nil, err187217 }188218 if err := os.WriteFile(filepath.Join(dir, "master.m3u8"), []byte(masterPlaylist(in)), 0o644); err != nil {189219 s.Stop(client)190220 return nil, err191221 }192- pb := &Playback{ID: id, Path: "/live/" + id + "/master.m3u8", Note: in.Note, Offset: in.Offset}222+ t.Mark("ready")223+ pb := &Playback{ID: id, Path: "/live/" + id + "/master.m3u8", Note: in.Note, Offset: in.Offset, Timing: t.Marks()}193224 if s.ln != nil {194225 pb.URL = "http://" + s.ln.Addr().String() + pb.Path195226 }227+ log.Printf("%s: %s", t.Label, t)228+ phase.Tunes.Add(t.Kind, phase.Ready, t.Since())196229 return pb, nil197230 }198231 199-// waitReady blocks until every media playlist lists two segments.200-func waitReady(ctx context.Context, sess *session, lists []string) error {232+// firstWrite marks the first write to w.233+type firstWrite struct {234+ w io.Writer235+ once sync.Once236+ mark func()237+}238+239+func (f *firstWrite) Write(p []byte) (int, error) {240+ f.once.Do(f.mark)241+ return f.w.Write(p)242+}243+244+// waitReady blocks until every media playlist lists two segments, marking245+// on t when the first segment is done.246+func waitReady(ctx context.Context, sess *session, lists []string, t *phase.Timer) error {201247 ctx, cancel := context.WithTimeout(ctx, 25*time.Second)202248 defer cancel()203- t := time.NewTicker(150 * time.Millisecond)204- defer t.Stop()249+ tick := time.NewTicker(50 * time.Millisecond)250+ defer tick.Stop()205251 for {206252 select {207253 case <-ctx.Done():208254 return fmt.Errorf("stream did not start in time: %s", sess.stderr.String())209255 case <-sess.done:210256 return fmt.Errorf("ffmpeg exited: %s", sess.stderr.String())211- case <-t.C:257+ case <-tick.C:212258 ready := 0213- for _, name := range lists {259+ for i, name := range lists {214260 b, err := os.ReadFile(filepath.Join(sess.dir, name))215- if err == nil && bytes.Count(b, []byte("#EXTINF")) >= 2 {261+ n := bytes.Count(b, []byte("#EXTINF"))262+ if err == nil && i == 0 && n >= 1 {263+ t.Mark("first segment")264+ }265+ if err == nil && n >= 2 {216266 ready++217267 }218268 }@@ -223,6 +273,24 @@ func waitReady(ctx context.Context, sess *session, lists []string) error {
223273 }224274 }225275 276+// watchProgress reads ffmpeg's progress reports until it exits, calling277+// encoded once the first frame is out.278+func watchProgress(r io.ReadCloser, encoded func()) {279+ defer r.Close()280+ sc := bufio.NewScanner(r)281+ done := false282+ for sc.Scan() {283+ if n, ok := strings.CutPrefix(sc.Text(), "frame="); ok && !done {284+ if frames, _ := strconv.Atoi(strings.TrimSpace(n)); frames > 0 {285+ encoded()286+ done = true287+ }288+ }289+ }290+ // ffmpeg must never wait on a report nobody reads.291+ _, _ = io.Copy(io.Discard, r)292+}293+226294 // liveWindow is how far back a viewer can pause or rewind live TV.227295 const liveWindow = 30 * time.Minute228296 @@ -259,6 +327,7 @@ func subtitled(in tuner.Input) bool { return in.Subtitles != nil && !in.VOD }
259327 // stream by libx264.260328 func ffmpegArgs(in tuner.Input, dir string) []string {261329 args := []string{"-hide_banner", "-loglevel", "error", "-nostdin"}330+ args = append(args, probeArgs(in)...)262331 args = append(args, in.Args...)263332 args = append(args, "-map", in.Video)264333 for _, a := range in.Audio {@@ -304,6 +373,21 @@ func ffmpegArgs(in tuner.Input, dir string) []string {
304373 )305374 }306375 376+// probeArgs keep ffmpeg's first look at a live input short. Before it377+// starts, ffmpeg reads an MPEG-TS input for as long as it may analyze it378+// (5 seconds by default) even once it knows every stream, and for a live379+// input that is as long a wait. Half a second finds a broadcast's audio380+// tracks and a custom channel's picture and sound. A broadcast's picture381+// size may only come with its first full frame, later, which ffmpeg takes382+// in stride. A recording is read as fast as it can be, so it keeps the383+// default.384+func probeArgs(in tuner.Input) []string {385+ if in.VOD {386+ return nil387+ }388+ return []string{"-probesize", "2000000", "-analyzeduration", "500000"}389+}390+307391 // masterPlaylist describes the session's streams to the player: the video,308392 // each audio track as a selectable rendition, and closed captions.309393 func masterPlaylist(in tuner.Input) string {@@ -375,7 +459,10 @@ func TrackName(a tuner.AudioTrack, i int) (name, lang string) {
375459 }376460 377461 // Stop ends client's stream and removes its files.378-func (s *Server) Stop(client string) {462+func (s *Server) Stop(client string) { s.stop(client) }463+464+// stop ends client's stream, reporting whether it had one.465+func (s *Server) stop(client string) bool {379466 s.mu.Lock()380467 sess := s.sessions[client]381468 delete(s.sessions, client)@@ -383,6 +470,7 @@ func (s *Server) Stop(client string) {
383470 if sess != nil {384471 sess.stop()385472 }473+ return sess != nil386474 }387475 388476 func (sess *session) stop() {@@ -0,0 +1,60 @@
1+package stream2+3+import (4+ "context"5+ "io"6+ "os/exec"7+ "slices"8+ "testing"9+ "time"10+11+ "airwaves/internal/phase"12+ "airwaves/internal/tuner"13+)14+15+// TestLiveStartTimesItsSteps feeds a live MPEG-TS source in real time, as16+// a custom channel does: the stream's steps are timed in order, and it's17+// ready once two segments' worth (four seconds) has come in, not after18+// ffmpeg's default five seconds of looking at its input first.19+func TestLiveStartTimesItsSteps(t *testing.T) {20+ if testing.Short() {21+ t.Skip("starts ffmpeg")22+ }23+ s, err := NewServer(true)24+ if err != nil {25+ t.Skip(err)26+ }27+ defer s.Close()28+ in := tuner.Input{29+ Args: []string{"-f", "mpegts", "-i", "pipe:0"}, Video: "0:v:0", Audio: []tuner.AudioTrack{{Map: "0:a:0"}},30+ Source: func(ctx context.Context, w io.Writer) error {31+ cmd := exec.CommandContext(ctx, s.FFmpeg(), "-hide_banner", "-loglevel", "error", "-nostdin",32+ "-re", "-f", "lavfi", "-i", "testsrc2=s=640x360:r=30", "-re", "-f", "lavfi", "-i", "sine=f=440:r=48000",33+ "-c:v", "libx264", "-preset", "veryfast", "-tune", "zerolatency", "-g", "60", "-c:a", "aac", "-f", "mpegts", "pipe:1")34+ cmd.Stdout = w35+ return cmd.Run()36+ },37+ }38+ timer := phase.New("test", "folder")39+ pb, err := s.Start(phase.NewContext(context.Background(), timer), "test", in)40+ if err != nil {41+ t.Fatal(err)42+ }43+ defer s.Stop("test")44+ var names []string45+ var ready time.Duration46+ for _, m := range pb.Timing {47+ names = append(names, m.Name)48+ if m.Name == "ready" {49+ ready = m.At50+ }51+ }52+ want := []string{"ffmpeg started", "first data", "first frame", "first segment", "ready"}53+ if !slices.Equal(names, want) {54+ t.Errorf("steps %q, want %q", names, want)55+ }56+ t.Logf("steps: %s", timer)57+ if ready < 3*time.Second || ready > 5*time.Second {58+ t.Errorf("ready after %v, want about four seconds", ready)59+ }60+}@@ -12,6 +12,7 @@ import (
1212 "time"1313 1414 "airwaves/internal/cc"15+ "airwaves/internal/phase"1516 )1617 1718 // Videos' English captions reach viewers as CEA-608 closed captions in the@@ -218,6 +219,7 @@ func subtitle(ctx context.Context, ffmpeg string, it item, skip int64, s *span,
218219 wait, cancel := context.WithTimeout(ctx, captionsWait)219220 _, _ = it.Captions.get(wait)220221 cancel()222+ phase.Step(ctx, "captions")221223 if cues, ok := it.Captions.ready(); ok {222224 out.captions.set(s, cc.Shift(cues, -offset))223225 return ""@@ -92,7 +92,7 @@ func TestFeedCountsWholeFrames(t *testing.T) {
9292 if err != nil {9393 t.Fatal(err)9494 }95- f := newFeed(w, 4, 10)95+ f := newFeed(w, 4, 10, nil)9696 got := make(chan []byte)9797 go func() {9898 b, _ := readAll(r)@@ -447,7 +447,7 @@ func jellyfinItem(c *jellyfin.Client, v jellyfin.Video, mbps float64) (item, boo
447447 return it, true448448 }449449 450-// jellyfinInput reads a video from the server in real time: the original450+// jellyfinInput reads a video from the server: the original451451 // file, seeking here, or with a bitrate cap the server's transcode, which452452 // it seeks in. The token goes in a header, never the URL, so it stays out453453 // of logs.@@ -462,9 +462,9 @@ func jellyfinInput(c *jellyfin.Client, v jellyfin.Video, mbps float64) func(time
462462 if offset > 0 {463463 args = append(args, "-ss", strconv.FormatFloat(offset.Seconds(), 'f', 3, 64))464464 }465- return append(args, "-re", "-i", c.FileURL(v))465+ return append(args, "-i", c.FileURL(v))466466 }467467 t := jellyfin.Transcode{Bitrate: int(mbps * 1e6), MaxWidth: loopWidth, MaxHeight: loopHeight}468- return append(args, "-re", "-i", c.TranscodeURL(v, offset, t))468+ return append(args, "-i", c.TranscodeURL(v, offset, t))469469 }470470 }@@ -300,7 +300,9 @@ func TestJellyfinInput(t *testing.T) {
300300 args := decodeArgs(it, 90*loopFPS) // 90 s in301301 in := at(args, "-i")302302 h := at(args, "-headers")303- if in < 1 || args[in-1] != "-re" || h < 0 || h > in {303+ // Not read in real time: the stream is paced on its way to the304+ // encoder.305+ if in < 1 || slices.Contains(args, "-re") || h < 0 || h > in {304306 t.Fatalf("maxBitrate %v: args %q", mbps, args)305307 }306308 if hdr := args[h+1]; !strings.HasPrefix(hdr, "Authorization: MediaBrowser ") || !strings.Contains(hdr, `Token="session-1"`) || !strings.HasSuffix(hdr, "\r\n") {@@ -61,6 +61,14 @@ const youtubeConfig = "youtube.json"
6161 6262 var numbered = regexp.MustCompile(`^(\d+[.-]\d+)\s+(.+)$`)6363 64+// closeChannel stops what a channel that left the library does in the65+// background.66+func closeChannel(c Channel) {67+ if y, ok := c.(*YouTube); ok {68+ y.close()69+ }70+}71+6472 // Channels returns the folder channels that have something to play, by6573 // number.6674 func (l *Library) Channels() []Channel {@@ -151,6 +159,9 @@ func (l *Library) scan() []Entry {
151159 }152160 c, ok := l.channels[fd.path]153161 if !ok || c.sig != sig {162+ if ok {163+ closeChannel(c.Channel)164+ }154165 c = libraryChannel{sig: sig}155166 if strings.Contains(sig, "\x00jellyfin\x00") {156167 j := &Jellyfin{Num: fd.number, Title: fd.name, Config: cfg, FFmpeg: l.FFmpeg, Loudness: l.Loudness}@@ -174,8 +185,9 @@ func (l *Library) scan() []Entry {
174185 keep[fd.path] = true175186 list = append(list, Entry{Folder: filepath.Base(fd.path), Channel: c.Channel})176187 }177- for p := range l.channels {188+ for p, c := range l.channels {178189 if !keep[p] {190+ closeChannel(c.Channel)179191 delete(l.channels, p)180192 }181193 }@@ -19,6 +19,7 @@ import (
1919 "unicode/utf8"2020 2121 "airwaves/internal/cc"22+ "airwaves/internal/phase"2223 )2324 2425 // item is one video in a looping channel.@@ -34,8 +35,9 @@ type item struct {
3435 Episode string3536 Image string3637 // Input, when set, returns ffmpeg input options (ending "-i <url>") that37- // play the video in real time from offset in, for sources that seek on38- // their side. Otherwise Path is read with a local seek.38+ // play the video from offset in, for sources that seek on their side.39+ // Otherwise Path is read with a local seek. The stream's pace is set40+ // as it goes to the encoder (see pacer), not by the input.3941 Input func(offset time.Duration) []string4042 // AudioStream is the sound's stream specifier in Path's or Input's4143 // streams; "0:a:0" when empty.@@ -61,8 +63,7 @@ type item struct {
6163 // opened is how to play an item found when it opens.6264 type opened struct {6365 // Args are ffmpeg input options for one or more inputs (picture first,64- // each ending "-i <url>") that play in real time from the offset asked65- // for.66+ // each ending "-i <url>") that play from the offset asked for.6667 Args []string6768 // Audio is the sound's stream specifier ("1:a:0"), empty for silence.6869 Audio string@@ -76,6 +77,11 @@ const (
7677 loopHeight = 7207778 loopFPS = 307879 sampleRate = 4800080+ // lead is how far ahead of real time a stream starts: its first81+ // seconds are decoded and encoded as fast as they can be, so a viewer82+ // tuning in waits for none of them, and after that the stream keeps83+ // that far ahead of the clock (see pacer).84+ lead = 6 * time.Second7985 // maxPrograms bounds a guide request on a channel of very short videos.8086 maxPrograms = 20008187 )@@ -393,8 +399,11 @@ func playCues(ctx context.Context, w io.Writer, ffmpeg, label string, cue cueFun
393399 close(encDone)394400 cancel() // the viewer left, or the encoder failed395401 }()396- out.video = newFeed(encVideo.w, 512, frameBytes) // up to 32 MB, under a second397- out.audio = newFeed(encAudio.w, 128, 4)402+ // The picture sets the pace: the decoders run as fast as it lets them,403+ // sound and all.404+ out.video = newFeed(encVideo.w, 512, frameBytes, &pacer{per: loopFPS, lead: lead, quit: ctx.Done()}) // up to 32 MB, under a second405+ out.audio = newFeed(encAudio.w, 128, 4, nil)406+ out.video.first = func() { phase.Step(ctx, "decoding") }398407 finish := func() {399408 // Ending the feeds ends the encoder.400409 out.video.close()@@ -403,12 +412,23 @@ func playCues(ctx context.Context, w io.Writer, ffmpeg, label string, cue cueFun
403412 }404413 defer finish()405414 415+ // What comes next is cued at the stream's own clock: where it was last416+ // cued, plus the frames since. It runs ahead of the clock on the wall417+ // by up to lead, which so holds from one video to the next; a stream418+ // that fell behind is cued at the wall's.419+ var at time.Time420+ var cued int64 // frames when last cued406421 ok := true407422 for ctx.Err() == nil {408- it, skip, err := cue(time.Now(), ok)423+ now := time.Now()424+ if s := at.Add(frameDuration(out.video.units - cued)); !at.IsZero() && s.After(now) {425+ now = s426+ }427+ it, skip, err := cue(now, ok)409428 if err != nil {410429 return err411430 }431+ at, cued = now, out.video.units412432 ok = decode(ctx, ffmpeg, label, it, skip, out)413433 }414434 finish()@@ -459,23 +479,67 @@ type feed struct {
459479 units int64 // from the decoders that have finished460480 bytes atomic.Int64 // fed so far461481 closed sync.Once482+ first func() // called as the first unit arrives, when set462483 }463484 464-func newFeed(dst *os.File, chunks, unit int) *feed {485+// newFeed starts a feed to dst. With pace set, units go to dst no faster486+// than pace allows; otherwise as soon as they arrive.487+func newFeed(dst *os.File, chunks, unit int, pace *pacer) *feed {465488 f := &feed{buf: make(chan []byte, chunks), unit: unit}466489 go func() {467490 defer dst.Close()468491 failed := false492+ var sent int64469493 for b := range f.buf {494+ if !failed && pace != nil {495+ failed = !pace.wait(sent / int64(unit))496+ }470497 if !failed {471498 _, err := dst.Write(b)472499 failed = err != nil473500 }501+ sent += int64(len(b))474502 }475503 }()476504 return f477505 }478506 507+// pacer holds a feed to real time, but for a lead: unit n goes out lead508+// before n/per seconds after the first did. Units that come late go out509+// at once, so a stream that falls behind (a slow start, a network510+// stall) catches up as fast as its decoder can.511+type pacer struct {512+ per int64 // units a second513+ lead time.Duration514+ quit <-chan struct{}515+ start time.Time516+ timer *time.Timer517+}518+519+// wait waits until unit n is due, and reports false if quit came first.520+func (p *pacer) wait(n int64) bool {521+ now := time.Now()522+ if p.start.IsZero() {523+ p.start = now524+ }525+ due := p.start.Add(time.Duration(n/p.per)*time.Second + time.Duration(n%p.per)*time.Second/time.Duration(p.per) - p.lead)526+ d := due.Sub(now)527+ if d <= 0 {528+ return true529+ }530+ if p.timer == nil {531+ p.timer = time.NewTimer(d)532+ } else {533+ p.timer.Reset(d)534+ }535+ select {536+ case <-p.timer.C:537+ return true538+ case <-p.quit:539+ return false540+ }541+}542+479543 // source returns a pipe for a decoder to write to, and done, which waits480544 // for what the decoder wrote to pass and returns how many units it was. A481545 // decoder cut off mid-frame has its frame made up, so the next decoder's@@ -494,6 +558,9 @@ func (f *feed) source(process func([]byte)) (w *os.File, done func() int64, err
494558 b := make([]byte, relayChunk)495559 k, err := io.ReadFull(r, b)496560 if k > 0 {561+ if n == 0 && f.first != nil {562+ f.first()563+ }497564 if process != nil {498565 process(b[:k-k%f.unit])499566 }@@ -562,7 +629,7 @@ func encodeArgs() []string {
562629 }563630 }564631 565-// decodeArgs plays one video in real time from skip frames in, as raw632+// decodeArgs plays one video from skip frames in, as raw566633 // picture (fd 3) and sound (fd 4): letterboxed to 720p at 30 fps, and padded567634 // or trimmed to exactly the frames the schedule gives it, so picture and568635 // sound stay in step from one video to the next.@@ -580,7 +647,7 @@ func (it item) inputs(skip int64) (input []string, audio string) {
580647 if skip > 0 {581648 input = append(input, "-ss", strconv.FormatFloat(float64(skip)/loopFPS, 'f', 3, 64))582649 }583- input = append(input, "-re", "-i", it.Path)650+ input = append(input, "-i", it.Path)584651 }585652 if !it.Audio {586653 return input, ""@@ -639,18 +706,21 @@ func decode(ctx context.Context, ffmpeg, label string, it item, skip int64, out
639706 // Opened sources are logged by Path, an ID: their addresses are640707 // long and may carry credentials.641708 name = it.Path642- began := time.Now()643709 o, err := it.Open(ctx, frameDuration(skip))644710 if err != nil {645711 if ctx.Err() != nil {646712 return true647713 }648714 log.Printf("%s: %s: %v", label, name, err)649- if left := it.Frames - skip - framesIn(time.Since(began)); left > 0 {715+ phase.Step(ctx, "open failed")716+ // All of it: the time spent opening played nothing, so the717+ // stream's clock didn't move (see playCues).718+ if left := it.Frames - skip; left > 0 {650719 decode(ctx, ffmpeg, label, item{Slate: true, Title: standBy, Frames: left}, 0, out)651720 }652721 return false653722 }723+ phase.Step(ctx, "opened")654724 input, sound = o.Args, o.Audio655725 it.AudioLang, it.Captions = o.AudioLang, o.Captions656726 default:@@ -726,7 +796,7 @@ func slateArgs(ffmpeg, caption string, frames int64) []string {
726796 }727797 src += fmt.Sprintf(",drawtext=text='%s':fontcolor=0x8c939f:fontsize=%d:x=(w-tw)/2:y=(h-th)/2", text, size)728798 }729- return decodeInputs([]string{"-re", "-f", "lavfi", "-i", src}, "", frames, "")799+ return decodeInputs([]string{"-f", "lavfi", "-i", src}, "", frames, "")730800 }731801 732802 // slateText is a caption as drawtext's quoted text: letters, digits and@@ -254,7 +254,8 @@ func BenchmarkLeveler(b *testing.B) {
254254 255255 // TestLevelingInTheStream plays two clips 12 dB apart through the engine,256256 // twice: the first time each is metered and remembered, so the second257-// time both start at the target, and the stream keeps time across them.257+// time both start at the target, and the stream keeps time across them,258+// its lead ahead.258259 func TestLevelingInTheStream(t *testing.T) {259260 ffmpeg, err := exec.LookPath("ffmpeg")260261 if err != nil {@@ -298,8 +299,8 @@ func TestLevelingInTheStream(t *testing.T) {
298299 t.Fatal(err)299300 }300301 probe, _ := exec.Command(ffprobe, "-v", "error", "-show_entries", "format=duration", "-of", "default=nw=1:nk=1", ts).Output()301- if d, _ := strconv.ParseFloat(strings.TrimSpace(string(probe)), 64); d < 9 || d > 11.5 {302- t.Errorf("%s: played %.1fs in 10.5s", name, d)302+ if d, _ := strconv.ParseFloat(strings.TrimSpace(string(probe)), 64); d < 10.5+lead.Seconds()-2.5 || d > 10.5+lead.Seconds()+0.5 {303+ t.Errorf("%s: played %.1fs in 10.5s with a %v lead", name, d, lead)303304 }304305 return ts305306 }@@ -59,23 +59,17 @@ type YouTube struct {
5959 busy bool // a refresh is running6060 tried map[string]time.Time // failed listings, by URL6161 feedDown map[string]string // failing feeds by URL: how they're stood in for62- resolved map[string]ytResolved // found ahead, by video ID62+ resolved map[string]ytResolved // found, by video ID (see ytahead.go)6363 ahead map[string]*ytAhead // being found ahead64+ missed map[string]time.Time // failed to be found ahead, by video ID65+ touched time.Time // when the channel was last tended66+ warming bool // finding what's on ahead (see warm)67+ closed bool // no longer in the library6468 logged string // the last save error logged6569 now func() time.Time6670 record recordFile6771 }6872 69-type ytResolved struct {70- info ytInfo71- at time.Time72-}73-74-type ytAhead struct {75- started bool // asking yt-dlp, rather than waiting to76- done chan struct{} // closed at the end77-}78-7973 const (8074 ytCatalogFile = ".catalog.json"8175 ytPlayoutFile = ".playout.json"@@ -251,11 +245,16 @@ func (y *YouTube) tend() {
251245 y.busy = true252246 go y.refresh()253247 }248+ y.touched = time.Now()249+ if ytWarm && !y.warming && !y.closed {250+ y.warming = true251+ go y.warm()252+ }254253 }255254 256255 func (y *YouTube) load() {257256 y.loaded = true258- y.tried, y.resolved, y.ahead = map[string]time.Time{}, map[string]ytResolved{}, map[string]*ytAhead{}257+ y.tried, y.resolved, y.ahead, y.missed = map[string]time.Time{}, map[string]ytResolved{}, map[string]*ytAhead{}, map[string]time.Time{}259258 y.feedDown = map[string]string{}260259 cat, err := readYtCatalog(filepath.Join(y.Dir, ytCatalogFile))261260 if err != nil && !errors.Is(err, os.ErrNotExist) {@@ -721,8 +720,7 @@ func (y *YouTube) cue() cueFunc {
721720 }722721 switch {723722 case left <= time.Second && next.ID != "":724- // Done: ffmpeg reads the first half second of input725- // at once, so videos end a little early.723+ // Done, to a frame or so.726724 on, tries = next, 0727725 case left < 15*time.Second || tries >= 2:728726 return item{Slate: true, Frames: max(framesIn(left), 1)}, 0, nil@@ -733,8 +731,10 @@ func (y *YouTube) cue() cueFunc {
733731 } else {734732 tries = 0735733 }734+ // A video opened again is found again: its addresses may be why735+ // it stopped.736736 last, played = on, now737- return y.item(on), max(framesIn(now.Sub(on.Start)), 0), nil737+ return y.item(on, tries > 0), max(framesIn(now.Sub(on.Start)), 0), nil738738 }739739 }740740 @@ -758,11 +758,13 @@ func (y *YouTube) after(a ytAiring) ytAiring {
758758 return next759759 }760760 761-func (y *YouTube) item(a ytAiring) item {761+// item is an airing to play. fresh finds its video's addresses anew762+// rather than using any found before.763+func (y *YouTube) item(a ytAiring, fresh bool) item {762764 return item{763765 Path: a.ID, Frames: framesIn(a.End.Sub(a.Start)), Audio: true, Title: a.Title,764766 Open: func(ctx context.Context, offset time.Duration) (opened, error) {765- return y.open(ctx, a, offset)767+ return y.open(ctx, a, offset, fresh)766768 },767769 Level: y.level(a.ID),768770 }@@ -770,9 +772,10 @@ func (y *YouTube) item(a ytAiring) item {
770772 771773 // open finds where to stream an airing's video from, and its English772774 // captions, and starts finding the next one's shortly before it's due.773-func (y *YouTube) open(ctx context.Context, a ytAiring, offset time.Duration) (opened, error) {775+func (y *YouTube) open(ctx context.Context, a ytAiring, offset time.Duration, fresh bool) (opened, error) {774776 began := time.Now()775- info, err := y.resolve(ctx, a.ID)777+ r, err := y.resolve(ctx, a.ID, fresh)778+ info := r.info776779 y.mu.Lock()777780 if err != nil {778781 if v := y.videos[a.ID]; v != nil && gone(err) {@@ -792,12 +795,8 @@ func (y *YouTube) open(ctx context.Context, a ytAiring, offset time.Duration) (o
792795 go y.prefetch(ctx, next)793796 }794797 in := info.inputs()795- o := opened{AudioLang: info.audioLang()}798+ o := opened{AudioLang: info.audioLang(), Captions: r.caps}796799 o.Args, o.Audio = ytInputArgs(in, offset, httpChunks(y.FFmpeg))797- if sub, ok := info.captions(); ok {798- client := y.tool().client()799- o.Captions = newCaptions(a.ID, func(ctx context.Context) ([]byte, error) { return ytCaptions(ctx, client, sub) })800- }801800 log.Printf("%s: on air %s %s in (format %s, %dp, found in %s)", y.label(), a.ID, offset.Round(time.Second),802801 info.FormatID, in[0].Height, time.Since(began).Round(100*time.Millisecond))803802 return o, nil@@ -858,87 +857,6 @@ func ytGet(ctx context.Context, client *http.Client, address string) ([]byte, er
858857 return io.ReadAll(io.LimitReader(resp.Body, 16<<20))859858 }860859 861-// resolve finds where to stream a video from, using what was found ahead862-// of time, or waiting for it when that's under way. The addresses expire863-// after some hours.864-func (y *YouTube) resolve(ctx context.Context, id string) (ytInfo, error) {865- y.mu.Lock()866- f := y.ahead[id]867- busy := f != nil && f.started868- y.mu.Unlock()869- if busy {870- select {871- case <-f.done:872- case <-ctx.Done():873- return ytInfo{}, ctx.Err()874- }875- }876- y.mu.Lock()877- r, ok := y.resolved[id]878- delete(y.resolved, id)879- height := y.cfg.MaxHeight880- y.mu.Unlock()881- if ok && time.Since(r.at) < time.Hour {882- return r.info, nil883- }884- return y.fetch(ctx, id, height)885-}886-887-// fetch asks yt-dlp where to stream a video from.888-func (y *YouTube) fetch(ctx context.Context, id string, height int) (ytInfo, error) {889- ctx, cancel := context.WithTimeout(ctx, ytResolveTime)890- defer cancel()891- raw, err := y.tool().run(ctx, "-J", "--no-playlist", "-f", ytFormatSpec(height), ytWatchURL(id))892- if err != nil {893- return ytInfo{}, err894- }895- info, err := parseYtInfo(raw)896- if err == nil && len(info.inputs()) == 0 {897- err = errors.New("yt-dlp found no stream")898- }899- return info, err900-}901-902-// prefetch resolves an airing's video a minute before it starts, so the903-// change from one video to the next is quick.904-func (y *YouTube) prefetch(ctx context.Context, a ytAiring) {905- y.mu.Lock()906- if y.ahead[a.ID] != nil {907- y.mu.Unlock()908- return909- }910- f := &ytAhead{done: make(chan struct{})}911- y.ahead[a.ID] = f912- height := y.cfg.MaxHeight913- y.mu.Unlock()914- defer func() {915- y.mu.Lock()916- delete(y.ahead, a.ID)917- y.mu.Unlock()918- close(f.done)919- }()920- select {921- case <-ctx.Done():922- return923- case <-time.After(time.Until(a.Start.Add(-time.Minute))):924- }925- y.mu.Lock()926- f.started = true927- y.mu.Unlock()928- info, err := y.fetch(ctx, a.ID, height)929- if err != nil {930- return // open tries again, and says why931- }932- y.mu.Lock()933- for id, r := range y.resolved {934- if time.Since(r.at) > time.Hour {935- delete(y.resolved, id)936- }937- }938- y.resolved[a.ID] = ytResolved{info, time.Now()}939- y.mu.Unlock()940-}941-942860 // ytFormatSpec picks H.264 picture up to height and AAC sound, English943861 // sound of a video dubbed in several languages, else the best single file944862 // up to height.@@ -946,8 +864,8 @@ func ytFormatSpec(height int) string {
946864 return fmt.Sprintf("bv*[height<=%[1]d][vcodec^=avc1]+ba[ext=m4a][language^=en]/bv*[height<=%[1]d][vcodec^=avc1]+ba[ext=m4a]/b[height<=%[1]d]", height)947865 }948866 949-// ytInputArgs are ffmpeg's inputs for a resolved video, played in real950-// time from offset in: picture first, with yt-dlp's headers, reconnecting867+// ytInputArgs are ffmpeg's inputs for a resolved video, played from868+// offset in: picture first, with yt-dlp's headers, reconnecting951869 // when a connection drops. A stream that fails for good stops ffmpeg952870 // (-xerror) rather than ending the video early, so it can be reopened.953871 // chunked fetches in 10 MB ranges as yt-dlp does, which YouTube serves@@ -970,7 +888,7 @@ func ytInputArgs(fs []ytFormat, offset time.Duration, chunked bool) (args []stri
970888 if offset > 0 {971889 args = append(args, "-ss", strconv.FormatFloat(offset.Seconds(), 'f', 3, 64))972890 }973- args = append(args, "-re", "-i", f.URL)891+ args = append(args, "-i", f.URL)974892 if audio == "" && f.ACodec != "none" && f.ACodec != "" {975893 audio = fmt.Sprintf("%d:a:0", i)976894 }@@ -130,7 +130,7 @@ func TestYtInfoInputs(t *testing.T) {
130130 t.Errorf("audio %q, %d inputs: %q", audio, countInputs(args), args)131131 }132132 joined := strings.Join(args, " ")133- for _, want := range []string{"-ss 90.000 -re -i https://rr1---sn-test.googlevideo.com/videoplayback?itag=298",133+ for _, want := range []string{"-ss 90.000 -i https://rr1---sn-test.googlevideo.com/videoplayback?itag=298",134134 "-request_size 10485760", "User-Agent: Mozilla/5.0", "\r\n"} {135135 if !strings.Contains(joined, want) {136136 t.Errorf("args lack %q:\n%s", want, joined)@@ -469,8 +469,9 @@ func TestYouTubeGuide(t *testing.T) {
469469 // TestOpenFailureKeepsTheSchedule plays three two-second items whose470470 // sources open per play: the first fails to open after a while, the second471471 // opens separate picture and sound inputs, the third fails at once. Slates472-// fill the failures' time, so the second starts on schedule, and the473-// stream is a single H.264 and AAC stream as long as the time it ran.472+// fill the failures' time, so each item is cued on schedule by the473+// stream's clock, and the stream is a single H.264 and AAC stream as long474+// as the time it ran and its lead.474475 func TestOpenFailureKeepsTheSchedule(t *testing.T) {475476 ffmpeg, err := exec.LookPath("ffmpeg")476477 if err != nil {@@ -491,26 +492,26 @@ func TestOpenFailureKeepsTheSchedule(t *testing.T) {
491492 t.Fatalf("make %s: %v: %s", out, err, b)492493 }493494 }494- var openedAt time.Duration495- var start time.Time495+ var cuedAt []time.Duration // by the stream's clock496496 items := []item{497497 {Path: "fails-slowly", Frames: 2 * loopFPS, Open: func(ctx context.Context, _ time.Duration) (opened, error) {498498 time.Sleep(400 * time.Millisecond)499499 return opened{}, errors.New("yt-dlp: Sign in to confirm you're not a bot")500500 }},501501 {Path: "plays", Frames: 2 * loopFPS, Open: func(ctx context.Context, _ time.Duration) (opened, error) {502- openedAt = time.Since(start)503- return opened{Args: []string{"-re", "-i", picture, "-re", "-i", sound}, Audio: "1:a:0"}, nil502+ return opened{Args: []string{"-i", picture, "-i", sound}, Audio: "1:a:0"}, nil504503 }},505504 {Path: "fails", Frames: 2 * loopFPS, Open: func(ctx context.Context, _ time.Duration) (opened, error) {506505 return opened{}, errors.New("no stream")507506 }},508507 }509508 n := 0509+ var start time.Time510510 cue := func(now time.Time, _ bool) (item, int64, error) {511511 if n == 0 {512512 start = now513513 }514+ cuedAt = append(cuedAt, now.Sub(start))514515 if n == len(items) {515516 return item{Slate: true, Frames: 10 * loopFPS}, 0, nil516517 }@@ -523,10 +524,15 @@ func TestOpenFailureKeepsTheSchedule(t *testing.T) {
523524 if err := playCues(ctx, &out, ffmpeg, "test", cue); err != nil {524525 t.Fatal(err)525526 }526- // ffmpeg reads the first half second of each input at once (-re's527- // initial burst), so items end up to that much early.528- if openedAt < 1400*time.Millisecond || openedAt > 2400*time.Millisecond {529- t.Errorf("the second item opened %v in, want two seconds", openedAt)527+ // The time the first spent failing to open is the stream's lead's: it528+ // played nothing.529+ for i, want := range []time.Duration{0, 2 * time.Second, 4 * time.Second, 6 * time.Second} {530+ if i >= len(cuedAt) {531+ t.Fatalf("cued %d times, want %d", len(cuedAt), len(items)+1)532+ }533+ if d := cuedAt[i] - want; d < -time.Second/loopFPS || d > time.Second/loopFPS {534+ t.Errorf("item %d cued %v in, want %v", i, cuedAt[i], want)535+ }530536 }531537 ts := filepath.Join(dir, "out.ts")532538 if err := os.WriteFile(ts, out.Bytes(), 0o644); err != nil {@@ -539,9 +545,10 @@ func TestOpenFailureKeepsTheSchedule(t *testing.T) {
539545 got := string(probe)540546 if i := strings.Index(got, "duration="); i < 0 {541547 t.Errorf("no duration:\n%s", got)542- } else if d, _ := strconv.ParseFloat(strings.TrimSpace(got[i+len("duration="):]), 64); d < 4 || d > 6 {543- // Less the time spent opening, while nothing plays.544- t.Errorf("played %.2fs in 5.5s", d)548+ } else if d, _ := strconv.ParseFloat(strings.TrimSpace(got[i+len("duration="):]), 64); d < 5.5+lead.Seconds()-2.5 || d > 5.5+lead.Seconds()+0.5 {549+ // Its lead ahead of the clock, less the time it took to start and550+ // what the encoder held when it stopped.551+ t.Errorf("played %.2fs in 5.5s with a %v lead", d, lead)545552 }546553 for _, want := range []string{"codec_name=h264|width=1280|height=720", "codec_name=aac"} {547554 if !strings.Contains(got, want) {@@ -0,0 +1,274 @@
1+package vchan2+3+import (4+ "context"5+ "errors"6+ "log"7+ "net/url"8+ "strconv"9+ "time"10+11+ "airwaves/internal/phase"12+)13+14+// Finding where to stream a video from takes yt-dlp seconds, too long to15+// wait for when tuning in. So each YouTube channel keeps the addresses of16+// the video on now and the next found ahead of time (see warm), and finds17+// them again before they expire. That's one yt-dlp run per video as it18+// comes up, and one more for a video longer than its addresses last.19+20+// ytResolved is a video's details with its addresses, found ahead.21+type ytResolved struct {22+ info ytInfo23+ // caps are its English captions, fetched as soon as it's found, so24+ // tuning in doesn't wait for them either; nil when it has none.25+ caps *captions26+ height int // the most picture asked for27+ until time.Time // when its addresses may stop working28+}29+30+type ytAhead struct {31+ started bool // asking yt-dlp, rather than waiting to32+ done chan struct{} // closed at the end33+}34+35+const (36+ // ytWarmIdle is how long a channel nobody lists, guides or watches37+ // keeps finding its videos ahead.38+ ytWarmIdle = 6 * time.Hour39+ // ytWarmEvery is how often it looks at what to find.40+ ytWarmEvery = 30 * time.Second41+ // ytMinLeft is how long addresses must still work to be used, or42+ // kept: they're found again sooner.43+ ytMinLeft = 20 * time.Minute44+ // ytMissedRetry spaces out tries to find a video ahead that failed.45+ ytMissedRetry = 10 * time.Minute46+)47+48+// ytWarm turns finding videos ahead on; tests turn it off.49+var ytWarm = true50+51+// warm keeps the addresses of the videos on now and next found, so52+// tuning in never waits for yt-dlp. It runs while the channel is tended53+// (listed, guided or watched) and stops ytWarmIdle after it last was, or54+// once the channel leaves the library.55+func (y *YouTube) warm() {56+ for {57+ y.mu.Lock()58+ if y.closed || time.Since(y.touched) > ytWarmIdle {59+ y.warming = false60+ y.mu.Unlock()61+ return62+ }63+ ids := y.warmDue(time.Now())64+ y.mu.Unlock()65+ for _, id := range ids {66+ y.lookAhead(id)67+ }68+ time.Sleep(ytWarmEvery)69+ }70+}71+72+// warmDue lists the videos on and next whose addresses need finding, and73+// forgets addresses that expired. y.mu is held.74+func (y *YouTube) warmDue(now time.Time) []string {75+ for id, r := range y.resolved {76+ if now.After(r.until) {77+ delete(y.resolved, id)78+ }79+ }80+ if !y.ready() || y.err != nil {81+ return nil82+ }83+ on, next, _, ok := y.airingAt(y.clock())84+ if !ok {85+ return nil86+ }87+ if next.ID == "" {88+ next = y.after(on)89+ }90+ var ids []string91+ for _, a := range []ytAiring{on, next} {92+ if a.ID == "" || y.ahead[a.ID] != nil || now.Sub(y.missed[a.ID]) < ytMissedRetry {93+ continue94+ }95+ if v := y.videos[a.ID]; v != nil && v.Skip == "gone" {96+ continue97+ }98+ if r, ok := y.resolved[a.ID]; ok && r.height == y.cfg.MaxHeight && r.until.Sub(now) >= ytMinLeft {99+ continue100+ }101+ if len(ids) == 0 || ids[0] != a.ID {102+ ids = append(ids, a.ID)103+ }104+ }105+ return ids106+}107+108+// lookAhead finds a video's addresses now, one video at a time across109+// channels, so a tune for it waits for this rather than asking yt-dlp too.110+func (y *YouTube) lookAhead(id string) {111+ y.mu.Lock()112+ if y.ahead[id] != nil {113+ y.mu.Unlock()114+ return115+ }116+ f := &ytAhead{done: make(chan struct{})}117+ y.ahead[id] = f118+ height := y.cfg.MaxHeight119+ y.mu.Unlock()120+ tool := y.tool()121+ tool.ahead.Lock()122+ y.mu.Lock()123+ f.started = true124+ y.mu.Unlock()125+ ctx, cancel := context.WithTimeout(context.Background(), ytResolveTime)126+ info, err := y.fetch(ctx, id, height)127+ cancel()128+ tool.ahead.Unlock()129+ y.mu.Lock()130+ delete(y.ahead, id)131+ switch {132+ case err == nil:133+ delete(y.missed, id)134+ y.keep(id, info, height)135+ case gone(err):136+ if v := y.videos[id]; v != nil {137+ v.Skip, v.New, v.Looked = "gone", false, y.clock()138+ y.changed()139+ }140+ log.Printf("%s: %s can't play: %v", y.label(), id, err)141+ default:142+ y.missed[id] = time.Now()143+ log.Printf("%s: finding %s ahead: %v", y.label(), id, err)144+ }145+ y.mu.Unlock()146+ close(f.done)147+}148+149+// keep remembers a video's addresses until they expire, and starts150+// fetching its captions. y.mu is held.151+func (y *YouTube) keep(id string, info ytInfo, height int) ytResolved {152+ r := ytResolved{info: info, height: height, until: addressesUntil(info, time.Now())}153+ if old, ok := y.resolved[id]; ok && old.caps != nil && !old.caps.failed() {154+ r.caps = old.caps // the same captions, fetched already or under way155+ } else if sub, ok := info.captions(); ok {156+ client := y.tool().client()157+ r.caps = newCaptions(id, func(ctx context.Context) ([]byte, error) { return ytCaptions(ctx, client, sub) })158+ r.caps.start()159+ }160+ y.resolved[id] = r161+ return r162+}163+164+// addressesUntil is when a video's addresses, found at, may stop working:165+// ten minutes before the earliest "expire" they carry (YouTube's last six166+// hours), or an hour after they were found when they carry none.167+func addressesUntil(info ytInfo, at time.Time) time.Time {168+ until := at.Add(time.Hour)169+ var earliest time.Time170+ for _, f := range info.inputs() {171+ u, err := url.Parse(f.URL)172+ if err != nil {173+ return until174+ }175+ sec, err := strconv.ParseInt(u.Query().Get("expire"), 10, 64)176+ if err != nil || sec <= 0 {177+ return until178+ }179+ if t := time.Unix(sec, 0); earliest.IsZero() || t.Before(earliest) {180+ earliest = t181+ }182+ }183+ if earliest.IsZero() {184+ return until185+ }186+ if latest := at.Add(6 * time.Hour); earliest.After(latest) {187+ return latest188+ }189+ return earliest.Add(-10 * time.Minute)190+}191+192+// resolve finds where to stream a video from, and its captions: what was193+// found ahead of time, waiting for it when that's under way, else yt-dlp194+// now. fresh skips what was found before. What's found is kept for the195+// next tune.196+func (y *YouTube) resolve(ctx context.Context, id string, fresh bool) (ytResolved, error) {197+ y.mu.Lock()198+ f := y.ahead[id]199+ busy := f != nil && f.started200+ y.mu.Unlock()201+ if busy && !fresh {202+ select {203+ case <-f.done:204+ case <-ctx.Done():205+ return ytResolved{}, ctx.Err()206+ }207+ phase.Step(ctx, "waited for yt-dlp")208+ }209+ y.mu.Lock()210+ r, ok := y.resolved[id]211+ height := y.cfg.MaxHeight212+ if ok && !fresh && r.height == height && time.Until(r.until) >= ytMinLeft {213+ if r.caps != nil && r.caps.failed() {214+ r = y.keep(id, r.info, height) // the captions, tried again215+ }216+ y.mu.Unlock()217+ phase.Step(ctx, "found ahead")218+ return r, nil219+ }220+ if fresh {221+ delete(y.resolved, id)222+ }223+ y.mu.Unlock()224+ info, err := y.fetch(ctx, id, height)225+ phase.Step(ctx, "yt-dlp")226+ if err != nil {227+ return ytResolved{}, err228+ }229+ y.mu.Lock()230+ r = y.keep(id, info, height)231+ y.mu.Unlock()232+ return r, nil233+}234+235+// fetch asks yt-dlp where to stream a video from.236+func (y *YouTube) fetch(ctx context.Context, id string, height int) (ytInfo, error) {237+ ctx, cancel := context.WithTimeout(ctx, ytResolveTime)238+ defer cancel()239+ raw, err := y.tool().run(ctx, "-J", "--no-playlist", "-f", ytFormatSpec(height), ytWatchURL(id))240+ if err != nil {241+ return ytInfo{}, err242+ }243+ info, err := parseYtInfo(raw)244+ if err == nil && len(info.inputs()) == 0 {245+ err = errors.New("yt-dlp found no stream")246+ }247+ return info, err248+}249+250+// prefetch finds an airing's video a minute before it starts, unless it251+// has been already, so the change from one video to the next is quick252+// even while the channel isn't finding videos ahead.253+func (y *YouTube) prefetch(ctx context.Context, a ytAiring) {254+ select {255+ case <-ctx.Done():256+ return257+ case <-time.After(time.Until(a.Start.Add(-time.Minute))):258+ }259+ y.mu.Lock()260+ r, ok := y.resolved[a.ID]261+ found := ok && r.height == y.cfg.MaxHeight && time.Until(r.until) >= ytMinLeft262+ y.mu.Unlock()263+ if !found {264+ y.lookAhead(a.ID)265+ }266+}267+268+// close stops the channel finding videos ahead, once it's left the269+// library.270+func (y *YouTube) close() {271+ y.mu.Lock()272+ y.closed = true273+ y.mu.Unlock()274+}@@ -0,0 +1,131 @@
1+package vchan2+3+import (4+ "context"5+ "net/http"6+ "os"7+ "path/filepath"8+ "strconv"9+ "strings"10+ "testing"11+ "time"12+13+ "airwaves/internal/phase"14+)15+16+// TestYouTubeFindsAhead: the videos on now and next are found ahead,17+// once each, with their captions, so tuning in uses what was found18+// without yt-dlp; addresses near their expiry are found again, and a19+// video opened again is found anew.20+func TestYouTubeFindsAhead(t *testing.T) {21+ calls := filepath.Join(t.TempDir(), "yt-dlp.log")22+ t.Setenv("AIRWAVES_FAKE_YTDLP", calls)23+ runs := func() int {24+ raw, _ := os.ReadFile(calls)25+ return strings.Count(string(raw), " -f ")26+ }27+ t0 := time.Now().UTC().Truncate(time.Second)28+ stub := &feedStub{down: true}29+ y := &YouTube{Num: "1.8", Dir: t.TempDir(), loaded: true, now: func() time.Time { return t0 },30+ YtDlp: &YtDlp{Path: os.Args[0], HTTP: &http.Client{Transport: stub}},31+ cfg: ytConfig{MaxHeight: 720},32+ pool: &ytPool{Basis: "test", Lock: time.Hour}, plan: ytPlayout{Basis: "test", Airings: []ytAiring{33+ {ID: "aaaaaaaaaaa", Channel: "UCa", Title: "A", Start: t0.Add(-time.Minute), End: t0.Add(10 * time.Minute)},34+ {ID: "bbbbbbbbbbb", Channel: "UCb", Title: "B", Start: t0.Add(10 * time.Minute), End: t0.Add(30 * time.Minute)},35+ {ID: "ccccccccccc", Channel: "UCa", Title: "C", Start: t0.Add(30 * time.Minute), End: t0.Add(40 * time.Minute)},36+ }},37+ resolved: map[string]ytResolved{}, ahead: map[string]*ytAhead{}, missed: map[string]time.Time{},38+ }39+ due := func() []string {40+ y.mu.Lock()41+ defer y.mu.Unlock()42+ return y.warmDue(time.Now())43+ }44+45+ ids := due()46+ if strings.Join(ids, " ") != "aaaaaaaaaaa bbbbbbbbbbb" {47+ t.Fatalf("due %q, want what's on and next", ids)48+ }49+ for _, id := range ids {50+ y.lookAhead(id)51+ }52+ if n := runs(); n != 2 {53+ t.Fatalf("%d yt-dlp runs, want 2", n)54+ }55+ if ids := due(); len(ids) != 0 {56+ t.Errorf("due %q once found", ids)57+ }58+59+ // Tuning in uses what was found.60+ timer := phase.New("tune", "youtube")61+ r, err := y.resolve(phase.NewContext(context.Background(), timer), "aaaaaaaaaaa", false)62+ if err != nil || len(r.info.inputs()) == 0 {63+ t.Fatalf("resolve: %v %+v", err, r)64+ }65+ if n := runs(); n != 2 || !timer.Has("found ahead") {66+ t.Errorf("tuning in ran yt-dlp (%d runs; steps %s)", n, timer)67+ }68+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)69+ defer cancel()70+ if r.caps == nil {71+ t.Fatal("no captions")72+ } else if cues, err := r.caps.get(ctx); err != nil || len(cues) != 1 {73+ t.Errorf("captions: %v %+v", err, cues)74+ }75+ // And again: what's found is kept for the next tune.76+ if again, err := y.resolve(context.Background(), "aaaaaaaaaaa", false); err != nil || runs() != 2 || again.caps != r.caps {77+ t.Errorf("a second tune ran yt-dlp, or fetched the captions again: %v, %d runs", err, runs())78+ }79+ y.mu.Lock()80+ bcaps := y.resolved["bbbbbbbbbbb"].caps81+ y.mu.Unlock()82+ if bcaps == nil {83+ t.Fatal("no captions for the next video")84+ }85+ _, _ = bcaps.get(ctx)86+ stub.mu.Lock()87+ if stub.captions != 2 {88+ t.Errorf("fetched captions %d times, want once for each video", stub.captions)89+ }90+ stub.mu.Unlock()91+92+ // Addresses about to expire are found again.93+ y.mu.Lock()94+ b := y.resolved["bbbbbbbbbbb"]95+ b.until = time.Now().Add(ytMinLeft / 2)96+ y.resolved["bbbbbbbbbbb"] = b97+ y.mu.Unlock()98+ if ids := due(); strings.Join(ids, " ") != "bbbbbbbbbbb" {99+ t.Errorf("due %q with B's addresses expiring", ids)100+ }101+102+ // A video opened again is found anew.103+ if _, err := y.resolve(context.Background(), "aaaaaaaaaaa", true); err != nil || runs() != 3 {104+ t.Errorf("a fresh resolve: %v, %d runs", err, runs())105+ }106+107+ // Once the next video is on, the one after it is found.108+ t0 = t0.Add(11 * time.Minute)109+ if ids := due(); strings.Join(ids, " ") != "bbbbbbbbbbb ccccccccccc" {110+ t.Errorf("due %q after A, want B (expiring) and C", ids)111+ }112+}113+114+func TestAddressesUntil(t *testing.T) {115+ at := time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC)116+ exp := func(d time.Duration) string {117+ return "https://rr1.example/videoplayback?expire=" + strconv.FormatInt(at.Add(d).Unix(), 10) + "&sig=x"118+ }119+ two := ytInfo{RequestedFormats: []ytFormat{{URL: exp(6 * time.Hour), VCodec: "avc1"}, {URL: exp(5 * time.Hour), VCodec: "none", ACodec: "mp4a"}}}120+ if got, want := addressesUntil(two, at), at.Add(5*time.Hour-10*time.Minute); !got.Equal(want) {121+ t.Errorf("two addresses: until %v, want %v", got, want)122+ }123+ none := ytInfo{ytFormat: ytFormat{URL: "https://example.com/v.mp4", VCodec: "avc1", ACodec: "mp4a"}}124+ if got, want := addressesUntil(none, at), at.Add(time.Hour); !got.Equal(want) {125+ t.Errorf("no expiry: until %v, want %v", got, want)126+ }127+ far := ytInfo{ytFormat: ytFormat{URL: exp(48 * time.Hour), VCodec: "avc1", ACodec: "mp4a"}}128+ if got, want := addressesUntil(far, at), at.Add(6*time.Hour); !got.Equal(want) {129+ t.Errorf("far expiry: until %v, want %v", got, want)130+ }131+}@@ -22,13 +22,18 @@ func TestMain(m *testing.M) {
2222 if os.Getenv("AIRWAVES_FAKE_YTDLP") != "" {2323 os.Exit(fakeYtDlp(os.Args[1:]))2424 }25+ // Channels find videos ahead only in the tests of that, so the rest26+ // see only the yt-dlp runs they cause.27+ ytWarm = false2528 os.Exit(m.Run())2629 }2730 2831 // fakeYtDlp answers yt-dlp's command line, noting each run's arguments in2932 // the file AIRWAVES_FAKE_YTDLP names. A channel tab's newest uploads are a3033 // new 15-minute upload, "fresh000001", then the channel's v000 to v013;31-// any video's details give it 15 minutes, but a "Premiere…" is upcoming.34+// any video's details give it 15 minutes, and with a format asked for, an35+// address that expires in six hours and English captions, but a36+// "Premiere…" is upcoming.3237 // A playlist is testdata/yt_playlist.json, with AIRWAVES_FAKE_PLAYLIST's3338 // videos added to the end when set.3439 func fakeYtDlp(args []string) int {@@ -57,6 +62,11 @@ func fakeYtDlp(args []string) int {
5762 "description": "Details of " + id, "live_status": "not_live", "media_type": "video"}5863 if strings.HasPrefix(id, "Premiere") {5964 out = map[string]any{"id": id, "title": "Looked up " + id, "live_status": "is_upcoming", "media_type": "video"}65+ } else if slices.Contains(args, "-f") {66+ // Where to stream it from, for six hours, and its captions.67+ out.(map[string]any)["url"] = fmt.Sprintf("https://example.com/%s.mp4?expire=%d", id, now+6*3600)68+ out.(map[string]any)["vcodec"], out.(map[string]any)["acodec"] = "avc1.4d401f", "mp4a.40.2"69+ out.(map[string]any)["automatic_captions"] = map[string]any{"en": []map[string]any{{"ext": "json3", "url": "https://example.com/captions/" + id}}}6070 }6171 case strings.Contains(u, "playlist?list="):6272 var p map[string]any@@ -77,16 +87,23 @@ func fakeYtDlp(args []string) int {
7787 return 07888 }7989 80-// feedStub serves a channel's feed, or fails with HTTP 404 while down.90+// feedStub serves a channel's feed, or fails with HTTP 404 while down,91+// and captions, counting them.8192 type feedStub struct {82- mu sync.Mutex83- down bool84- feed []byte93+ mu sync.Mutex94+ down bool95+ feed []byte96+ captions int8597 }8698 8799 func (f *feedStub) RoundTrip(r *http.Request) (*http.Response, error) {88100 f.mu.Lock()89101 defer f.mu.Unlock()102+ if strings.HasPrefix(r.URL.String(), "https://example.com/captions/") {103+ f.captions++104+ caps := `{"events":[{"tStartMs":1000,"dDurationMs":2000,"segs":[{"utf8":"Hello"}]}]}`105+ return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(caps)), Request: r}, nil106+ }90107 if f.down || !strings.HasPrefix(r.URL.String(), "https://www.youtube.com/feeds/videos.xml?channel_id=UCa") {91108 return &http.Response{StatusCode: http.StatusNotFound, Body: io.NopCloser(strings.NewReader("")), Request: r}, nil92109 }@@ -0,0 +1,224 @@
1+// tunebench measures how long channel changes take on a running airwavesd,2+// as the app sees them: the tune call, then playback in headless Chrome3+// with hls.js set up as the app sets it up, to the first frame and through4+// the first seconds, counting stalls. Tunes run one at a time, as their own5+// client, and each is stopped before the next.6+//7+// node scripts/tunebench.mjs --server http://nas:8089 --plan 9.1x3,1.2x3,201.0x28+//9+// Options:10+// --server URL the airwavesd to measure (required)11+// --plan LIST channel numbers, each with "xN" for N tunes; tunes go12+// round by round through the list, so a channel's tunes13+// are spread out14+// --watch S seconds to watch each tune after it plays (30)15+// --gap S seconds between tunes (5)16+// --client ID the client ID to tune as ("bench")17+// --token T the API token, when the server has one18+// --out FILE append each tune's result to FILE as JSON lines19+// --label NAME names the run in the results ("run")20+// --chrome PATH a Chrome or Chromium that plays H.264 (default: the21+// installed Google Chrome)22+//23+// Playwright is loaded from PLAYWRIGHT (a path to its index.mjs) or the24+// "playwright" package.25+import fs from 'node:fs';26+import path from 'node:path';27+import { fileURLToPath } from 'node:url';28+29+const args = parseArgs(process.argv.slice(2));30+if (!args.server || !args.plan) {31+ console.error('usage: node scripts/tunebench.mjs --server URL --plan 9.1x3,1.2x3 [--watch 30] [--gap 5] [--out FILE]');32+ process.exit(2);33+}34+const server = args.server.replace(/\/+$/, '');35+const watch = Number(args.watch ?? 30);36+const gap = Number(args.gap ?? 5);37+const client = args.client || 'bench';38+const label = args.label || 'run';39+const headers = { 'Content-Type': 'application/json', ...(args.token ? { Authorization: `Bearer ${args.token}` } : {}) };40+41+const here = path.dirname(fileURLToPath(import.meta.url));42+const hlsjs = path.join(here, '..', 'frontend', 'dist', 'vendor', 'hls.min.js');43+const { chromium } = await import(process.env.PLAYWRIGHT || 'playwright');44+45+function parseArgs(argv) {46+ const out = {};47+ for (let i = 0; i < argv.length; i++) {48+ const a = argv[i];49+ if (!a.startsWith('--')) continue;50+ const [k, v] = a.slice(2).split('=', 2);51+ out[k] = v ?? argv[++i];52+ }53+ return out;54+}55+56+// plan expands "9.1x3,1.2x2" into tunes, round by round.57+function plan(spec) {58+ const want = spec.split(',').map((s) => s.trim()).filter(Boolean).map((s) => {59+ const [number, n] = s.split('x');60+ return { number, n: Number(n || 1) };61+ });62+ const out = [];63+ for (let round = 0; want.some((w) => w.n > round); round++) {64+ for (const w of want) if (w.n > round) out.push(w.number);65+ }66+ return out;67+}68+69+async function api(method, p, body) {70+ const r = await fetch(server + p, { method, headers, body: body ? JSON.stringify(body) : undefined });71+ const text = await r.text();72+ let json;73+ try { json = JSON.parse(text); } catch { json = { error: text }; }74+ if (!r.ok) throw new Error(`${p}: HTTP ${r.status}: ${json.error || text}`.slice(0, 300));75+ return json;76+}77+78+async function kinds() {79+ const snap = await api('GET', '/api/snapshot');80+ const m = new Map();81+ for (const c of snap.custom || []) m.set(c.number, c.kind || 'folder');82+ return m;83+}84+85+const sleep = (ms) => new Promise((r) => setTimeout(r, ms));86+const quantile = (xs, q) => {87+ const s = xs.filter((x) => x != null && Number.isFinite(x)).sort((a, b) => a - b);88+ if (!s.length) return null;89+ return s[Math.min(s.length - 1, Math.max(0, Math.ceil(q * s.length) - 1))];90+};91+const secs = (ms) => (ms == null ? '-' : (ms / 1000).toFixed(2));92+93+// The page plays a stream as the app does (frontend/dist/app.js attempt()),94+// and keeps a record of what happened when.95+const pageScript = () => {96+ window.bench = (src, watchMs) => new Promise((resolve) => {97+ const video = document.querySelector('video');98+ const t0 = Date.now();99+ const rec = { start: t0, events: {}, stalls: [], errors: [], buffers: [] };100+ const mark = (name) => { if (rec.events[name] == null) rec.events[name] = Date.now() - t0; };101+ const h = new window.Hls({ liveSyncDurationCount: 2, maxBufferLength: 12, subtitleDisplay: false });102+ const E = window.Hls.Events;103+ h.on(E.MANIFEST_PARSED, (_, d) => { mark('manifest'); rec.audioTracks = (d.audioTracks || []).length; rec.subtitleTracks = (d.subtitleTracks || []).length; });104+ h.on(E.LEVEL_LOADED, (_, d) => { mark('level'); if (rec.firstLevel == null) rec.firstLevel = { segments: d.details.fragments.length, targetDuration: d.details.targetduration, first: d.details.fragments.map((f) => +f.duration.toFixed(2)).slice(0, 4) }; });105+ h.on(E.FRAG_LOADED, (_, d) => { if (d.frag.type === 'main') mark('fragLoaded'); });106+ h.on(E.FRAG_BUFFERED, (_, d) => { if (d.frag.type === 'main') mark('fragBuffered'); });107+ h.on(E.BUFFER_CREATED, (_, d) => { rec.buffers = Object.keys(d.tracks || {}); });108+ h.on(E.ERROR, (_, d) => { rec.errors.push(`${d.fatal ? 'fatal ' : ''}${d.details}`); });109+ let playing = false;110+ let stallAt = 0;111+ video.addEventListener('playing', () => {112+ if (!playing) { playing = true; mark('playing'); rec.startPos = video.currentTime; rec.liveEdge = h.liveSyncPosition; }113+ if (stallAt) { rec.stalls[rec.stalls.length - 1].ms = Date.now() - stallAt; stallAt = 0; }114+ });115+ video.addEventListener('waiting', () => {116+ if (!playing || stallAt) return;117+ stallAt = Date.now();118+ rec.stalls.push({ at: stallAt - t0, ms: null });119+ });120+ // A picture that stops without a waiting event counts too.121+ let lastPos = -1;122+ let still = 0;123+ const tick = setInterval(() => {124+ if (!playing || video.paused) return;125+ if (video.currentTime === lastPos && !stallAt) {126+ if (++still === 2) rec.stalls.push({ at: Date.now() - t0 - 1000, ms: null, frozen: true });127+ } else still = 0;128+ lastPos = video.currentTime;129+ }, 500);130+ video.muted = true;131+ h.loadSource(src);132+ h.attachMedia(video);133+ video.play().catch(() => {});134+ const giveUp = setTimeout(() => finish(), watchMs + 20000);135+ const finish = () => {136+ clearInterval(tick);137+ clearTimeout(giveUp);138+ if (stallAt) rec.stalls[rec.stalls.length - 1].ms = Date.now() - stallAt;139+ const q = video.getVideoPlaybackQuality ? video.getVideoPlaybackQuality() : {};140+ rec.dropped = q.droppedVideoFrames;141+ rec.frames = q.totalVideoFrames;142+ rec.width = video.videoWidth;143+ rec.latency = h.latency;144+ rec.played = video.currentTime - (rec.startPos || 0);145+ h.destroy();146+ resolve(rec);147+ };148+ // Watch for watchMs after playing starts, or give up 20 s after that.149+ const wait = setInterval(() => {150+ if (rec.events.playing != null && Date.now() - t0 - rec.events.playing >= watchMs) { clearInterval(wait); finish(); }151+ }, 200);152+ });153+};154+155+const browser = await chromium.launch({156+ ...(args.chrome ? { executablePath: args.chrome } : { channel: 'chrome' }),157+ args: ['--autoplay-policy=no-user-gesture-required', '--mute-audio'],158+});159+const page = await browser.newPage();160+await page.setContent('<!doctype html><video width="640" height="360"></video>');161+await page.addScriptTag({ path: hlsjs });162+await page.evaluate(pageScript);163+164+const kindOf = await kinds();165+const tunes = plan(args.plan);166+const results = [];167+console.log(`${tunes.length} tunes on ${server}, ${watch} s each, ${gap} s apart`);168+for (const [i, number] of tunes.entries()) {169+ if (i > 0) await sleep(gap * 1000);170+ const kind = kindOf.get(number) || 'antenna';171+ const r = { label, number, kind, at: new Date().toISOString() };172+ const t0 = Date.now();173+ try {174+ const pb = await api('POST', '/api/tune', { client, number });175+ r.api = Date.now() - t0;176+ r.timing = pb.timing;177+ const rec = await page.evaluate(([src, ms]) => window.bench(src, ms), [server + pb.path, watch * 1000]);178+ const off = rec.start - t0; // tune call to the page's start179+ r.page = rec;180+ r.ttff = rec.events.playing != null ? off + rec.events.playing : null;181+ if (r.ttff == null) r.error = `never played${rec.errors.length ? ': ' + rec.errors.join(' ') : ''}`;182+ r.stalls = rec.stalls.length;183+ r.stallMs = rec.stalls.reduce((s, x) => s + (x.ms || 0), 0);184+ } catch (e) {185+ r.error = String(e.message || e);186+ }187+ try { await api('POST', '/api/stop', { client }); } catch (e) { r.stopError = String(e.message || e); }188+ results.push(r);189+ if (args.out) fs.appendFileSync(args.out, JSON.stringify(r) + '\n');190+ const ev = r.page ? r.page.events : {};191+ const t = r.timing ? ' | server: ' + r.timing.map((m) => `${m.name} ${secs(m.ms)}`).join(', ') : '';192+ console.log(`${String(i + 1).padStart(2)} ${number.padEnd(6)} ${kind.padEnd(8)} api ${secs(r.api)} s, first frame ${secs(r.ttff)} s` +193+ ` (manifest +${secs(ev.manifest)}, fragment +${secs(ev.fragLoaded)}, playing +${secs(ev.playing)}),` +194+ ` stalls ${r.stalls ?? '-'}${r.stallMs ? ` (${secs(r.stallMs)} s)` : ''}${r.page && r.page.errors.length ? ' errors ' + r.page.errors.join(' ') : ''}${r.error ? ' ERROR ' + r.error : ''}${t}`);195+}196+await browser.close();197+198+// The summary: per kind, the median and worst of each measure.199+const byKind = new Map();200+for (const r of results) {201+ if (!byKind.has(r.kind)) byKind.set(r.kind, []);202+ byKind.get(r.kind).push(r);203+}204+console.log('\nkind tunes api p50 api max frame p50 frame max stalls failed');205+for (const [kind, rs] of byKind) {206+ const ok = rs.filter((r) => !r.error);207+ console.log(`${kind.padEnd(9)} ${String(rs.length).padStart(5)} ${secs(quantile(ok.map((r) => r.api), 0.5)).padStart(7)} ${secs(quantile(ok.map((r) => r.api), 1)).padStart(7)}` +208+ ` ${secs(quantile(ok.map((r) => r.ttff), 0.5)).padStart(9)} ${secs(quantile(ok.map((r) => r.ttff), 1)).padStart(9)}` +209+ ` ${String(ok.reduce((s, r) => s + (r.stalls || 0), 0)).padStart(6)} ${String(rs.length - ok.length).padStart(6)}`);210+}211+212+// The server's steps, when it reports them: per kind, each step's median.213+const steps = [...byKind].map(([kind, rs]) => {214+ const at = new Map();215+ for (const r of rs) for (const m of r.timing || []) {216+ if (!at.has(m.name)) at.set(m.name, []);217+ at.get(m.name).push(m.ms);218+ }219+ return [kind, [...at].map(([name, ms]) => `${name} ${secs(quantile(ms, 0.5))}`).join(', ')];220+}).filter(([, s]) => s);221+if (steps.length) {222+ console.log('\nserver steps, median seconds from the tune call');223+ for (const [kind, s] of steps) console.log(`${kind.padEnd(9)} ${s}`);224+}