torrentclaw/unarr
1
0
Fork 0
Code Issues Pull requests Projects Releases 4 Packages Wiki Activity Actions

feat(stream)!: retire WebRTC, HLS-only, bump 0.9.4
Some checks failed
Release / release (push) Failing after 0s
Details
Release / docker (push) Has been skipped
Details
Release / virustotal (push) Failing after 0s
Details

Browse source 
Drops the custom WebRTC DataChannel pipeline + pion deps + WSS signaling
client + wire framing. Every in-browser playback now uses HLS over HTTP
from the daemon (Tailscale/LAN/UPnP). Browser P2P never re-enabled.

Wire renames (incompatible with web < 2026-05-26): agent.WebRTCSession
=> agent.StreamSession, SyncResponse.WebRTCSessions (JSON: webrtcSessions)
=> StreamSessions (JSON: streamSessions). MIN_AGENT_VERSION is bumped
to 0.9.4 on the web side so older agents see an upgrade card.

Also fixes the libx264 'VBV bitrate > level limit' abort by clamping
the encoder bitrate to the effective output height instead of the
requested label (carried over from the prior 0.9.3 unreleased work).

The seed_file vertical (mode=seed_file handler + engine.SeedFile) was
retired with the in-browser P2P player. [downloads.webrtc] config block
deleted; existing TOML files with the section still parse fine.
This commit is contained in:
Deivid Soto 2026-05-26 18:04:35 +02:00
parent 9176e877eb
commit ca7de23a56
33 changed files with 207 additions and 2854 deletions
Download patch file Download diff file Expand all files Collapse all files

8
internal/agent/daemon.go
View file

@ -38,7 +38,7 @@ type Daemon struct {
// Callbacks — set by cmd/daemon.go before calling Run.
OnTasksClaimed func(tasks []Task)
OnStreamRequested func(req StreamRequest)
OnWebRTCSession func(sess WebRTCSession)
OnStreamSession func(sess StreamSession)
OnControlAction func(action, taskID string, deleteFiles bool)
GetActiveCount func() int // returns number of active downloads (wired from manager)
@ -210,9 +210,9 @@ func (d *Daemon) Run(ctx context.Context) error {
d.OnStreamRequested(req)
}
}
d.sync.OnWebRTCSession = func(sess WebRTCSession) {
if d.OnWebRTCSession != nil {
d.OnWebRTCSession(sess)
d.sync.OnStreamSession = func(sess StreamSession) {
if d.OnStreamSession != nil {
d.OnStreamSession(sess)
}
}
d.sync.OnUpgrade = func(version string) {

258
internal/agent/signal_client.go
View file

@ -1,258 +0,0 @@
package agent
import (
"bufio"
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"strings"
"time"
)
// SignalRole identifies who produced a signalling message. The opposite role
// receives it.
type SignalRole string
const (
SignalRoleBrowser SignalRole = "browser"
SignalRoleAgent SignalRole = "agent"
)
// SignalMessageType matches the server-side z.enum on
// /api/internal/stream/signal/[sessionId] route.
type SignalMessageType string
const (
SignalMsgOffer SignalMessageType = "offer"
SignalMsgAnswer SignalMessageType = "answer"
SignalMsgCandidate SignalMessageType = "candidate"
SignalMsgCandidateEnd SignalMessageType = "candidate-end"
SignalMsgBye SignalMessageType = "bye"
)
// SignalMessage mirrors the bus envelope on the web side.
type SignalMessage struct {
From SignalRole `json:"from"`
Type SignalMessageType `json:"type"`
Payload string `json:"payload"`
TS int64 `json:"ts"`
}
// PostSignal enqueues a signalling message produced by this agent. The
// browser receives it on its next SSE event push.
func (c *Client) PostSignal(ctx context.Context, sessionID string, msg SignalMessage) error {
body := map[string]any{
"from": string(SignalRoleAgent),
"type": string(msg.Type),
"payload": msg.Payload,
}
path := fmt.Sprintf("/api/internal/stream/signal/%s", sessionID)
return c.doPost(ctx, path, body, &struct {
OK bool `json:"ok"`
}{})
}
// SignalEventStream wraps an open SSE connection. Read messages from Events()
// until the channel closes (server timeout or context cancel). Always defer
// Close() to release the underlying response body.
type SignalEventStream struct {
resp *http.Response
cancel context.CancelFunc
events chan SignalMessage
errs chan error
done chan struct{}
}
// Events streams browser-produced messages addressed to the agent.
// The channel closes when the SSE connection ends; the caller should then
// call Close() and reopen if it wants to keep listening.
func (s *SignalEventStream) Events() <-chan SignalMessage { return s.events }
// Err returns the terminating error (if any) once Events() has closed.
func (s *SignalEventStream) Err() error {
select {
case err := <-s.errs:
return err
default:
return nil
}
}
// Close cancels the underlying HTTP request and waits for the reader goroutine
// to drain. Safe to call more than once.
func (s *SignalEventStream) Close() error {
if s.cancel != nil {
s.cancel()
}
if s.resp != nil {
s.resp.Body.Close()
}
<-s.done
return nil
}
// OpenSignalStream opens a long-lived SSE connection to the signal events
// endpoint. Caller MUST cancel ctx (or call Close()) to free resources.
//
// The server caps each response at ~25 s; OpenSignalStream surfaces the
// disconnect by closing the events channel. Caller should reopen until the
// session ends.
func (c *Client) OpenSignalStream(ctx context.Context, sessionID string) (*SignalEventStream, error) {
streamCtx, cancel := context.WithCancel(ctx)
url := fmt.Sprintf("%s/api/internal/stream/signal/%s/events", c.baseURL(), sessionID)
req, err := http.NewRequestWithContext(streamCtx, http.MethodGet, url, nil)
if err != nil {
cancel()
return nil, fmt.Errorf("open signal stream: %w", err)
}
req.Header.Set("Accept", "text/event-stream")
req.Header.Set("Authorization", "Bearer "+c.apiKey)
req.Header.Set("User-Agent", c.userAgent)
req.Header.Set("Cache-Control", "no-cache")
// Use a per-call client with no timeout (SSE connections are long).
sseClient := &http.Client{}
resp, err := sseClient.Do(req)
if err != nil {
cancel()
return nil, fmt.Errorf("open signal stream: %w", err)
}
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
resp.Body.Close()
cancel()
return nil, fmt.Errorf("open signal stream: HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
}
stream := &SignalEventStream{
resp: resp,
cancel: cancel,
events: make(chan SignalMessage, 8),
errs: make(chan error, 1),
done: make(chan struct{}),
}
go stream.read()
return stream, nil
}
// sseMaxLineBytes caps the size of a single SSE line. Real signalling lines
// are JSON payloads of a few hundred bytes; 256 KiB is generous enough to
// survive a future schema bump but small enough that a hostile or buggy
// server cannot grow daemon memory by streaming a single line forever.
const sseMaxLineBytes = 256 * 1024
// sseMaxEventBytes caps the total bytes buffered across the lines of one
// SSE event. Without a cap, a peer could send unbounded `data:` continuation
// lines and OOM the daemon between blank-line dispatches.
const sseMaxEventBytes = 1024 * 1024
func (s *SignalEventStream) read() {
defer close(s.done)
defer close(s.events)
scanner := bufio.NewScanner(s.resp.Body)
scanner.Buffer(make([]byte, 16*1024), sseMaxLineBytes)
var dataBuf bytes.Buffer
var eventName string
for scanner.Scan() {
line := strings.TrimRight(scanner.Text(), "\r")
if line == "" {
// End of an event — dispatch if we have data.
if dataBuf.Len() == 0 {
eventName = ""
continue
}
if eventName == "" || eventName == "signal" {
var msg SignalMessage
if err := json.Unmarshal(dataBuf.Bytes(), &msg); err == nil {
select {
case s.events <- msg:
case <-s.resp.Request.Context().Done():
return
}
}
}
dataBuf.Reset()
eventName = ""
continue
}
if strings.HasPrefix(line, ":") {
// SSE comment (heartbeat); ignore.
continue
}
if strings.HasPrefix(line, "event:") {
eventName = strings.TrimSpace(line[len("event:"):])
continue
}
if strings.HasPrefix(line, "data:") {
payload := strings.TrimSpace(line[len("data:"):])
// Refuse to grow the event buffer past the cap. Reset so a
// well-formed event after the offender can still be parsed,
// and surface an error so SignalLoop reconnects.
if dataBuf.Len()+len(payload)+1 > sseMaxEventBytes {
dataBuf.Reset()
eventName = ""
select {
case s.errs <- fmt.Errorf("sse: event exceeded %d bytes", sseMaxEventBytes):
default:
}
return
}
if dataBuf.Len() > 0 {
dataBuf.WriteByte('\n')
}
dataBuf.WriteString(payload)
continue
}
// id:, retry:, anything else — ignore for now.
}
if err := scanner.Err(); err != nil {
select {
case s.errs <- err:
default:
}
}
}
// SignalLoop runs an SSE consumer that reconnects automatically on disconnect.
// onMessage is called for every browser-produced message. Returns when ctx is
// cancelled. Reconnect backoff is fixed at 1 s — the server already paces
// reconnects with `retry: 1500` headers so churn is bounded.
func (c *Client) SignalLoop(ctx context.Context, sessionID string, onMessage func(SignalMessage)) error {
for ctx.Err() == nil {
stream, err := c.OpenSignalStream(ctx, sessionID)
if err != nil {
select {
case <-time.After(time.Second):
case <-ctx.Done():
return ctx.Err()
}
continue
}
for msg := range stream.Events() {
onMessage(msg)
}
streamErr := stream.Err()
stream.Close()
if ctx.Err() != nil {
return ctx.Err()
}
// Server closes the SSE every ~25 s; reconnect immediately.
// Hard error → small backoff so we don't hammer.
if streamErr != nil {
select {
case <-time.After(time.Second):
case <-ctx.Done():
return ctx.Err()
}
}
}
return ctx.Err()
}

196
internal/agent/signal_client_test.go
View file

@ -1,196 +0,0 @@
package agent
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
"time"
)
// fakeSSEServer streams a fixed set of SSE events then closes the connection.
func fakeSSEServer(t *testing.T, msgs []SignalMessage, holdOpenAfter bool) *httptest.Server {
t.Helper()
return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("Authorization") != "Bearer test-key" {
http.Error(w, "auth", http.StatusUnauthorized)
return
}
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
flusher, ok := w.(http.Flusher)
if !ok {
t.Fatal("server: ResponseWriter is not a Flusher")
}
fmt.Fprint(w, "retry: 1500\n\n")
flusher.Flush()
for _, m := range msgs {
data, _ := json.Marshal(m)
fmt.Fprintf(w, "id: %d\nevent: signal\ndata: %s\n\n", m.TS, data)
flusher.Flush()
}
// Send a heartbeat comment to verify it's ignored.
fmt.Fprint(w, ": heartbeat\n\n")
flusher.Flush()
if holdOpenAfter {
// Hold the connection until the client disconnects so the test can
// exercise stream.Close().
<-r.Context().Done()
}
}))
}
func TestSignalStreamReadsMessages(t *testing.T) {
want := []SignalMessage{
{From: SignalRoleBrowser, Type: SignalMsgOffer, Payload: "{sdp:1}", TS: 1},
{From: SignalRoleBrowser, Type: SignalMsgCandidate, Payload: "{cand:1}", TS: 2},
}
srv := fakeSSEServer(t, want, false)
defer srv.Close()
c := NewClient(srv.URL, "test-key", "test-ua")
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
stream, err := c.OpenSignalStream(ctx, "session-1")
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
var got []SignalMessage
for m := range stream.Events() {
got = append(got, m)
if len(got) == len(want) {
break
}
}
if len(got) != len(want) {
t.Fatalf("got %d messages, want %d", len(got), len(want))
}
for i, m := range got {
if m.From != want[i].From || m.Type != want[i].Type || m.Payload != want[i].Payload {
t.Errorf("[%d] mismatch: %+v want %+v", i, m, want[i])
}
}
}
func TestSignalStreamPropagatesAuthError(t *testing.T) {
srv := fakeSSEServer(t, nil, false)
defer srv.Close()
c := NewClient(srv.URL, "wrong-key", "test-ua")
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
_, err := c.OpenSignalStream(ctx, "session-1")
if err == nil {
t.Fatal("expected auth error, got nil")
}
}
func TestSignalStreamCloseCancelsRead(t *testing.T) {
srv := fakeSSEServer(t, nil, true)
defer srv.Close()
c := NewClient(srv.URL, "test-key", "test-ua")
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
stream, err := c.OpenSignalStream(ctx, "session-1")
if err != nil {
t.Fatalf("open: %v", err)
}
// Close on a separate goroutine then make sure the events channel drains.
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
time.Sleep(50 * time.Millisecond)
stream.Close()
}()
for range stream.Events() {
// drain
}
wg.Wait()
}
// TestSignalStreamRejectsOversizedEvent verifies that a hostile or buggy
// server sending an unbounded `data:` event surfaces an error and stops
// the reader instead of growing daemon memory forever.
func TestSignalStreamRejectsOversizedEvent(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("Authorization") != "Bearer test-key" {
http.Error(w, "auth", http.StatusUnauthorized)
return
}
w.Header().Set("Content-Type", "text/event-stream")
flusher := w.(http.Flusher)
// Send many data: continuation lines until we blow past the
// per-event cap. Each chunk is a short legitimate-looking line.
chunk := "data: " + strings.Repeat("x", 4096) + "\n"
fmt.Fprint(w, "event: signal\n")
for i := 0; i < (sseMaxEventBytes/4096)+8; i++ {
fmt.Fprint(w, chunk)
}
flusher.Flush()
<-r.Context().Done()
}))
defer srv.Close()
c := NewClient(srv.URL, "test-key", "test-ua")
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
stream, err := c.OpenSignalStream(ctx, "session-overflow")
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
for range stream.Events() {
// Should never receive a parsed event — the over-sized buffer must
// be rejected before dispatch.
}
if err := stream.Err(); err == nil {
t.Fatal("expected error from oversized event, got nil")
}
}
func TestPostSignalSendsCorrectBody(t *testing.T) {
var bodySeen map[string]any
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("Authorization") != "Bearer test-key" {
http.Error(w, "auth", http.StatusUnauthorized)
return
}
_ = json.NewDecoder(r.Body).Decode(&bodySeen)
w.Header().Set("Content-Type", "application/json")
fmt.Fprint(w, `{"ok":true}`)
}))
defer srv.Close()
c := NewClient(srv.URL, "test-key", "test-ua")
err := c.PostSignal(context.Background(), "sess-x", SignalMessage{
Type: SignalMsgAnswer,
Payload: "{sdp:answer}",
})
if err != nil {
t.Fatalf("post: %v", err)
}
if bodySeen["from"] != string(SignalRoleAgent) {
t.Errorf("expected from=agent, got %v", bodySeen["from"])
}
if bodySeen["type"] != string(SignalMsgAnswer) {
t.Errorf("expected type=answer, got %v", bodySeen["type"])
}
if bodySeen["payload"] != "{sdp:answer}" {
t.Errorf("expected payload mismatch, got %v", bodySeen["payload"])
}
}

10
internal/agent/sync.go
View file

@ -29,7 +29,7 @@ type SyncClient struct {
OnNewTasks func(tasks []Task)
OnControl func(action, taskID string, deleteFiles bool)
OnStreamRequest func(req StreamRequest)
OnWebRTCSession func(sess WebRTCSession)
OnStreamSession func(sess StreamSession)
OnUpgrade func(version string)
OnScan func()
OnWatchingChange func(watching bool)
@ -199,10 +199,10 @@ func (sc *SyncClient) processResponse(resp *SyncResponse) {
}
}
// WebRTC streaming sessions
for _, ws := range resp.WebRTCSessions {
if sc.OnWebRTCSession != nil {
sc.OnWebRTCSession(ws)
// HLS streaming sessions.
for _, ws := range resp.StreamSessions {
if sc.OnStreamSession != nil {
sc.OnStreamSession(ws)
}
}

33
internal/agent/types.go
View file

@ -374,29 +374,22 @@ type LibraryDeleteRequest struct {
FilePath string `json:"filePath"`
}
// WebRTCSession is a request to open a streaming session for a browser
// player. Transport selects the on-the-wire protocol: empty/"webrtc" runs the
// legacy custom WebRTC DataChannel pipeline; "hls" spawns an HLS session
// (ffmpeg producing fragmented MP4 served over HTTP). The CLI must POST an
// SDP answer to /api/internal/stream/signal/<sessionId> for WebRTC sessions
// and register the HLS session in the StreamServer's HLS registry for HLS
// sessions; either way the source bytes come from FilePath (or, when only
// InfoHash is set, from a download_task on disk).
type WebRTCSession struct {
SessionID string `json:"sessionId"`
// Transport selects the streaming protocol. "" or "webrtc" → legacy
// WebRTC + MSE pipeline (Phase 1). "hls" → HLS over HTTP (Phase 2).
Transport string `json:"transport,omitempty"`
FilePath string `json:"filePath,omitempty"`
InfoHash string `json:"infoHash,omitempty"`
TaskID string `json:"taskId,omitempty"`
FileName string `json:"fileName,omitempty"`
FileSize int64 `json:"fileSize,omitempty"`
// StreamSession is a request to open an HLS streaming session for an
// in-browser player. The CLI registers the HLS session in the StreamServer's
// HLS registry; source bytes come from FilePath (or, when only InfoHash is
// set, from a download_task on disk).
type StreamSession struct {
SessionID string `json:"sessionId"`
FilePath string `json:"filePath,omitempty"`
InfoHash string `json:"infoHash,omitempty"`
TaskID string `json:"taskId,omitempty"`
FileName string `json:"fileName,omitempty"`
FileSize int64 `json:"fileSize,omitempty"`
// Quality target the daemon should aim for when transcoding. One of
// "2160p" | "1080p" | "720p" | "480p" | "original" | "" (defer to config).
Quality string `json:"quality,omitempty"`
// AudioIndex selects the source audio track (-map 0:a:N). -1 means
// "use the default/first track" (HLS) or ignored (WebRTC).
// "use the default/first track".
AudioIndex int `json:"audioIndex,omitempty"`
}
@ -405,7 +398,7 @@ type SyncResponse struct {
NewTasks []Task `json:"newTasks,omitempty"`
Controls []ControlAction `json:"controls,omitempty"`
StreamRequests []StreamRequest `json:"streamRequests,omitempty"`
WebRTCSessions []WebRTCSession `json:"webrtcSessions,omitempty"`
StreamSessions []StreamSession `json:"streamSessions,omitempty"`
Watching bool `json:"watching"`
Upgrade *UpgradeSignal `json:"upgrade,omitempty"`
Scan bool `json:"scan,omitempty"`