feat: add playback speed controls and scale buffering

README.md

@@ -15,6 +15,7 @@ Restream brings your IPTV subscription to the browser. Host it on your own compu
 - ⚡ **Shared library cache**: cached listings stay available while the server refreshes them in the background, reducing repeated requests to your provider.
 - 🔐 **Optional OIDC sign-in**: use your identity provider with viewer and administrator roles, or run without a login on your private network.
 - 📱 **Desktop and mobile**: responsive browsing, touch controls, and fullscreen playback.
+- ⏩ **Playback speed**: watch movies and episodes at 0.5× to 2× speed from the player controls. The buffer adjusts to your speed; changing speed briefly reloads at your current position.
 - 💬 **Subtitles**: choose provider subtitles or embedded captions from the player's CC menu. Preview your preferred font, size, color, and background in Settings, saved for this browser. Availability varies by movie and episode.
 
 ## Get started

internal/app/auth_playback_integration_test.go

@@ -64,6 +64,7 @@ func TestOIDCPlaybackBelongsToItsUser(t *testing.T) {
 		{"GET", "/api/sessions/test-1", ""},
 		{"POST", "/api/sessions/test-1/heartbeat", ""},
 		{"POST", "/api/sessions/test-1/seek", `{"position":5}`},
+		{"POST", "/api/sessions/test-1/speed", `{"position":5,"rate":2}`},
 		{"DELETE", "/api/sessions/test-1", ""},
 		{"GET", "/api/streams/test-1/1/index.m3u8", ""},
 		{"GET", "/api/streams/test-1/1/seg-000001.ts", ""},

internal/app/playback_integration_test.go

@@ -172,16 +172,32 @@ func TestPortalToBrowserPlayback(t *testing.T) {
 		t.Fatalf("duration not probed: %+v", session)
 	}
 	assertLocalMedia(t, server.URL, session.URL, portal.URL)
+	previous := session.URL
+	for _, invalid := range []string{`{"position":5,"rate":0}`, `{"position":-1,"rate":2}`, `{"position":5,"rate":3}`, `{"position":5}`, `{"position":5,"rate":null}`} {
+		status, body = request(t, server.URL, "POST", "/api/sessions/"+session.ID+"/speed", invalid)
+		if status != 400 {
+			t.Fatalf("invalid playback speed accepted: %d %s", status, body)
+		}
+	}
+	status, body = request(t, server.URL, "POST", "/api/sessions/"+session.ID+"/speed", `{"position":5,"rate":2}`)
+	if status != 200 {
+		t.Fatal(status, body)
+	}
+	session = awaitPlayable(t, server.URL, session.ID)
+	if session.URL == previous || session.Offset != 5 || session.PlaybackRate != 2 || manager.Active() != 1 {
+		t.Fatalf("playback speed lost slot or position: %+v", session)
+	}
+	assertLocalMedia(t, server.URL, session.URL, portal.URL)
 	if mediaRequests.Load() == 0 {
 		t.Fatal("media bypassed central server")
 	}
-	previous := session.URL
+	previous = session.URL
 	status, body = request(t, server.URL, "POST", "/api/sessions/"+session.ID+"/seek", `{"position":5}`)
 	if status != 200 {
 		t.Fatal(status, body)
 	}
 	session = awaitPlayable(t, server.URL, session.ID)
-	if session.URL == previous || session.Offset != 5 || manager.Active() != 1 {
+	if session.URL == previous || session.Offset != 5 || session.PlaybackRate != 2 || manager.Active() != 1 {
 		t.Fatalf("seek lost slot or timeline: %+v", session)
 	}
 	assertLocalMedia(t, server.URL, session.URL, portal.URL)
@@ -194,7 +210,7 @@ func TestPortalToBrowserPlayback(t *testing.T) {
 		t.Fatal("stopped media still accessible")
 	}
 	if os.Getenv("RESTREAM_BROWSER_TESTS") == "1" {
-		for _, check := range []struct{ script, mode string }{{"subtitles-playback.mjs", "copy"}, {"subtitles-playback.mjs", "transcode"}, {"embedded-captions-playback.mjs", "copy"}, {"embedded-captions-playback.mjs", "transcode"}, {"real-playback.mjs", "transcode"}} {
+		for _, check := range []struct{ script, mode string }{{"speed-playback.mjs", "copy"}, {"speed-playback.mjs", "transcode"}, {"subtitles-playback.mjs", "copy"}, {"subtitles-playback.mjs", "transcode"}, {"embedded-captions-playback.mjs", "copy"}, {"embedded-captions-playback.mjs", "transcode"}, {"real-playback.mjs", "transcode"}} {
 			if err := manager.Configure(1, check.mode); err != nil {
 				t.Fatal(err)
 			}

internal/app/preferences_integration_test.go

@@ -80,6 +80,9 @@ func (p *policyStreams) Heartbeat(id string) error { _, err := p.Get(id); return
 func (p *policyStreams) Seek(_ context.Context, id string, _ float64) (stream.Session, error) {
 	return p.Get(id)
 }
+func (p *policyStreams) SetPlaybackRate(_ context.Context, id string, _, _ float64) (stream.Session, error) {
+	return p.Get(id)
+}
 func (p *policyStreams) Stop(id string) error {
 	p.mu.Lock()
 	defer p.mu.Unlock()

internal/app/server.go

@@ -30,6 +30,7 @@ type Streams interface {
 	Get(string) (stream.Session, error)
 	Heartbeat(string) error
 	Seek(context.Context, string, float64) (stream.Session, error)
+	SetPlaybackRate(context.Context, string, float64, float64) (stream.Session, error)
 	Stop(string) error
 	Active() int
 	MediaHandler() http.Handler
@@ -157,6 +158,7 @@ func New(cfg Config, provider model.Provider, streams Streams) (*Server, error)
 	mux.HandleFunc("GET /api/sessions/{id}", s.session)
 	mux.HandleFunc("POST /api/sessions/{id}/heartbeat", s.heartbeat)
 	mux.HandleFunc("POST /api/sessions/{id}/seek", s.seek)
+	mux.HandleFunc("POST /api/sessions/{id}/speed", s.playbackSpeed)
 	mux.HandleFunc("DELETE /api/sessions/{id}", s.stop)
 	if streams != nil {
 		mux.Handle("/api/streams/", s.policyMedia(streams.MediaHandler()))
@@ -833,6 +835,33 @@ func (s *Server) seek(w http.ResponseWriter, r *http.Request) {
 	}
 	respond(w, 200, session)
 }
+func (s *Server) playbackSpeed(w http.ResponseWriter, r *http.Request) {
+	if !s.sessionAllowed(w, r, r.PathValue("id")) {
+		return
+	}
+	if s.streams == nil {
+		streamError(w, stream.ErrNotFound)
+		return
+	}
+	var body struct {
+		Position float64 `json:"position"`
+		Rate     float64 `json:"rate"`
+	}
+	if !decode(w, r, &body) {
+		return
+	}
+	if !validPosition(body.Position) || math.IsNaN(body.Rate) || body.Rate < 0.5 || body.Rate > 2 {
+		fail(w, 400, "invalid", "Choose a playback speed between 0.5× and 2× and a valid position.")
+		return
+	}
+	session, err := s.streams.SetPlaybackRate(r.Context(), r.PathValue("id"), body.Position, body.Rate)
+	if err != nil {
+		streamError(w, err)
+		return
+	}
+	respond(w, 200, session)
+}
+
 func (s *Server) stop(w http.ResponseWriter, r *http.Request) {
 	if !s.sessionOwned(w, r, r.PathValue("id")) {
 		return

internal/stream/buffering_integration_test.go

@@ -0,0 +1,129 @@
+package stream
+
+import (
+	"context"
+	"errors"
+	"io"
+	"math"
+	"net/http"
+	"net/http/httptest"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/pushpinderbal/restream/internal/model"
+)
+
+func TestPlaybackBufferFollowsSpeed(t *testing.T) {
+	if _, err := exec.LookPath("ffmpeg"); err != nil {
+		t.Skip("ffmpeg required")
+	}
+	video := filepath.Join(t.TempDir(), "source.mp4")
+	cmd := exec.Command("ffmpeg", "-hide_banner", "-loglevel", "error", "-f", "lavfi", "-i", "testsrc2=size=160x90:rate=10", "-t", "150", "-c:v", "libx264", "-preset", "ultrafast", "-g", "10", "-pix_fmt", "yuv420p", "-movflags", "+faststart", video)
+	if out, err := cmd.CombinedOutput(); err != nil {
+		t.Fatalf("generate media: %v: %s", err, out)
+	}
+	upstream := httptest.NewServer(http.FileServer(http.Dir(filepath.Dir(video))))
+	defer upstream.Close()
+	m, err := New(Config{MaxStreams: 1, TranscodeMode: "copy", SessionTTL: time.Minute}, func(context.Context, model.Item) (model.Source, error) {
+		return model.Source{URL: upstream.URL + "/source.mp4"}, nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer m.Close()
+	media := httptest.NewServer(m.MediaHandler())
+	defer media.Close()
+	s, err := m.Start(context.Background(), model.Item{ID: "episode", Kind: "episode"}, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	s = waitState(t, m, s.ID, "ready")
+	if s.PlaybackRate != 1 {
+		t.Fatalf("default speed: %+v", s)
+	}
+	edge := func() float64 {
+		t.Helper()
+		response, err := http.Get(media.URL + s.URL)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer response.Body.Close()
+		body, err := io.ReadAll(response.Body)
+		if err != nil || response.StatusCode != 200 {
+			t.Fatalf("read playlist: %d %v", response.StatusCode, err)
+		}
+		duration := 0.0
+		for _, line := range strings.Split(string(body), "\n") {
+			if value, ok := strings.CutPrefix(line, "#EXTINF:"); ok {
+				seconds, err := strconv.ParseFloat(strings.TrimSuffix(value, ","), 64)
+				if err != nil {
+					t.Fatal(err)
+				}
+				duration += seconds
+			}
+		}
+		return duration
+	}
+	waitEdge := func(want float64) {
+		t.Helper()
+		deadline := time.Now().Add(5 * time.Second)
+		for time.Now().Before(deadline) {
+			if edge() >= want {
+				return
+			}
+			time.Sleep(50 * time.Millisecond)
+		}
+		t.Fatalf("initial buffer did not reach %v seconds: %v", want, edge())
+	}
+	assertPacing := func(rate float64) {
+		t.Helper()
+		before, started := edge(), time.Now()
+		// Ten seconds separates the 20% headroom from real-time pacing even
+		// when the playlist advances in whole one-second segments.
+		time.Sleep(10 * time.Second)
+		got := (edge() - before) / time.Since(started).Seconds()
+		if math.Abs(got-rate) > 0.15 {
+			t.Fatalf("relay advanced at %.2fx; want %.2fx", got, rate)
+		}
+	}
+	waitEdge(30)
+	assertPacing(1.2)
+	for _, rate := range []float64{0, 0.49, 2.01, math.NaN(), math.Inf(1)} {
+		if _, err := m.SetPlaybackRate(context.Background(), s.ID, 10, rate); !errors.Is(err, ErrInvalid) {
+			t.Fatalf("invalid speed %v accepted: %v", rate, err)
+		}
+	}
+	previous := s.URL
+	s, err = m.SetPlaybackRate(context.Background(), s.ID, 10, 2)
+	if err != nil || s.Offset != 10 || s.PlaybackRate != 2 || s.URL == previous || m.Active() != 1 {
+		t.Fatalf("speed change lost position or slot: %+v %v", s, err)
+	}
+	s = waitState(t, m, s.ID, "ready")
+	waitEdge(60)
+	assertPacing(2.4)
+	previous = s.URL
+	s, err = m.Seek(context.Background(), s.ID, 20)
+	if err != nil || s.PlaybackRate != 2 || s.Offset != 20 || s.URL == previous {
+		t.Fatalf("seek lost selected speed: %+v %v", s, err)
+	}
+	s = waitState(t, m, s.ID, "ready")
+	previous = s.URL
+	s, err = m.SetPlaybackRate(context.Background(), s.ID, 20, 1)
+	if err != nil || s.PlaybackRate != 1 || s.Offset != 20 || s.URL == previous {
+		t.Fatalf("return to normal speed: %+v %v", s, err)
+	}
+	if err := m.Stop(s.ID); err != nil || m.Active() != 0 {
+		t.Fatalf("stop sped-up playback: %v", err)
+	}
+	s, err = m.Start(context.Background(), model.Item{ID: "live", Kind: "live"}, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := m.SetPlaybackRate(context.Background(), s.ID, 0, 2); !errors.Is(err, ErrInvalid) {
+		t.Fatalf("live speed change accepted: %v", err)
+	}
+}

internal/stream/stream.go

@@ -42,6 +42,7 @@ type Session struct {
 	State          string          `json:"state"`
 	Duration       float64         `json:"duration"`
 	Offset         float64         `json:"offset"`
+	PlaybackRate   float64         `json:"playbackRate"`
 	Error          string          `json:"error,omitempty"`
 	Subtitles      []SubtitleTrack `json:"subtitles,omitempty"`
 	SubtitleOffset *float64        `json:"subtitleOffset,omitempty"`
@@ -53,6 +54,7 @@ type generation struct {
 	dir       string
 	done      chan struct{}
 	subtitles *subtitleState
+	rate      float64
 }
 
 type probeResult struct {
@@ -198,7 +200,7 @@ func (m *Manager) Start(ctx context.Context, item model.Item, start float64) (Se
 		logger = slog.Default()
 	}
 	e := &entry{item: item, transcodeMode: m.cfg.TranscodeMode, heartbeat: time.Now(), logger: logger.With("session_id", id)}
-	e.session = Session{ID: id, Duration: validDuration(item.Duration)}
+	e.session = Session{ID: id, Duration: validDuration(item.Duration), PlaybackRate: 1}
 	m.sessions[id] = e
 	m.launchLocked(e, start, nil)
 	return e.session, nil
@@ -208,7 +210,7 @@ func (m *Manager) launchLocked(e *entry, pos float64, prior <-chan struct{}) {
 	m.next++
 	number := m.next
 	ctx, cancel := context.WithCancel(m.ctx)
-	g := &generation{number: number, cancel: cancel, dir: filepath.Join(m.dir, e.session.ID, strconv.FormatUint(number, 10)), done: make(chan struct{})}
+	g := &generation{number: number, cancel: cancel, dir: filepath.Join(m.dir, e.session.ID, strconv.FormatUint(number, 10)), done: make(chan struct{}), rate: e.session.PlaybackRate}
 	e.gen = g
 	e.session.State = "starting"
 	e.session.Offset = pos
@@ -252,6 +254,19 @@ func (m *Manager) Heartbeat(id string) error {
 }
 
 func (m *Manager) Seek(ctx context.Context, id string, position float64) (Session, error) {
+	return m.restart(ctx, id, position, nil)
+}
+
+// SetPlaybackRate restarts the relay at the viewer's position so FFmpeg's input
+// pacing and initial buffer match the selected speed. Seeking retains this rate.
+func (m *Manager) SetPlaybackRate(ctx context.Context, id string, position, rate float64) (Session, error) {
+	if math.IsNaN(rate) || rate < 0.5 || rate > 2 {
+		return Session{}, ErrInvalid
+	}
+	return m.restart(ctx, id, position, &rate)
+}
+
+func (m *Manager) restart(ctx context.Context, id string, position float64, rate *float64) (Session, error) {
 	if err := ctx.Err(); err != nil {
 		return Session{}, err
 	}
@@ -273,8 +288,14 @@ func (m *Manager) Seek(ctx context.Context, id string, position float64) (Sessio
 	if e.session.Duration > 0 && position >= e.session.Duration {
 		return Session{}, ErrInvalid
 	}
+	if rate != nil {
+		if *rate == e.session.PlaybackRate {
+			return e.session, nil
+		}
+		e.session.PlaybackRate = *rate
+	}
 	old := e.gen
-	e.logger.Debug("Stream seeking", "offset", position)
+	e.logger.Debug("Stream seeking", "offset", position, "playback_rate", e.session.PlaybackRate)
 	old.cancel()
 	go m.discard(old)
 	m.launchLocked(e, position, old.done)
@@ -474,10 +495,13 @@ func (m *Manager) run(ctx context.Context, e *entry, g *generation, pos float64,
 	}
 	m.update(e, g, func(s *Session) { s.Duration = duration })
 	live := source.Live || e.item.Kind == "live"
-	args := []string{"-hide_banner", "-loglevel", "error", "-nostdin", "-y", "-readrate", "1"}
+	args := []string{"-hide_banner", "-loglevel", "error", "-nostdin", "-y"}
 	if !live {
-		// Fill initial segments immediately, then pace the relay.
-		args = append(args, "-readrate_initial_burst", "8")
+		// Preserve 30 seconds of viewing time at every speed, then build the
+		// server buffer with 20% headroom over the viewer's consumption rate.
+		args = append(args, "-readrate", strconv.FormatFloat(g.rate*1.2, 'f', -1, 64), "-readrate_initial_burst", strconv.FormatFloat(30*g.rate, 'f', -1, 64))
+	} else {
+		args = append(args, "-readrate", "1")
 	}
 	args = append(args, "-rw_timeout", "10000000", "-protocol_whitelist", "http,https,tcp,tls,crypto")
 	if pos > 0 {