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

feat(stream): persistent stream server with file swapping

Browse source
This commit is contained in:
Deivid Soto 2026-04-07 19:08:37 +02:00
parent 080fdf4d76
commit 5994a30447
11 changed files with 354 additions and 282 deletions
Download patch file Download diff file Expand all files Collapse all files

111
internal/cmd/daemon.go
View file

@ -151,6 +151,9 @@ func runDaemonStart() error {
DownloadDir: cfg.Download.Dir,
PollInterval: pollInterval,
HeartbeatInterval: heartbeatInterval,
StreamPort: cfg.Download.StreamPort,
LanIP: engine.LanIP(),
TailscaleIP: engine.TailscaleIP(),
}
// Create transport: Hybrid (WS + HTTP fallback) or HTTP-only
@ -236,6 +239,15 @@ func runDaemonStart() error {
},
}, reporter, torrentDl, debridDl, engine.NewUsenetDownloader(httpT.Client()))
// Create persistent stream server — lives for the entire daemon lifecycle.
// One port, one server, swap files with SetFile(). No more port churn.
streamSrv := engine.NewStreamServer(cfg.Download.StreamPort)
if err := streamSrv.Listen(ctx); err != nil {
return fmt.Errorf("start stream server: %w", err)
}
// Update heartbeat with actual port (may differ if configured port was busy)
d.UpdateStreamPort(streamSrv.Port())
// Wire state tracking
d.GetActiveCount = manager.ActiveCount
d.GetCleanableBytes = CleanableBytes
@ -254,7 +266,7 @@ func runDaemonStart() error {
cancelStreamTask(taskID)
})
// Wire: stream requested on active download → start HTTP server
// Wire: stream requested on active download → set file on persistent server
reporter.SetStreamRequestedHandler(func(taskID string) {
task := manager.GetTask(taskID)
if task == nil {
@ -264,19 +276,18 @@ func runDaemonStart() error {
if task.GetStreamURL() != "" {
return // already streaming
}
srv, err := torrentDl.StartStream(taskID)
provider, err := torrentDl.GetStreamProvider(taskID)
if err != nil {
log.Printf("[%s] stream failed: %v", taskID[:8], err)
return
}
// Register server before setting URL to avoid TOCTOU race
streamRegistry.mu.Lock()
streamRegistry.servers[taskID] = srv
streamRegistry.mu.Unlock()
task.SetStreamURL(srv.URL())
cancelStreamContexts()
streamSrv.SetFile(provider, taskID)
task.SetStreamURL(streamSrv.URLsJSON())
log.Printf("[%s] streaming active download: %s", taskID[:8], provider.FileName())
// Start watch progress reporter
go engine.NewWatchReporter(agentClient, srv, taskID).Run(ctx)
go engine.NewWatchReporter(agentClient, streamSrv, taskID).Run(ctx)
})
// Wire: daemon claimed tasks -> manager
@ -288,15 +299,15 @@ func runDaemonStart() error {
if isStreamingTask(t.ID) {
continue
}
// Only 1 stream at a time: cancel all existing streams
cancelAllStreams()
// Only 1 stream at a time: cancel existing stream goroutines + clear file
cancelStreamContexts()
streamSrv.ClearFile()
// Reserve slot before spawning goroutine to prevent TOCTOU race.
// streamCancel is stored in streamRegistry and called by cancelAllStreams/cancelStreamTask.
streamCtx, streamCancel := context.WithCancel(ctx) //nolint:gosec // G118: cancel ownership transferred to streamRegistry
streamRegistry.mu.Lock()
streamRegistry.cancels[t.ID] = streamCancel
streamRegistry.mu.Unlock()
go handleStreamTask(streamCtx, t, reporter, cfg, agentClient)
go handleStreamTask(streamCtx, t, reporter, cfg, agentClient, streamSrv)
} else if t.ForceStart || manager.HasCapacity() {
manager.Submit(ctx, t)
} else {
@ -305,16 +316,13 @@ func runDaemonStart() error {
}
}
// Wire: stream requests for completed downloads → serve file from disk
// Wire: stream requests for completed downloads → set file on persistent server
d.OnStreamRequested = func(sr agent.StreamRequest) {
// Skip if already streaming this task
if isStreamingTask(sr.TaskID) {
// Skip if already serving this task
if streamSrv.CurrentTaskID() == sr.TaskID {
return
}
// Only 1 stream at a time: cancel all existing streams
cancelAllStreams()
filePath := sr.FilePath
info, err := os.Stat(filePath)
if err != nil {
@ -351,43 +359,24 @@ func runDaemonStart() error {
log.Printf("[%s] resolved directory to video file: %s", sr.TaskID[:8], filepath.Base(filePath))
}
srv := engine.NewStreamServerFromDisk(filePath, cfg.Download.StreamPort)
streamURL, err := srv.Start(ctx)
if err != nil {
log.Printf("[%s] stream failed: %v", sr.TaskID[:8], err)
go func() {
if _, err := transport.SendProgress(ctx, agent.StatusUpdate{
TaskID: sr.TaskID,
Status: "failed",
ErrorMessage: fmt.Sprintf("stream server start failed: %v", err),
}); err != nil {
log.Printf("[%s] stream error report failed: %v", sr.TaskID[:8], err)
}
}()
return
}
// Cancel any active stream goroutines and swap file on the persistent server
cancelStreamContexts()
streamSrv.SetFile(engine.NewDiskFileProvider(filePath), sr.TaskID)
streamRegistry.mu.Lock()
streamRegistry.servers[sr.TaskID] = srv
streamRegistry.mu.Unlock()
log.Printf("[%s] streaming from disk: %s → %s", sr.TaskID[:8], filepath.Base(sr.FilePath), streamURL)
log.Printf("[%s] streaming from disk: %s → %s", sr.TaskID[:8], filepath.Base(filePath), streamSrv.URL())
// Start watch progress reporter
go engine.NewWatchReporter(agentClient, srv, sr.TaskID).Run(ctx)
go engine.NewWatchReporter(agentClient, streamSrv, sr.TaskID).Run(ctx)
// Report stream URL back to the server via transport
// Notify server that stream is ready (clears streamRequested flag)
go func() {
if _, err := transport.SendProgress(ctx, agent.StatusUpdate{
TaskID: sr.TaskID,
StreamURL: streamURL,
TaskID: sr.TaskID,
StreamReady: true,
}); err != nil {
log.Printf("[%s] stream URL report failed: %v", sr.TaskID[:8], err)
log.Printf("[%s] stream ready report failed: %v", sr.TaskID[:8], err)
}
}()
// Auto-shutdown after 30 min of idle (no HTTP requests)
go startIdleGuard(ctx, srv, sr.TaskID)
}
// Wire: WS control actions (pause/cancel/stream pushed from server)
@ -396,34 +385,41 @@ func runDaemonStart() error {
case "cancel":
manager.CancelTask(taskID)
cancelStreamTask(taskID)
if streamSrv.CurrentTaskID() == taskID {
streamSrv.ClearFile()
}
case "pause":
manager.PauseTask(taskID)
cancelStreamTask(taskID)
if streamSrv.CurrentTaskID() == taskID {
streamSrv.ClearFile()
}
case "resume":
log.Printf("[%s] resume requested via WebSocket, triggering poll", taskID[:8])
d.TriggerPoll()
case "stream":
// Skip if already streaming this task
if isStreamingTask(taskID) {
if streamSrv.CurrentTaskID() == taskID {
return
}
task := manager.GetTask(taskID)
if task == nil || task.GetStreamURL() != "" {
return
}
// Only 1 stream at a time: cancel all existing streams
cancelAllStreams()
srv, err := torrentDl.StartStream(taskID)
provider, err := torrentDl.GetStreamProvider(taskID)
if err != nil {
log.Printf("[%s] stream failed: %v", taskID[:8], err)
return
}
streamRegistry.mu.Lock()
streamRegistry.servers[taskID] = srv
streamRegistry.mu.Unlock()
task.SetStreamURL(srv.URL())
cancelStreamContexts()
streamSrv.SetFile(provider, taskID)
task.SetStreamURL(streamSrv.URLsJSON())
log.Printf("[%s] streaming via WS: %s", taskID[:8], provider.FileName())
case "stop-stream":
cancelStreamTask(taskID)
if streamSrv.CurrentTaskID() == taskID {
streamSrv.ClearFile()
}
}
}
@ -477,10 +473,15 @@ func runDaemonStart() error {
errCh <- d.Run(ctx)
}()
// Start idle guard for the persistent stream server
go startIdleGuard(ctx, streamSrv)
// Wait for signal or error
select {
case sig := <-sigCh:
fmt.Printf("\n Received %s, shutting down...\n", sig)
cancelStreamContexts()
streamSrv.Shutdown(context.Background())
cancel()
// Give active downloads 30s to finish
@ -492,6 +493,8 @@ func runDaemonStart() error {
return nil
case err := <-errCh:
cancelStreamContexts()
streamSrv.Shutdown(context.Background())
cancel()
return err
}

14
internal/cmd/stream.go
View file

@ -127,14 +127,14 @@ func runStream(input string, port int, noOpen bool, playerCmd string) error {
}
// Start HTTP server
srv := engine.NewStreamServer(eng, port)
streamURL, err := srv.Start(ctx)
if err != nil {
srv := engine.NewStreamServer(port)
if err := srv.Listen(ctx); err != nil {
eng.Shutdown(context.Background())
return fmt.Errorf("start server: %w", err)
}
srv.SetFile(eng, "cli-stream")
fmt.Printf(" URL: %s\n", streamURL)
fmt.Printf(" URL: %s\n", srv.URL())
fmt.Println()
// Buffer before opening player
@ -159,15 +159,15 @@ func runStream(input string, port int, noOpen bool, playerCmd string) error {
// Open player
if !noOpen {
playerName, _, openErr := engine.OpenPlayer(streamURL, playerCmd)
playerName, _, openErr := engine.OpenPlayer(srv.URL(), playerCmd)
if openErr != nil {
yellow.Printf(" Could not open player: %s\n", openErr)
fmt.Printf(" Open this URL in your player: %s\n", streamURL)
fmt.Printf(" Open this URL in your player: %s\n", srv.URL())
} else {
green.Printf(" Opened in %s\n", playerName)
}
} else {
fmt.Printf(" Open this URL in your player: %s\n", streamURL)
fmt.Printf(" Open this URL in your player: %s\n", srv.URL())
}
fmt.Println()

93
internal/cmd/stream_handler.go
View file

@ -16,8 +16,8 @@ import (
const streamIdleTimeout = 30 * time.Minute
// startIdleGuard monitors a stream server and cancels the task after inactivity.
func startIdleGuard(ctx context.Context, srv *engine.StreamServer, taskID string) {
// startIdleGuard monitors the persistent stream server and clears the file after inactivity.
func startIdleGuard(ctx context.Context, srv *engine.StreamServer) {
ticker := time.NewTicker(60 * time.Second)
defer ticker.Stop()
for {
@ -25,78 +25,69 @@ func startIdleGuard(ctx context.Context, srv *engine.StreamServer, taskID string
case <-ctx.Done():
return
case <-ticker.C:
if srv.IdleSince() > streamIdleTimeout {
log.Printf("[%s] stream idle timeout (%v no HTTP requests), shutting down", taskID[:8], streamIdleTimeout)
cancelStreamTask(taskID)
return
if srv.HasFile() && srv.IdleSince() > streamIdleTimeout {
taskID := srv.CurrentTaskID()
short := taskID
if len(short) > 8 {
short = short[:8]
}
log.Printf("[%s] stream idle timeout (%v no HTTP requests), clearing file", short, streamIdleTimeout)
cancelStreamContexts()
srv.ClearFile()
}
}
}
}
// streamRegistry tracks active stream tasks and servers for cancellation.
// streamRegistry tracks active stream goroutine contexts for cancellation.
// There is only ONE persistent StreamServer — no per-task servers.
var streamRegistry = struct {
mu sync.Mutex
cancels map[string]context.CancelFunc
servers map[string]*engine.StreamServer // servers for active download streams
}{
cancels: make(map[string]context.CancelFunc),
servers: make(map[string]*engine.StreamServer),
}
// cancelAllStreams cancels all active stream tasks and servers (only 1 stream at a time).
func cancelAllStreams() {
// cancelStreamContexts cancels all active stream goroutines (download engines, etc.).
// Does NOT touch the persistent server — call srv.ClearFile() separately if needed.
func cancelStreamContexts() {
streamRegistry.mu.Lock()
cancels := make(map[string]context.CancelFunc, len(streamRegistry.cancels))
for k, v := range streamRegistry.cancels {
cancels[k] = v
delete(streamRegistry.cancels, k)
}
servers := make(map[string]*engine.StreamServer, len(streamRegistry.servers))
for k, v := range streamRegistry.servers {
servers[k] = v
delete(streamRegistry.servers, k)
}
streamRegistry.mu.Unlock()
for _, cancel := range cancels {
cancel()
}
for _, srv := range servers {
srv.Shutdown(context.Background())
}
}
// isStreamingTask returns true if there is an active stream (goroutine or server) for the given task.
// isStreamingTask returns true if there is an active stream goroutine for the given task.
func isStreamingTask(taskID string) bool {
streamRegistry.mu.Lock()
defer streamRegistry.mu.Unlock()
_, inCancels := streamRegistry.cancels[taskID]
_, inServers := streamRegistry.servers[taskID]
return inCancels || inServers
_, ok := streamRegistry.cancels[taskID]
return ok
}
// cancelStreamTask cancels a running stream task and shuts down any stream server.
// cancelStreamTask cancels a specific stream goroutine.
func cancelStreamTask(taskID string) {
streamRegistry.mu.Lock()
cancel, hasCancel := streamRegistry.cancels[taskID]
cancel, ok := streamRegistry.cancels[taskID]
delete(streamRegistry.cancels, taskID)
srv, hasSrv := streamRegistry.servers[taskID]
delete(streamRegistry.servers, taskID)
streamRegistry.mu.Unlock()
if hasCancel {
if ok {
cancel()
}
if hasSrv {
srv.Shutdown(context.Background())
}
}
// handleStreamTask manages a streaming task lifecycle outside the Manager.
// It creates a StreamEngine, buffers, starts an HTTP server, and reports
// progress until the task is cancelled or the download completes.
func handleStreamTask(parentCtx context.Context, at agent.Task, reporter *engine.ProgressReporter, cfg config.Config, agentClient *agent.Client) {
// handleStreamTask manages a streaming task lifecycle for active torrent downloads.
// It creates a StreamEngine, buffers, sets the file on the persistent server,
// and reports progress until the task is cancelled or the download completes.
func handleStreamTask(parentCtx context.Context, at agent.Task, reporter *engine.ProgressReporter, cfg config.Config, agentClient *agent.Client, srv *engine.StreamServer) {
ctx, cancel := context.WithCancel(parentCtx)
defer cancel()
@ -108,6 +99,10 @@ func handleStreamTask(parentCtx context.Context, at agent.Task, reporter *engine
streamRegistry.mu.Lock()
delete(streamRegistry.cancels, at.ID)
streamRegistry.mu.Unlock()
// Clear file from persistent server if we're still the current task
if srv.CurrentTaskID() == at.ID {
srv.ClearFile()
}
}()
task := engine.NewTaskFromAgent(at)
@ -148,36 +143,18 @@ func handleStreamTask(parentCtx context.Context, at agent.Task, reporter *engine
return
}
// 4. Start HTTP server
srv := engine.NewStreamServer(eng, cfg.Download.StreamPort)
streamURL, err := srv.Start(ctx)
if err != nil {
task.ErrorMessage = "start HTTP server: " + err.Error()
task.Transition(engine.StatusFailed)
return
}
streamRegistry.mu.Lock()
streamRegistry.servers[at.ID] = srv
streamRegistry.mu.Unlock()
defer func() {
srv.Shutdown(context.Background())
streamRegistry.mu.Lock()
delete(streamRegistry.servers, at.ID)
streamRegistry.mu.Unlock()
}()
// 5. Report stream URLs — JSON with all network options for smart resolution
// 4. Set file on the persistent stream server (instant, no port binding)
srv.SetFile(eng, at.ID)
task.StreamURL = srv.URLsJSON()
log.Printf("[%s] stream ready: %s (primary: %s)", at.ID[:8], task.StreamURL, streamURL)
log.Printf("[%s] stream ready: %s (url: %s)", at.ID[:8], eng.FileName(), srv.URL())
// 5b. Start watch progress reporter (tracks Range requests for playback position)
// 5. Start watch progress reporter
if agentClient != nil {
watchReporter := engine.NewWatchReporter(agentClient, srv, at.ID)
go watchReporter.Run(ctx)
}
// 6. Start idle guard + progress loop
go startIdleGuard(ctx, srv, at.ID)
// 6. Progress loop until download completes or cancelled
eng.StartProgressLoop(ctx)
progressTicker := time.NewTicker(3 * time.Second)
defer progressTicker.Stop()