feat(downloads): distribute preparation and delivery (#607)

* feat(downloads): distribute preparation and delivery

* fix(downloads): harden distributed job routing

* fix(downloads): release probe capacity and bound tracking
This commit is contained in:
Quick
2026-08-12 15:08:51 -04:00
committed by GitHub
parent 24c3feeb64
commit 9c69760106
29 changed files with 2126 additions and 90 deletions
+13 -1
View File
@@ -135,6 +135,18 @@ transcoding.
Multi-host operators can use those examples as a starting point for a dedicated worker Compose
file connected to the deployment's shared PostgreSQL and Redis services.
Proxy nodes serve source downloads from the same absolute media paths used by direct playback.
Prepared-download work can also run on transcode nodes and the resulting file can be served by a
proxy node when `download.artifact_dir` is explicitly configured. Mount that directory at the same
absolute path on the API, transcode, and proxy nodes. With the default node-local artifact directory,
Silo keeps preparation and prepared-file delivery on the API node. Downloads with a configured
server-wide or per-user bandwidth limit also remain API-local so those aggregate limits stay exact.
Clients discover distributed delivery through `proxy_delivery` on the download capability response.
When it is true, they may opt into `GET` or `HEAD /api/v1/downloads/{id}/file-proxy` and
`/api/v1/direct-download-proxy`; those routes may return a temporary redirect to a proxy node. The
established `/file` and `/direct-download` routes keep serving bytes directly with their existing
status-code contract.
### Deployment Notes
The default compose stack intentionally bundles PostgreSQL and Redis for ease of setup and assumes a fresh install without those services already available. If you already operate PostgreSQL and Redis, omit those examples from compose and point Silo at your existing infrastructure instead. For serious installs, PostgreSQL is better on a separate VM or a managed service so upgrades, tuning, and backups are isolated from the app host. Redis can stay local for many installs, but externalizing it is also reasonable if you already operate shared infrastructure.
@@ -158,7 +170,7 @@ through the admin UI after first launch.
| `integrated` | Full server: API + frontend + scanner + transcode (default) |
| `api` | API server only, no local transcoding |
| `proxy` | Stream proxy node that connects to the shared deployment database and Redis |
| `transcode` | HLS transcode worker node that connects to the shared deployment database and Redis |
| `transcode` | HLS and prepared-download worker node that connects to the shared deployment database and Redis |
### PostgreSQL Auto-Tuning
+16 -9
View File
@@ -721,6 +721,7 @@ func main() {
handler = srv.Handler()
} else {
srv := transcodenode.NewServer(watcher, tracker)
srv.SetInputPathAuthorizer(transcodenode.NewMediaRootAuthorizer(catalog.NewFolderRepository(pool)))
srv.SetFFmpegLogSink(playback.NewSlogFFmpegLogSink(slog.Default(), nodeID))
// Read jellycompat reconstruction recipes central wrote at transcode
// start, so this node can rebuild a Jellyfin transcode after its own
@@ -2060,20 +2061,26 @@ func main() {
// Download prepare-to-file pipeline (Phase 3): a durable, leased encode
// queue hosted on the task manager. Built here (before Start) and shared
// with the API via deps so the download service can enqueue jobs.
liveDownloadConfig := func() *config.Config {
if deps.LiveConfig != nil {
if c := deps.LiveConfig(); c != nil {
return c
}
}
return deps.Config
}
var downloadWorkPlanner nodepool.TranscodeWorkPlanner
if deps.NodePlanner != nil {
downloadWorkPlanner = deps.NodePlanner
}
preparer := downloads.NewNodeAwarePreparer(downloads.NewPlaybackPreparer(), downloadWorkPlanner, liveDownloadConfig)
artifactMgr := downloads.NewArtifactManager(
downloads.NewArtifactRepository(deps.DB),
downloads.NewRepository(deps.DB),
deps.FileRepo,
downloads.NewPlaybackPreparer(),
preparer,
deps.NodeID,
func() *config.Config {
if deps.LiveConfig != nil {
if c := deps.LiveConfig(); c != nil {
return c
}
}
return deps.Config
},
liveDownloadConfig,
func(ctx context.Context, d *downloads.Download) {
if deps.EventsHub == nil {
return
+227 -19
View File
@@ -4,9 +4,14 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"net/http"
"net/url"
"strconv"
"strings"
"sync"
"time"
"github.com/go-chi/chi/v5"
@@ -14,7 +19,9 @@ import (
"github.com/Silo-Server/silo-server/internal/catalog"
"github.com/Silo-Server/silo-server/internal/downloads"
"github.com/Silo-Server/silo-server/internal/httpstream"
"github.com/Silo-Server/silo-server/internal/nodepool"
"github.com/Silo-Server/silo-server/internal/playback"
"github.com/Silo-Server/silo-server/internal/streamtoken"
)
// DownloadService is the interface that the download handler depends on. A
@@ -44,14 +51,51 @@ type DownloadService interface {
// DownloadHandler handles download endpoints.
type DownloadHandler struct {
svc DownloadService
svc DownloadService
nodePlanner nodepool.DownloadPlanner
jwtSecret func() string
proxyHTTPClient *http.Client
preflightMu sync.Mutex
preflightCache map[string]proxyPreflightResult
}
type proxyPreflightResult struct {
reachable bool
expiresAt time.Time
}
// NewDownloadHandler creates a new DownloadHandler.
func NewDownloadHandler(svc DownloadService) *DownloadHandler {
return &DownloadHandler{svc: svc}
return &DownloadHandler{
svc: svc,
preflightCache: make(map[string]proxyPreflightResult),
proxyHTTPClient: &http.Client{
Timeout: 3 * time.Second,
CheckRedirect: func(_ *http.Request, _ []*http.Request) error {
return http.ErrUseLastResponse
},
},
}
}
// SetProxyDelivery enables stateless proxy-node egress after the API service
// has authorized and resolved a download target. It is optional; integrated
// deployments continue to serve bytes locally.
func (h *DownloadHandler) SetProxyDelivery(planner nodepool.DownloadPlanner, jwtSecret func() string) {
h.nodePlanner = planner
h.jwtSecret = jwtSecret
}
type downloadFileResolver interface {
ResolveDirectFile(ctx context.Context, userID, fileID int, format string, filter catalog.AccessFilter) (*downloads.FileTarget, error)
ResolveManagedFile(ctx context.Context, userID int, profileID, deviceID, downloadID string, filter catalog.AccessFilter) (*downloads.FileTarget, error)
}
const (
proxyDownloadTokenTTL = 15 * time.Minute
proxyPreflightCacheTTL = 5 * time.Second
)
// downloadRequest represents the JSON body for POST /downloads.
type downloadRequest struct {
ContentID string `json:"content_id"`
@@ -124,6 +168,7 @@ type downloadCapabilityResponse struct {
SeasonDownload bool `json:"season_download"`
SeriesMonitoring bool `json:"series_monitoring"`
MonitoringModes []string `json:"monitoring_modes,omitempty"`
ProxyDelivery bool `json:"proxy_delivery"`
}
func toDownloadResponse(d *downloads.Download) downloadResponse {
@@ -179,6 +224,7 @@ func (h *DownloadHandler) HandleCapability(w http.ResponseWriter, r *http.Reques
return
}
_, _, proxyDelivery := h.proxyTarget()
writeJSON(w, http.StatusOK, downloadCapabilityResponse{
Enabled: capInfo.Enabled,
DownloadAllowed: capInfo.DownloadAllowed,
@@ -188,6 +234,7 @@ func (h *DownloadHandler) HandleCapability(w http.ResponseWriter, r *http.Reques
SeasonDownload: capInfo.SeasonDownload,
SeriesMonitoring: capInfo.SeriesMonitoring,
MonitoringModes: capInfo.MonitoringModes,
ProxyDelivery: proxyDelivery,
})
}
@@ -382,6 +429,17 @@ func (h *DownloadHandler) HandlePatchDownload(w http.ResponseWriter, r *http.Req
// HandleDownloadFile handles GET /downloads/{id}/file.
func (h *DownloadHandler) HandleDownloadFile(w http.ResponseWriter, r *http.Request) {
h.handleDownloadFile(w, r, false)
}
// HandleDownloadFileViaProxy serves the additive proxy-aware download route.
// Clients opt into its 307 response only after discovering proxy_delivery in
// the download capability; the established file route retains 200/206.
func (h *DownloadHandler) HandleDownloadFileViaProxy(w http.ResponseWriter, r *http.Request) {
h.handleDownloadFile(w, r, true)
}
func (h *DownloadHandler) handleDownloadFile(w http.ResponseWriter, r *http.Request, delegate bool) {
userID := apimw.GetUserID(r.Context())
if userID == 0 {
writeError(w, http.StatusUnauthorized, "unauthorized", "Authentication required")
@@ -399,35 +457,53 @@ func (h *DownloadHandler) HandleDownloadFile(w http.ResponseWriter, r *http.Requ
profileID, deviceID, _, _ := managedIdentity(r)
filter := requestAccessFilter(r)
if delegate && deviceID != "" {
handled, err := h.redirectManagedDownload(r.Context(), w, r, userID, profileID, deviceID, id, filter)
if err != nil {
h.writeDownloadFileError(w, r, id, err)
return
}
if handled {
return
}
}
// Full media downloads outlive the server's absolute WriteTimeout; roll
// the write deadline with progress instead.
sw := httpstream.NewRollingDeadlineWriter(w)
if err := h.svc.ServeFile(r.Context(), sw, r, userID, profileID, deviceID, id, filter); err != nil {
if errors.Is(err, downloads.ErrNotFound) || errors.Is(err, catalog.ErrItemNotFound) {
writeError(w, http.StatusNotFound, "not_found", "Download not found")
return
}
if errors.Is(err, downloads.ErrDownloadNotActive) {
writeError(w, http.StatusConflict, "download_inactive", "This download is no longer active")
return
}
if errors.Is(err, downloads.ErrProfileRequired) {
writeError(w, http.StatusBadRequest, "profile_required", "A profile is required for managed downloads")
return
}
if errors.Is(err, downloads.ErrFeatureDisabled) || errors.Is(err, downloads.ErrDownloadNotAllowed) {
writeError(w, http.StatusForbidden, "forbidden", "You are not allowed to download")
return
}
h.writeDownloadFileError(w, r, id, err)
}
}
func (h *DownloadHandler) writeDownloadFileError(w http.ResponseWriter, r *http.Request, id string, err error) {
switch {
case errors.Is(err, downloads.ErrNotFound), errors.Is(err, catalog.ErrItemNotFound):
writeError(w, http.StatusNotFound, "not_found", "Download not found")
case errors.Is(err, downloads.ErrDownloadNotActive):
writeError(w, http.StatusConflict, "download_inactive", "This download is no longer active")
case errors.Is(err, downloads.ErrProfileRequired):
writeError(w, http.StatusBadRequest, "profile_required", "A profile is required for managed downloads")
case errors.Is(err, downloads.ErrFeatureDisabled), errors.Is(err, downloads.ErrDownloadNotAllowed):
writeError(w, http.StatusForbidden, "forbidden", "You are not allowed to download")
default:
slog.ErrorContext(r.Context(), "failed to serve download file", "component", "api", "download_id", id, "error", err)
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to serve download")
return
}
}
// HandleDirectDownload handles GET /direct-download?file_id=N. A legacy
// format=original query is accepted, but direct downloads remain original-only.
func (h *DownloadHandler) HandleDirectDownload(w http.ResponseWriter, r *http.Request) {
h.handleDirectDownload(w, r, false)
}
// HandleDirectDownloadViaProxy serves the additive proxy-aware direct-download
// route while the established endpoint retains its 200/206 response contract.
func (h *DownloadHandler) HandleDirectDownloadViaProxy(w http.ResponseWriter, r *http.Request) {
h.handleDirectDownload(w, r, true)
}
func (h *DownloadHandler) handleDirectDownload(w http.ResponseWriter, r *http.Request, delegate bool) {
userID := apimw.GetUserID(r.Context())
if userID == 0 {
writeError(w, http.StatusUnauthorized, "unauthorized", "Authentication required")
@@ -446,12 +522,144 @@ func (h *DownloadHandler) HandleDirectDownload(w http.ResponseWriter, r *http.Re
}
filter := requestAccessFilter(r)
if delegate {
if handled, redirectErr := h.redirectDirectDownload(r.Context(), w, r, userID, fileID, r.URL.Query().Get("format"), filter); redirectErr != nil {
h.writeDownloadError(w, redirectErr)
return
} else if handled {
return
}
}
if err := h.svc.ServeDirect(r.Context(), w, r, userID, fileID, r.URL.Query().Get("format"), filter); err != nil {
h.writeDownloadError(w, err)
return
}
}
func (h *DownloadHandler) redirectDirectDownload(ctx context.Context, w http.ResponseWriter, r *http.Request, userID, fileID int, format string, filter catalog.AccessFilter) (bool, error) {
resolver, secret, ok := h.proxyTarget()
if !ok {
return false, nil
}
target, err := resolver.ResolveDirectFile(ctx, userID, fileID, format, filter)
if err != nil {
return false, err
}
return h.redirectToProxy(w, r, secret, target, userID, "")
}
func (h *DownloadHandler) redirectManagedDownload(ctx context.Context, w http.ResponseWriter, r *http.Request, userID int, profileID, deviceID, downloadID string, filter catalog.AccessFilter) (bool, error) {
resolver, secret, ok := h.proxyTarget()
if !ok {
return false, nil
}
target, err := resolver.ResolveManagedFile(ctx, userID, profileID, deviceID, downloadID, filter)
if err != nil {
return false, err
}
return h.redirectToProxy(w, r, secret, target, userID, profileID)
}
func (h *DownloadHandler) proxyTarget() (downloadFileResolver, string, bool) {
resolver, ok := h.svc.(downloadFileResolver)
if !ok || h.nodePlanner == nil || h.jwtSecret == nil {
return nil, "", false
}
secret := strings.TrimSpace(h.jwtSecret())
if secret == "" {
return nil, "", false
}
return resolver, secret, true
}
func (h *DownloadHandler) redirectToProxy(w http.ResponseWriter, r *http.Request, secret string, target *downloads.FileTarget, userID int, profileID string) (bool, error) {
if target == nil || strings.TrimSpace(target.Path) == "" {
return false, fmt.Errorf("download target is empty")
}
if !target.ProxyEligible {
return false, nil
}
reservationKey := target.DownloadID
if reservationKey == "" {
reservationKey = fmt.Sprintf("direct-%d-%d", userID, target.MediaFileID)
}
sessionID := fmt.Sprintf("download-%s-%d", reservationKey, time.Now().UnixNano())
plan := h.nodePlanner.PlanDownload(sessionID)
if plan.ProxyNode == nil {
return false, nil
}
releaseReservation := func() { h.nodePlanner.ReleaseSession(sessionID) }
token, err := streamtoken.Sign(streamtoken.Claims{
SessionID: sessionID,
MediaPath: target.Path,
PlayMethod: streamtoken.PlayMethodDownload,
UserID: userID,
ProfileID: profileID,
MediaFileID: target.MediaFileID,
}, secret, proxyDownloadTokenTTL)
if err != nil {
releaseReservation()
return false, fmt.Errorf("sign proxy download token: %w", err)
}
location := strings.TrimRight(plan.ProxyNode.URL, "/") + "/downloads/file/" + url.PathEscape(token)
cacheKey := strings.TrimRight(plan.ProxyNode.URL, "/") + "\x00" + target.Path
if !h.proxyCanServe(r.Context(), cacheKey, location) {
releaseReservation()
return false, nil
}
// A HEAD request only proves that the proxy can read the selected path; the
// proxy deliberately does not count it as an active transfer. Release the
// provisional slot before redirecting so a probe cannot hold max_jobs
// capacity until the planner's reservation bridge expires.
if r.Method == http.MethodHead {
releaseReservation()
}
http.Redirect(w, r, location, http.StatusTemporaryRedirect)
return true, nil
}
func (h *DownloadHandler) proxyCanServe(ctx context.Context, cacheKey, location string) bool {
now := time.Now()
h.preflightMu.Lock()
for key, cached := range h.preflightCache {
if !now.Before(cached.expiresAt) {
delete(h.preflightCache, key)
}
}
if cached, ok := h.preflightCache[cacheKey]; ok && now.Before(cached.expiresAt) {
h.preflightMu.Unlock()
return cached.reachable
}
h.preflightMu.Unlock()
req, err := http.NewRequestWithContext(ctx, http.MethodHead, location, nil)
if err != nil {
return false
}
client := h.proxyHTTPClient
if client == nil {
client = http.DefaultClient
}
resp, err := client.Do(req)
reachable := false
if err != nil {
slog.DebugContext(ctx, "download proxy preflight failed; serving locally", "component", "api", "error", err)
} else {
_ = resp.Body.Close()
reachable = resp.StatusCode >= http.StatusOK && resp.StatusCode < http.StatusMultipleChoices
}
if err == nil && !reachable {
slog.DebugContext(ctx, "download proxy preflight unavailable; serving locally", "component", "api", "status", resp.StatusCode)
}
h.preflightMu.Lock()
if h.preflightCache == nil {
h.preflightCache = make(map[string]proxyPreflightResult)
}
h.preflightCache[cacheKey] = proxyPreflightResult{reachable: reachable, expiresAt: now.Add(proxyPreflightCacheTTL)}
h.preflightMu.Unlock()
return reachable
}
// requireManaged validates a managed (device-scoped) request: authentication, a
// configured service, and the device + profile identity (device_id from the
// X-Silo-Device-Id header only, never the body). On failure it writes the error
+233
View File
@@ -6,6 +6,7 @@ import (
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/go-chi/chi/v5"
@@ -14,6 +15,8 @@ import (
"github.com/Silo-Server/silo-server/internal/auth"
"github.com/Silo-Server/silo-server/internal/catalog"
"github.com/Silo-Server/silo-server/internal/downloads"
"github.com/Silo-Server/silo-server/internal/nodepool"
"github.com/Silo-Server/silo-server/internal/streamtoken"
)
// fakeDownloadService is a programmable DownloadService for handler tests. It
@@ -81,6 +84,21 @@ type identityCall struct {
downloadID string
}
type proxyDownloadService struct {
*fakeDownloadService
directTarget *downloads.FileTarget
managedTarget *downloads.FileTarget
resolveErr error
}
func (s *proxyDownloadService) ResolveDirectFile(context.Context, int, int, string, catalog.AccessFilter) (*downloads.FileTarget, error) {
return s.directTarget, s.resolveErr
}
func (s *proxyDownloadService) ResolveManagedFile(context.Context, int, string, string, string, catalog.AccessFilter) (*downloads.FileTarget, error) {
return s.managedTarget, s.resolveErr
}
func (f *fakeDownloadService) Capability(context.Context, int) (downloads.Capability, error) {
return f.capability, nil
}
@@ -283,6 +301,25 @@ func TestHandleCapability(t *testing.T) {
if len(resp.QualityPresets) != 1 || resp.QualityPresets[0] != downloads.QualityOriginal {
t.Fatalf("quality presets = %v, want [original]", resp.QualityPresets)
}
if resp.ProxyDelivery {
t.Fatal("proxy delivery advertised without a configured planner")
}
}
func TestHandleCapabilityAdvertisesAdditiveProxyDelivery(t *testing.T) {
svc := &proxyDownloadService{fakeDownloadService: &fakeDownloadService{}}
h := NewDownloadHandler(svc)
h.SetProxyDelivery(nodepool.NewPlanner(nodepool.NewProxyPool(), nodepool.NewTranscodePool()), func() string { return "secret" })
rec := httptest.NewRecorder()
h.HandleCapability(rec, downloadTestRequest(http.MethodGet, "/downloads/capability", nil, 7, "", ""))
var resp downloadCapabilityResponse
if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
}
if !resp.ProxyDelivery {
t.Fatalf("capability = %+v", resp)
}
}
func TestHandleCapabilityUnauthorized(t *testing.T) {
@@ -429,6 +466,47 @@ func TestManagedFileThreadsIdentity(t *testing.T) {
}
}
func TestManagedFileRedirectsAuthorizedArtifactToProxy(t *testing.T) {
const secret = "download-proxy-secret"
proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodHead {
t.Errorf("preflight method = %s", r.Method)
}
w.WriteHeader(http.StatusOK)
}))
defer proxy.Close()
svc := &proxyDownloadService{
fakeDownloadService: &fakeDownloadService{},
managedTarget: &downloads.FileTarget{Path: "/artifacts/movie.mp4", DownloadID: "dl1", MediaFileID: 42, ProxyEligible: true},
}
proxies := nodepool.NewProxyPool()
proxies.SetNodes([]*nodepool.Node{{URL: proxy.URL, Enabled: true, Healthy: true}})
h := NewDownloadHandler(svc)
h.SetProxyDelivery(nodepool.NewPlanner(proxies, nodepool.NewTranscodePool()), func() string { return secret })
req := withChiID(downloadTestRequest(http.MethodGet, "/downloads/dl1/file-proxy", nil, 7, "pA", "devA"), "dl1")
rec := httptest.NewRecorder()
h.HandleDownloadFileViaProxy(rec, req)
if rec.Code != http.StatusTemporaryRedirect {
t.Fatalf("status = %d, want 307 (body: %s)", rec.Code, rec.Body.String())
}
location := rec.Header().Get("Location")
prefix := proxy.URL + "/downloads/file/"
if !strings.HasPrefix(location, prefix) {
t.Fatalf("Location = %q", location)
}
claims, err := streamtoken.Verify(strings.TrimPrefix(location, prefix), secret)
if err != nil {
t.Fatal(err)
}
if claims.PlayMethod != streamtoken.PlayMethodDownload || claims.MediaPath != "/artifacts/movie.mp4" || claims.UserID != 7 || claims.ProfileID != "pA" {
t.Fatalf("claims = %+v", claims)
}
if svc.gotServe != (identityCall{}) {
t.Fatalf("local ServeFile was called: %+v", svc.gotServe)
}
}
func TestManagedListThreadsIdentity(t *testing.T) {
svc := &fakeDownloadService{}
h := NewDownloadHandler(svc)
@@ -514,6 +592,161 @@ func TestHandleDirectDownloadThreadsOriginalFormat(t *testing.T) {
}
}
func TestHandleDirectDownloadViaProxyRedirectsToProxy(t *testing.T) {
const secret = "download-proxy-secret"
maxJobs := 1
proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusOK)
}))
defer proxy.Close()
svc := &proxyDownloadService{
fakeDownloadService: &fakeDownloadService{},
directTarget: &downloads.FileTarget{Path: "/media/movie.mkv", MediaFileID: 42, ProxyEligible: true},
}
proxies := nodepool.NewProxyPool()
proxies.SetNodes([]*nodepool.Node{{URL: proxy.URL + "/", Enabled: true, Healthy: true, MaxJobs: &maxJobs}})
planner := nodepool.NewPlanner(proxies, nodepool.NewTranscodePool())
h := NewDownloadHandler(svc)
h.SetProxyDelivery(planner, func() string { return secret })
rec := httptest.NewRecorder()
h.HandleDirectDownloadViaProxy(rec, downloadTestRequest(http.MethodHead, "/direct-download-proxy?file_id=42", nil, 7, "", ""))
if rec.Code != http.StatusTemporaryRedirect {
t.Fatalf("status = %d, want 307 (body: %s)", rec.Code, rec.Body.String())
}
location := rec.Header().Get("Location")
claims, err := streamtoken.Verify(strings.TrimPrefix(location, proxy.URL+"/downloads/file/"), secret)
if err != nil {
t.Fatal(err)
}
if claims.MediaPath != "/media/movie.mkv" || claims.MediaFileID != 42 {
t.Fatalf("claims = %+v", claims)
}
if svc.gotDirectFileID != 0 {
t.Fatalf("local ServeDirect called for file %d", svc.gotDirectFileID)
}
// HEAD never becomes an active proxy transfer, so it must return the only
// available slot immediately rather than waiting for reservation expiry.
if plan := planner.PlanDownload("after-head"); plan.ProxyNode == nil {
t.Fatal("HEAD request retained the proxy's only job slot")
}
planner.ReleaseSession("after-head")
}
func TestHandleDirectDownloadFallsBackWithoutHealthyProxy(t *testing.T) {
svc := &proxyDownloadService{
fakeDownloadService: &fakeDownloadService{},
directTarget: &downloads.FileTarget{Path: "/media/movie.mkv", MediaFileID: 42, ProxyEligible: true},
}
proxies := nodepool.NewProxyPool()
proxies.SetNodes([]*nodepool.Node{{URL: "https://proxy.example", Enabled: true, Healthy: false}})
h := NewDownloadHandler(svc)
h.SetProxyDelivery(nodepool.NewPlanner(proxies, nodepool.NewTranscodePool()), func() string { return "secret" })
rec := httptest.NewRecorder()
h.HandleDirectDownloadViaProxy(rec, downloadTestRequest(http.MethodGet, "/direct-download-proxy?file_id=42", nil, 7, "", ""))
if rec.Code != http.StatusOK || svc.gotDirectFileID != 42 {
t.Fatalf("status = %d file = %d body = %s", rec.Code, svc.gotDirectFileID, rec.Body.String())
}
}
func TestHandleDirectDownloadFallsBackWhenProxyCannotServePath(t *testing.T) {
proxy := httptest.NewServer(http.NotFoundHandler())
defer proxy.Close()
svc := &proxyDownloadService{
fakeDownloadService: &fakeDownloadService{},
directTarget: &downloads.FileTarget{Path: "/media/movie.mkv", MediaFileID: 42, ProxyEligible: true},
}
proxies := nodepool.NewProxyPool()
proxies.SetNodes([]*nodepool.Node{{URL: proxy.URL, Enabled: true, Healthy: true}})
h := NewDownloadHandler(svc)
h.SetProxyDelivery(nodepool.NewPlanner(proxies, nodepool.NewTranscodePool()), func() string { return "secret" })
rec := httptest.NewRecorder()
h.HandleDirectDownloadViaProxy(rec, downloadTestRequest(http.MethodGet, "/direct-download-proxy?file_id=42", nil, 7, "", ""))
if rec.Code != http.StatusOK || svc.gotDirectFileID != 42 {
t.Fatalf("status = %d file = %d body = %s", rec.Code, svc.gotDirectFileID, rec.Body.String())
}
}
func TestHandleDirectDownloadEstablishedRouteKeepsLocalSuccessStatus(t *testing.T) {
proxyRequests := 0
proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
proxyRequests++
w.WriteHeader(http.StatusOK)
}))
defer proxy.Close()
svc := &proxyDownloadService{
fakeDownloadService: &fakeDownloadService{},
directTarget: &downloads.FileTarget{Path: "/media/movie.mkv", MediaFileID: 42, ProxyEligible: true},
}
proxies := nodepool.NewProxyPool()
proxies.SetNodes([]*nodepool.Node{{URL: proxy.URL, Enabled: true, Healthy: true}})
h := NewDownloadHandler(svc)
h.SetProxyDelivery(nodepool.NewPlanner(proxies, nodepool.NewTranscodePool()), func() string { return "secret" })
rec := httptest.NewRecorder()
h.HandleDirectDownload(rec, downloadTestRequest(http.MethodGet, "/direct-download?file_id=42", nil, 7, "", ""))
if rec.Code != http.StatusOK || svc.gotDirectFileID != 42 || proxyRequests != 0 {
t.Fatalf("status = %d file = %d proxy requests = %d", rec.Code, svc.gotDirectFileID, proxyRequests)
}
}
func TestHandleDirectDownloadViaProxyStaysLocalWhenTargetIneligible(t *testing.T) {
proxyRequests := 0
proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
proxyRequests++
w.WriteHeader(http.StatusOK)
}))
defer proxy.Close()
svc := &proxyDownloadService{
fakeDownloadService: &fakeDownloadService{},
directTarget: &downloads.FileTarget{Path: "/media/movie.mkv", MediaFileID: 42, ProxyEligible: false},
}
proxies := nodepool.NewProxyPool()
proxies.SetNodes([]*nodepool.Node{{URL: proxy.URL, Enabled: true, Healthy: true}})
h := NewDownloadHandler(svc)
h.SetProxyDelivery(nodepool.NewPlanner(proxies, nodepool.NewTranscodePool()), func() string { return "secret" })
rec := httptest.NewRecorder()
h.HandleDirectDownloadViaProxy(rec, downloadTestRequest(http.MethodGet, "/direct-download-proxy?file_id=42", nil, 7, "", ""))
if rec.Code != http.StatusOK || svc.gotDirectFileID != 42 || proxyRequests != 0 {
t.Fatalf("status = %d file = %d proxy requests = %d", rec.Code, svc.gotDirectFileID, proxyRequests)
}
}
func TestHandleDirectDownloadViaProxyMapsResolverError(t *testing.T) {
svc := &proxyDownloadService{fakeDownloadService: &fakeDownloadService{}, resolveErr: catalog.ErrItemNotFound}
h := NewDownloadHandler(svc)
h.SetProxyDelivery(nodepool.NewPlanner(nodepool.NewProxyPool(), nodepool.NewTranscodePool()), func() string { return "secret" })
rec := httptest.NewRecorder()
h.HandleDirectDownloadViaProxy(rec, downloadTestRequest(http.MethodGet, "/direct-download-proxy?file_id=42", nil, 7, "", ""))
if rec.Code != http.StatusNotFound {
t.Fatalf("status = %d, body = %s", rec.Code, rec.Body.String())
}
}
func TestProxyPreflightCachesReachabilityByNodeAndPath(t *testing.T) {
requests := 0
proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
requests++
w.WriteHeader(http.StatusOK)
}))
defer proxy.Close()
h := NewDownloadHandler(&fakeDownloadService{})
location := proxy.URL + "/downloads/file/token"
if !h.proxyCanServe(context.Background(), proxy.URL+"\x00/media/movie.mkv", location) {
t.Fatal("first preflight failed")
}
if !h.proxyCanServe(context.Background(), proxy.URL+"\x00/media/movie.mkv", location) {
t.Fatal("cached preflight failed")
}
if requests != 1 {
t.Fatalf("preflight requests = %d, want 1", requests)
}
}
func TestHandleDirectDownloadMissingFileID(t *testing.T) {
h := NewDownloadHandler(&fakeDownloadService{})
rec := httptest.NewRecorder()
+13
View File
@@ -1671,6 +1671,15 @@ func NewRouter(deps Dependencies) chi.Router {
// background worker.
downloadSvc.SetSubscriptions(downloads.NewSubscriptionRepository(deps.DB))
downloadHandler = handlers.NewDownloadHandler(downloadSvc)
if deps.NodePlanner != nil {
downloadHandler.SetProxyDelivery(deps.NodePlanner, func() string {
cfg := deps.CurrentConfig()
if cfg == nil {
return ""
}
return cfg.Auth.JWTSecret
})
}
if profileHandler != nil {
// Profiles may live outside Postgres (sqlite userdb backend), so
// deleting one cannot FK-cascade the shared user_devices table;
@@ -2672,12 +2681,16 @@ func NewRouter(deps Dependencies) chi.Router {
// HEAD natively.
r.Get("/{id}/file", downloadHandler.HandleDownloadFile)
r.Head("/{id}/file", downloadHandler.HandleDownloadFile)
r.Get("/{id}/file-proxy", downloadHandler.HandleDownloadFileViaProxy)
r.Head("/{id}/file-proxy", downloadHandler.HandleDownloadFileViaProxy)
r.Get("/{id}/manifest", downloadHandler.HandleManifest)
r.Get("/{id}/artwork/{kind}", downloadHandler.HandleArtwork)
r.Get("/{id}/subtitles/{ref}", downloadHandler.HandleSubtitle)
})
r.Get("/direct-download", downloadHandler.HandleDirectDownload)
r.Head("/direct-download", downloadHandler.HandleDirectDownload)
r.Get("/direct-download-proxy", downloadHandler.HandleDirectDownloadViaProxy)
r.Head("/direct-download-proxy", downloadHandler.HandleDirectDownloadViaProxy)
// Recipe gallery catalog (no profile required — purely static metadata).
recipeHandler := &handlers.RecipeHandler{}
+15 -1
View File
@@ -3,6 +3,7 @@ package config
import (
"errors"
"fmt"
"path/filepath"
"regexp"
"sort"
"strconv"
@@ -408,9 +409,22 @@ var defaultJellyfinCompatServerID = uuid.NewSHA1(
).String()
// DefaultTranscodeDir is the fallback playback.transcode_dir; download
// artifacts default to a sibling directory (see downloads.effectiveArtifactDir).
// artifacts default to a sibling directory (see EffectiveDownloadArtifactDir).
const DefaultTranscodeDir = "/tmp/silo-transcode"
// EffectiveDownloadArtifactDir resolves the shared prepared-download artifact
// root. Keeping this path rule in config lets API and transcode-node processes
// independently derive the same destination from the same live settings.
func EffectiveDownloadArtifactDir(artifactDir, transcodeDir string) string {
if artifactDir != "" {
return artifactDir
}
if transcodeDir == "" {
transcodeDir = DefaultTranscodeDir
}
return filepath.Join(filepath.Dir(transcodeDir), "silo-download-artifacts")
}
const DefaultJellyfinCompatEmulatedServerVersion = "10.12.0"
const DefaultJellyfinWebVersion = "10.11.6"
const DefaultJellyfinWebInstallDir = "/var/lib/silo/compat/jellyfin-web"
+7 -1
View File
@@ -3,7 +3,9 @@ package config
import (
"fmt"
"log/slog"
"path/filepath"
"strconv"
"strings"
"time"
"github.com/google/uuid"
@@ -603,8 +605,12 @@ func LoadFromDB(m map[string]string) (*Config, error) {
if artifactMaxBytes < 0 {
return nil, fmt.Errorf("invalid value for %q: must be non-negative", "download.artifact_max_bytes")
}
artifactDir := strings.TrimSpace(stringOr(m, "download.artifact_dir", ""))
if artifactDir != "" && !filepath.IsAbs(artifactDir) {
return nil, fmt.Errorf("invalid value for %q: must be an absolute path", "download.artifact_dir")
}
cfg.Download.TranscodeEnabled = downloadTranscodeEnabled
cfg.Download.ArtifactDir = stringOr(m, "download.artifact_dir", "")
cfg.Download.ArtifactDir = artifactDir
cfg.Download.MaxConcurrentPrepares = maxConcurrentPrepares
cfg.Download.ArtifactMaxBytes = artifactMaxBytes
+15
View File
@@ -35,6 +35,21 @@ func TestLoadFromDBMetadataPresignExpiryRejectsInvalidDuration(t *testing.T) {
}
}
func TestLoadFromDBDownloadArtifactDirRequiresAbsolutePath(t *testing.T) {
cfg, err := LoadFromDB(map[string]string{"download.artifact_dir": "/mnt/silo-downloads"})
if err != nil {
t.Fatalf("LoadFromDB() with absolute artifact dir: %v", err)
}
if cfg.Download.ArtifactDir != "/mnt/silo-downloads" {
t.Fatalf("artifact dir = %q", cfg.Download.ArtifactDir)
}
_, err = LoadFromDB(map[string]string{"download.artifact_dir": "relative/downloads"})
if err == nil || !strings.Contains(err.Error(), "download.artifact_dir") {
t.Fatalf("relative artifact dir error = %v", err)
}
}
func TestLoadFromDBJellyfinWebEnabledDefaultsToTrue(t *testing.T) {
cfg, err := LoadFromDB(map[string]string{})
if err != nil {
+119
View File
@@ -0,0 +1,119 @@
// Package downloadprepare defines the internal API used to execute a prepared
// download artifact on a dedicated transcode node.
package downloadprepare
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"strings"
"github.com/Silo-Server/silo-server/internal/playback"
)
// Request is the environment-neutral portion of a prepared-file recipe. The
// transcode node supplies its own FFmpeg path, hardware mode, and device list.
type Request struct {
JobID string `json:"job_id"`
InputPath string `json:"input_path"`
OutputPath string `json:"output_path"`
SourceVideoCodec string `json:"source_video_codec,omitempty"`
SourceVideoProfile string `json:"source_video_profile,omitempty"`
SourceVideoBitDepth int `json:"source_video_bit_depth,omitempty"`
SoftwareVideoDecode bool `json:"software_video_decode,omitempty"`
TargetCodecVideo string `json:"target_codec_video"`
TargetCodecAudio string `json:"target_codec_audio"`
TargetResolution string `json:"target_resolution,omitempty"`
TargetBitrateKbps int `json:"target_bitrate_kbps,omitempty"`
AudioTrackIndex int `json:"audio_track_index"`
TotalDuration float64 `json:"total_duration,omitempty"`
}
// NewRequest freezes the byte-affecting recipe while deliberately omitting
// environment-specific execution settings.
func NewRequest(jobID string, opts playback.TranscodeOpts, outputPath string) Request {
return Request{
JobID: jobID,
InputPath: opts.InputPath,
OutputPath: outputPath,
SourceVideoCodec: opts.SourceVideoCodec,
SourceVideoProfile: opts.SourceVideoProfile,
SourceVideoBitDepth: opts.SourceVideoBitDepth,
SoftwareVideoDecode: opts.SoftwareVideoDecode,
TargetCodecVideo: opts.TargetCodecVideo,
TargetCodecAudio: opts.TargetCodecAudio,
TargetResolution: opts.TargetResolution,
TargetBitrateKbps: opts.TargetBitrateKbps,
AudioTrackIndex: opts.AudioTrackIndex,
TotalDuration: opts.TotalDuration,
}
}
// TranscodeOpts reconstructs the prepared-file options using the selected
// node's live execution settings.
func (r Request) TranscodeOpts(ffmpegPath, hwAccel, hwDevice string, sink playback.FFmpegLogSink) playback.TranscodeOpts {
return playback.TranscodeOpts{
InputPath: r.InputPath,
SourceVideoCodec: r.SourceVideoCodec,
SourceVideoProfile: r.SourceVideoProfile,
SourceVideoBitDepth: r.SourceVideoBitDepth,
SoftwareVideoDecode: r.SoftwareVideoDecode,
TargetCodecVideo: r.TargetCodecVideo,
TargetCodecAudio: r.TargetCodecAudio,
TargetResolution: r.TargetResolution,
TargetBitrateKbps: r.TargetBitrateKbps,
AudioTrackIndex: r.AudioTrackIndex,
SubtitleTrackIndex: -1,
FFmpegPath: ffmpegPath,
HWAccel: hwAccel,
HWDevice: hwDevice,
TotalDuration: r.TotalDuration,
NodeType: "transcode",
ExecutionMode: "download_prepare",
FFmpegLogSink: sink,
}
}
// RemotePreparer executes one request on a selected transcode node.
type RemotePreparer interface {
Prepare(ctx context.Context, nodeURL, jwtSecret string, req Request) error
}
// HTTPPreparer implements RemotePreparer over the bearer-protected node API.
// The request context is the only overall timeout because a full-file encode
// may legitimately run for hours; artifact lease loss cancels that context.
type HTTPPreparer struct {
Client *http.Client
}
func (p HTTPPreparer) Prepare(ctx context.Context, nodeURL, jwtSecret string, req Request) error {
body, err := json.Marshal(req)
if err != nil {
return fmt.Errorf("remote download prepare: marshal request: %w", err)
}
httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(nodeURL, "/")+"/downloads/prepare", bytes.NewReader(body))
if err != nil {
return fmt.Errorf("remote download prepare: build request: %w", err)
}
httpReq.Header.Set("Content-Type", "application/json")
httpReq.Header.Set("Authorization", "Bearer "+jwtSecret)
client := p.Client
if client == nil {
client = http.DefaultClient
}
resp, err := client.Do(httpReq)
if err != nil {
return fmt.Errorf("remote download prepare: request: %w", err)
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode != http.StatusOK {
message, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
return fmt.Errorf("remote download prepare: node returned %d: %s", resp.StatusCode, strings.TrimSpace(string(message)))
}
_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4096))
return nil
}
@@ -0,0 +1,46 @@
package downloadprepare
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
)
func TestHTTPPreparerSendsAuthenticatedRecipe(t *testing.T) {
var got Request
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost || r.URL.Path != "/downloads/prepare" {
t.Fatalf("request = %s %s", r.Method, r.URL.Path)
}
if auth := r.Header.Get("Authorization"); auth != "Bearer secret" {
t.Fatalf("Authorization = %q", auth)
}
if err := json.NewDecoder(r.Body).Decode(&got); err != nil {
t.Fatal(err)
}
w.WriteHeader(http.StatusOK)
}))
defer server.Close()
want := Request{JobID: "job-1", InputPath: "/media/movie.mkv", OutputPath: "/artifacts/movie.mp4", TargetCodecVideo: "h264", TargetCodecAudio: "aac"}
if err := (HTTPPreparer{}).Prepare(context.Background(), server.URL+"/", "secret", want); err != nil {
t.Fatal(err)
}
if got != want {
t.Fatalf("request = %+v, want %+v", got, want)
}
}
func TestHTTPPreparerReportsNodeFailure(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
http.Error(w, "mount unavailable", http.StatusUnprocessableEntity)
}))
defer server.Close()
err := (HTTPPreparer{}).Prepare(context.Background(), server.URL, "secret", Request{JobID: "job-2"})
if err == nil {
t.Fatal("expected remote failure")
}
}
+1 -13
View File
@@ -57,12 +57,6 @@ func paramsHash(format, container, codecVideo, codecAudio, resolution string, au
return hex.EncodeToString(sum[:])
}
// defaultTranscodeDir is config's TranscodeDir default and roots
// prepared download artifacts when neither download.artifact_dir nor the
// transcode dir is configured. Keeping it absolute avoids writing artifacts
// relative to the process working directory.
const defaultTranscodeDir = config.DefaultTranscodeDir
// effectiveArtifactDir resolves where prepared artifacts are written: the
// configured download.artifact_dir when set, otherwise a dedicated directory
// alongside the transcode dir. The result is always rooted at a real volume,
@@ -73,13 +67,7 @@ const defaultTranscodeDir = config.DefaultTranscodeDir
// transcode dir, so an artifact dir nested under it would be wiped on the next
// transcode sweep.
func effectiveArtifactDir(artifactDir, transcodeDir string) string {
if artifactDir != "" {
return artifactDir
}
if transcodeDir == "" {
transcodeDir = defaultTranscodeDir
}
return filepath.Join(filepath.Dir(transcodeDir), "silo-download-artifacts")
return config.EffectiveDownloadArtifactDir(artifactDir, transcodeDir)
}
// artifactOutputPath derives a deterministic output path from
+10 -3
View File
@@ -349,10 +349,17 @@ func (m *ArtifactManager) encodeOne(ctx context.Context, a *Artifact) {
}
}
var size int64
if fi, statErr := os.Stat(a.OutputPath); statErr == nil {
size = fi.Size()
fi, err := os.Stat(a.OutputPath)
if err != nil {
// A remote node can report success while writing to a path that is not
// actually shared with the API node. Never publish that artifact: retry
// so another node or the local fallback can produce an observable file.
msg := fmt.Sprintf("prepared artifact output unavailable: %v", err)
slog.WarnContext(ctx, "prepared artifact output unavailable", "component", "downloads", "artifact_id", a.ID, "path", a.OutputPath, "error", err)
m.failJob(ctx, a, msg)
return
}
size := fi.Size()
// Fenced on lease ownership: if we lost the lease between encode and commit,
// applied is false and the current owner is responsible for flipping links —
// do not flip them here or we would race/duplicate that owner's work.
+88
View File
@@ -0,0 +1,88 @@
package downloads
import (
"context"
"log/slog"
"os"
"path/filepath"
"strings"
"github.com/google/uuid"
"github.com/Silo-Server/silo-server/internal/config"
"github.com/Silo-Server/silo-server/internal/downloadprepare"
"github.com/Silo-Server/silo-server/internal/nodepool"
"github.com/Silo-Server/silo-server/internal/playback"
)
// NodeAwarePreparer keeps artifact queue ownership central while executing the
// expensive FFmpeg process on a healthy transcode node when capacity permits.
// Integrated installations and saturated node pools fall back to local work.
type NodeAwarePreparer struct {
local EncodePreparer
planner nodepool.TranscodeWorkPlanner
liveCfg func() *config.Config
remote downloadprepare.RemotePreparer
}
func NewNodeAwarePreparer(local EncodePreparer, planner nodepool.TranscodeWorkPlanner, liveCfg func() *config.Config) *NodeAwarePreparer {
if local == nil {
local = playbackPreparer{}
}
return &NodeAwarePreparer{
local: local,
planner: planner,
liveCfg: liveCfg,
remote: downloadprepare.HTTPPreparer{},
}
}
func (p *NodeAwarePreparer) PrepareFile(ctx context.Context, opts playback.TranscodeOpts, outputPath string) error {
cfg := p.config()
jwtSecret := ""
if cfg != nil {
jwtSecret = strings.TrimSpace(cfg.Auth.JWTSecret)
}
// The default artifact directory is node-local. Remote preparation is safe
// only when the operator has deliberately configured a shared artifact path.
if cfg == nil || jwtSecret == "" || p.remote == nil || p.planner == nil || strings.TrimSpace(cfg.Download.ArtifactDir) == "" {
return p.local.PrepareFile(ctx, opts, outputPath)
}
jobID := strings.TrimSuffix(filepath.Base(outputPath), filepath.Ext(outputPath))
node, release := p.planner.ReserveTranscodeWork("download-prepare-" + jobID)
if node == nil {
return p.local.PrepareFile(ctx, opts, outputPath)
}
slog.InfoContext(ctx, "dispatching download artifact prepare", "component", "downloads", "job_id", jobID, "node", node.URL)
// A remote attempt gets its own staging output. If the HTTP connection fails
// while the node is still winding down FFmpeg, a local fallback can safely
// use outputPath.part without the two processes writing the same file.
remoteOutputPath := outputPath + ".remote-" + uuid.NewString() + ".mp4"
defer func() { _ = os.Remove(remoteOutputPath) }()
err := p.remote.Prepare(ctx, node.URL, jwtSecret, downloadprepare.NewRequest(jobID, opts, remoteOutputPath))
release()
// PrepareFile publishes its staging output with an atomic rename, so a
// visible file is complete even if the HTTP response was lost afterward.
if _, statErr := os.Stat(remoteOutputPath); statErr == nil {
if renameErr := os.Rename(remoteOutputPath, outputPath); renameErr == nil {
return nil
} else {
err = renameErr
}
} else if err == nil {
err = statErr
}
if ctx.Err() != nil {
return ctx.Err()
}
slog.WarnContext(ctx, "remote download artifact prepare unavailable; falling back to local", "component", "downloads", "job_id", jobID, "node", node.URL, "error", err)
return p.local.PrepareFile(ctx, opts, outputPath)
}
func (p *NodeAwarePreparer) config() *config.Config {
if p == nil || p.liveCfg == nil {
return nil
}
return p.liveCfg()
}
+197
View File
@@ -0,0 +1,197 @@
package downloads
import (
"context"
"errors"
"os"
"path/filepath"
"testing"
"github.com/Silo-Server/silo-server/internal/config"
"github.com/Silo-Server/silo-server/internal/downloadprepare"
"github.com/Silo-Server/silo-server/internal/nodepool"
"github.com/Silo-Server/silo-server/internal/playback"
)
type recordingEncodePreparer struct {
calls int
}
func (p *recordingEncodePreparer) PrepareFile(context.Context, playback.TranscodeOpts, string) error {
p.calls++
return nil
}
type recordingRemotePreparer struct {
nodeURL string
secret string
request downloadprepare.Request
}
type unavailableRemotePreparer struct{}
func (unavailableRemotePreparer) Prepare(context.Context, string, string, downloadprepare.Request) error {
return os.ErrNotExist
}
type responseLostRemotePreparer struct{}
func (responseLostRemotePreparer) Prepare(_ context.Context, _, _ string, req downloadprepare.Request) error {
if err := os.WriteFile(req.OutputPath, []byte("prepared"), 0o600); err != nil {
return err
}
return context.DeadlineExceeded
}
func (p *recordingRemotePreparer) Prepare(_ context.Context, nodeURL, secret string, req downloadprepare.Request) error {
p.nodeURL = nodeURL
p.secret = secret
p.request = req
return os.WriteFile(req.OutputPath, []byte("prepared"), 0o600)
}
func TestNodeAwarePreparerUsesLeastLoadedHealthyNode(t *testing.T) {
pool := nodepool.NewTranscodePool()
pool.SetNodes([]*nodepool.Node{
{URL: "http://busy", Enabled: true, Healthy: true, ActiveJobs: 3},
{URL: "http://idle", Enabled: true, Healthy: true, ActiveJobs: 1},
{URL: "http://unhealthy", Enabled: true, Healthy: false},
})
local := &recordingEncodePreparer{}
remote := &recordingRemotePreparer{}
cfg := &config.Config{}
cfg.Auth.JWTSecret = "secret"
cfg.Download.ArtifactDir = t.TempDir()
p := NewNodeAwarePreparer(local, nodepool.NewPlanner(nodepool.NewProxyPool(), pool), func() *config.Config { return cfg })
p.remote = remote
opts := playback.TranscodeOpts{InputPath: "/media/movie.mkv", TargetCodecVideo: "h264", TargetCodecAudio: "aac"}
outputPath := filepath.Join(cfg.Download.ArtifactDir, "job-1.mp4")
if err := p.PrepareFile(context.Background(), opts, outputPath); err != nil {
t.Fatal(err)
}
if local.calls != 0 {
t.Fatalf("local calls = %d, want 0", local.calls)
}
if remote.nodeURL != "http://idle" || remote.secret != "secret" {
t.Fatalf("remote call = node %q secret %q", remote.nodeURL, remote.secret)
}
if remote.request.InputPath != opts.InputPath || remote.request.OutputPath == outputPath || filepath.Dir(remote.request.OutputPath) != filepath.Dir(outputPath) || filepath.Ext(remote.request.OutputPath) != ".mp4" {
t.Fatalf("remote request = %+v", remote.request)
}
if got, err := os.ReadFile(outputPath); err != nil || string(got) != "prepared" {
t.Fatalf("promoted output = %q, %v", got, err)
}
}
func TestNodeAwarePreparerFallsBackLocallyWithoutEligibleCapacity(t *testing.T) {
limit := 1
pool := nodepool.NewTranscodePool()
pool.SetNodes([]*nodepool.Node{{URL: "http://full", Enabled: true, Healthy: true, ActiveJobs: 1, MaxJobs: &limit}})
local := &recordingEncodePreparer{}
cfg := &config.Config{}
cfg.Auth.JWTSecret = "secret"
cfg.Download.ArtifactDir = t.TempDir()
p := NewNodeAwarePreparer(local, nodepool.NewPlanner(nodepool.NewProxyPool(), pool), func() *config.Config { return cfg })
p.remote = &recordingRemotePreparer{}
if err := p.PrepareFile(context.Background(), playback.TranscodeOpts{}, "/artifacts/job-2.mp4"); err != nil {
t.Fatal(err)
}
if local.calls != 1 {
t.Fatalf("local calls = %d, want 1", local.calls)
}
}
func TestNodeAwarePreparerFallsBackLocallyWithoutNodeCredentials(t *testing.T) {
pool := nodepool.NewTranscodePool()
pool.SetNodes([]*nodepool.Node{{URL: "http://idle", Enabled: true, Healthy: true}})
local := &recordingEncodePreparer{}
p := NewNodeAwarePreparer(local, nodepool.NewPlanner(nodepool.NewProxyPool(), pool), func() *config.Config { return &config.Config{} })
p.remote = &recordingRemotePreparer{}
if err := p.PrepareFile(context.Background(), playback.TranscodeOpts{}, "/artifacts/job-3.mp4"); err != nil {
t.Fatal(err)
}
if local.calls != 1 {
t.Fatalf("local calls = %d, want 1", local.calls)
}
}
func TestNodeAwarePreparerKeepsDefaultNodeLocalArtifactDirLocal(t *testing.T) {
pool := nodepool.NewTranscodePool()
pool.SetNodes([]*nodepool.Node{{URL: "http://idle", Enabled: true, Healthy: true}})
local := &recordingEncodePreparer{}
cfg := &config.Config{}
cfg.Auth.JWTSecret = "secret"
p := NewNodeAwarePreparer(local, nodepool.NewPlanner(nodepool.NewProxyPool(), pool), func() *config.Config { return cfg })
p.remote = &recordingRemotePreparer{}
if err := p.PrepareFile(context.Background(), playback.TranscodeOpts{}, filepath.Join(t.TempDir(), "job-local.mp4")); err != nil {
t.Fatal(err)
}
if local.calls != 1 {
t.Fatalf("local calls = %d, want 1", local.calls)
}
}
func TestNodeAwarePreparerFallsBackLocallyWhenRemoteOutputIsUnavailable(t *testing.T) {
pool := nodepool.NewTranscodePool()
pool.SetNodes([]*nodepool.Node{{URL: "http://idle", Enabled: true, Healthy: true}})
local := &recordingEncodePreparer{}
cfg := &config.Config{}
cfg.Auth.JWTSecret = "secret"
cfg.Download.ArtifactDir = t.TempDir()
p := NewNodeAwarePreparer(local, nodepool.NewPlanner(nodepool.NewProxyPool(), pool), func() *config.Config { return cfg })
p.remote = unavailableRemotePreparer{}
if err := p.PrepareFile(context.Background(), playback.TranscodeOpts{}, filepath.Join(cfg.Download.ArtifactDir, "job-4.mp4")); err != nil {
t.Fatal(err)
}
if local.calls != 1 {
t.Fatalf("local calls = %d, want 1", local.calls)
}
}
func TestNodeAwarePreparerPromotesCompletedOutputAfterResponseLoss(t *testing.T) {
pool := nodepool.NewTranscodePool()
pool.SetNodes([]*nodepool.Node{{URL: "http://idle", Enabled: true, Healthy: true}})
local := &recordingEncodePreparer{}
cfg := &config.Config{}
cfg.Auth.JWTSecret = "secret"
cfg.Download.ArtifactDir = t.TempDir()
p := NewNodeAwarePreparer(local, nodepool.NewPlanner(nodepool.NewProxyPool(), pool), func() *config.Config { return cfg })
p.remote = responseLostRemotePreparer{}
outputPath := filepath.Join(cfg.Download.ArtifactDir, "job-5.mp4")
if err := p.PrepareFile(context.Background(), playback.TranscodeOpts{}, outputPath); err != nil {
t.Fatal(err)
}
if local.calls != 0 {
t.Fatalf("local calls = %d, want 0", local.calls)
}
if got, err := os.ReadFile(outputPath); err != nil || string(got) != "prepared" {
t.Fatalf("promoted output = %q, %v", got, err)
}
}
func TestNodeAwarePreparerDoesNotFallBackAfterLeaseCancellation(t *testing.T) {
pool := nodepool.NewTranscodePool()
pool.SetNodes([]*nodepool.Node{{URL: "http://idle", Enabled: true, Healthy: true}})
local := &recordingEncodePreparer{}
cfg := &config.Config{}
cfg.Auth.JWTSecret = "secret"
cfg.Download.ArtifactDir = t.TempDir()
p := NewNodeAwarePreparer(local, nodepool.NewPlanner(nodepool.NewProxyPool(), pool), func() *config.Config { return cfg })
p.remote = unavailableRemotePreparer{}
ctx, cancel := context.WithCancel(context.Background())
cancel()
err := p.PrepareFile(ctx, playback.TranscodeOpts{}, filepath.Join(cfg.Download.ArtifactDir, "job-6.mp4"))
if !errors.Is(err, context.Canceled) {
t.Fatalf("error = %v, want context canceled", err)
}
if local.calls != 0 {
t.Fatalf("local calls = %d, want 0", local.calls)
}
}
+77 -28
View File
@@ -74,6 +74,19 @@ type Capability struct {
MonitoringModes []string
}
// FileTarget is an already-authorized source or prepared artifact that may be
// served locally or delegated to a trusted proxy node. Callers must obtain it
// from the request-scoped resolver methods below; a path alone is never an
// authorization grant.
type FileTarget struct {
Path string
DownloadID string
MediaFileID int
// ProxyEligible is true only when the target lives on storage expected to
// be mounted at the same absolute path on proxy nodes.
ProxyEligible bool
}
// Service orchestrates download permission checks, quota enforcement, quality
// policy, file resolution, and file serving for both ephemeral/account-level
// rows (DeviceID == "") and managed device-library entries (DeviceID set).
@@ -847,26 +860,37 @@ func (s *Service) List(ctx context.Context, userID int, profileID, deviceID stri
// ServeDirect validates permissions and serves a file directly for browser
// download. No persistent download record is created.
func (s *Service) ServeDirect(ctx context.Context, w http.ResponseWriter, r *http.Request, userID, fileID int, format string, filter catalog.AccessFilter) error {
if _, _, err := s.downloadConfigForUser(ctx, userID, ""); err != nil {
target, err := s.ResolveDirectFile(ctx, userID, fileID, format, filter)
if err != nil {
return err
}
return s.serveLocalFile(ctx, w, r, target.Path, userID)
}
// ResolveDirectFile authorizes a browser-style original download without
// writing response bytes. It is used by the API before minting a proxy token.
func (s *Service) ResolveDirectFile(ctx context.Context, userID, fileID int, format string, filter catalog.AccessFilter) (*FileTarget, error) {
cfg, _, err := s.downloadConfigForUser(ctx, userID, "")
if err != nil {
return nil, err
}
if format != "" && format != FormatOriginal {
return ErrFormatUnavailable
return nil, ErrFormatUnavailable
}
file, err := s.fileRepo.GetByID(ctx, fileID)
if err != nil {
return fmt.Errorf("loading media file: %w", err)
return nil, fmt.Errorf("loading media file: %w", err)
}
if file == nil || file.MissingSince != nil {
return catalog.ErrItemNotFound
return nil, catalog.ErrItemNotFound
}
if err := s.itemAccess.EnsureAccessible(ctx, file.ContentID, filter); err != nil {
return err
return nil, err
}
if !catalog.FileAllowedByAccess(file, filter) {
return catalog.ErrItemNotFound
return nil, catalog.ErrItemNotFound
}
return s.serveLocalFile(ctx, w, r, file.FilePath, userID)
return &FileTarget{Path: file.FilePath, MediaFileID: file.ID, ProxyEligible: proxyDeliveryAllowed(cfg)}, nil
}
// ServeFile serves a download's file. Managed entries (device header present)
@@ -874,15 +898,19 @@ func (s *Service) ServeDirect(ctx context.Context, w http.ResponseWriter, r *htt
// before serving; ephemeral rows keep today's queued→downloading→completed
// behavior. The ephemeral path never serves a managed row, and vice versa.
func (s *Service) ServeFile(ctx context.Context, w http.ResponseWriter, r *http.Request, userID int, profileID, deviceID, downloadID string, filter catalog.AccessFilter) error {
if deviceID != "" {
target, err := s.ResolveManagedFile(ctx, userID, profileID, deviceID, downloadID, filter)
if err != nil {
return err
}
return s.serveLocalFile(ctx, w, r, target.Path, userID)
}
// Re-check policy — admin may have disabled downloads or revoked permission.
if _, _, err := s.downloadConfigForUser(ctx, userID, deviceID); err != nil {
return err
}
if deviceID != "" {
return s.serveManaged(ctx, w, r, userID, profileID, deviceID, downloadID, filter)
}
dl, err := s.repo.GetByID(ctx, downloadID)
if err != nil {
return err
@@ -925,25 +953,31 @@ func (s *Service) ServeFile(ctx context.Context, w http.ResponseWriter, r *http.
return nil
}
// serveManaged authorizes a managed entry on (user, profile, device), re-checks
// per-profile content access (invariant 2), and streams the original source.
func (s *Service) serveManaged(ctx context.Context, w http.ResponseWriter, r *http.Request, userID int, profileID, deviceID, downloadID string, filter catalog.AccessFilter) error {
// ResolveManagedFile authorizes a managed entry on (user, profile, device),
// re-checks current policy and content access, and returns the source/artifact
// path for trusted proxy delegation.
func (s *Service) ResolveManagedFile(ctx context.Context, userID int, profileID, deviceID, downloadID string, filter catalog.AccessFilter) (*FileTarget, error) {
cfg, _, err := s.downloadConfigForUser(ctx, userID, deviceID)
if err != nil {
return nil, err
}
if profileID == "" {
return ErrProfileRequired
return nil, ErrProfileRequired
}
dl, err := s.repo.GetManagedByID(ctx, downloadID, userID, profileID, deviceID)
if err != nil {
return err
return nil, err
}
if dl.Status == StatusRevoked {
return fmt.Errorf("download is revoked: %w", ErrDownloadNotActive)
return nil, fmt.Errorf("download is revoked: %w", ErrDownloadNotActive)
}
// A download id alone never authorizes access: re-check the requesting
// profile's content/library scope before serving any bytes.
if err := s.itemAccess.EnsureAccessible(ctx, dl.ContentID, filter); err != nil {
return err
return nil, err
}
return s.serveDownloadBytes(ctx, w, r, dl, userID, filter)
proxyEligible := proxyDeliveryAllowed(cfg)
return s.resolveDownloadBytesTarget(ctx, dl, filter, proxyEligible, proxyEligible && strings.TrimSpace(cfg.ArtifactDir) != "")
}
// PatchStatus lets a client confirm a managed entry's local state
@@ -1045,20 +1079,28 @@ func (s *Service) resolveFile(ctx context.Context, req CreateRequest) (*models.M
// (catalog.FileAllowedByAccess): library scope and the profile's max playback
// quality can change after a row was registered.
func (s *Service) serveDownloadBytes(ctx context.Context, w http.ResponseWriter, r *http.Request, dl *Download, userID int, filter catalog.AccessFilter) error {
target, err := s.resolveDownloadBytesTarget(ctx, dl, filter, false, false)
if err != nil {
return err
}
return s.serveLocalFile(ctx, w, r, target.Path, userID)
}
func (s *Service) resolveDownloadBytesTarget(ctx context.Context, dl *Download, filter catalog.AccessFilter, sourceProxyEligible, preparedProxyEligible bool) (*FileTarget, error) {
file, err := s.fileRepo.GetByID(ctx, dl.MediaFileID)
if err != nil {
return fmt.Errorf("loading media file: %w", err)
return nil, fmt.Errorf("loading media file: %w", err)
}
if file == nil {
return catalog.ErrItemNotFound
return nil, catalog.ErrItemNotFound
}
if dl.Format != FormatOriginal && dl.ArtifactID != "" {
if s.artifacts == nil {
return ErrFormatUnavailable
return nil, ErrFormatUnavailable
}
artifact, err := s.artifacts.Ready(ctx, dl.ArtifactID)
if err != nil {
return err
return nil, err
}
// The quality ceiling applies to what is actually served — the prepared
// artifact's resolution, not the source's (a 720p transcode of a 4K
@@ -1068,17 +1110,24 @@ func (s *Service) serveDownloadBytes(ctx context.Context, w http.ResponseWriter,
served.Resolution = artifact.Resolution
}
if !catalog.FileAllowedByAccess(&served, filter) {
return catalog.ErrItemNotFound
return nil, catalog.ErrItemNotFound
}
return s.serveLocalFile(ctx, w, r, artifact.OutputPath, userID)
return &FileTarget{Path: artifact.OutputPath, DownloadID: dl.ID, MediaFileID: file.ID, ProxyEligible: preparedProxyEligible}, nil
}
if file.MissingSince != nil {
return catalog.ErrItemNotFound
return nil, catalog.ErrItemNotFound
}
if !catalog.FileAllowedByAccess(file, filter) {
return catalog.ErrItemNotFound
return nil, catalog.ErrItemNotFound
}
return s.serveLocalFile(ctx, w, r, file.FilePath, userID)
return &FileTarget{Path: file.FilePath, DownloadID: dl.ID, MediaFileID: file.ID, ProxyEligible: sourceProxyEligible}, nil
}
// Download bandwidth settings are aggregate guarantees enforced by the API
// process. Until proxy nodes share a distributed limiter, keep capped transfers
// local rather than silently multiplying the configured limit per proxy node.
func proxyDeliveryAllowed(cfg config.DownloadConfig) bool {
return cfg.ServerBandwidthBPS <= 0 && cfg.UserBandwidthBPS <= 0
}
func (s *Service) serveLocalFile(ctx context.Context, w http.ResponseWriter, r *http.Request, path string, userID int) error {
+19
View File
@@ -0,0 +1,19 @@
package downloads
import (
"testing"
"github.com/Silo-Server/silo-server/internal/config"
)
func TestProxyDeliveryAllowedPreservesAggregateBandwidthLimits(t *testing.T) {
if !proxyDeliveryAllowed(config.DownloadConfig{}) {
t.Fatal("uncapped downloads should be proxy eligible")
}
if proxyDeliveryAllowed(config.DownloadConfig{ServerBandwidthBPS: 1}) {
t.Fatal("server-capped downloads must stay API-local")
}
if proxyDeliveryAllowed(config.DownloadConfig{UserBandwidthBPS: 1}) {
t.Fatal("user-capped downloads must stay API-local")
}
}
+88
View File
@@ -4,6 +4,8 @@ import (
"context"
"sync"
"time"
"github.com/google/uuid"
)
// Plan is the result of a node selection for one playback session.
@@ -20,6 +22,21 @@ type SessionPlanner interface {
PlanSession(sessionID, currentTranscodeURL string, needsTranscode bool, estBitrateKbps int) Plan
}
// DownloadPlanner selects proxy nodes for unbounded file delivery. Downloads
// have no predictable bitrate, so implementations must not admit them onto a
// proxy with a configured bandwidth cap.
type DownloadPlanner interface {
PlanDownload(sessionID string) Plan
ReleaseSession(sessionID string)
}
// TranscodeWorkPlanner reserves capacity for non-streaming GPU work such as a
// prepared download. The returned release function must be called after the
// remote operation ends or falls back locally.
type TranscodeWorkPlanner interface {
ReserveTranscodeWork(workID string) (*Node, func())
}
// reservation bridges the gap between assigning a session to a node and the
// node's health reports reflecting that session.
//
@@ -91,6 +108,41 @@ func (p *Planner) PlanSession(sessionID, currentTranscodeURL string, needsTransc
return p.PlanSessionWith(sessionID, currentTranscodeURL, needsTranscode, estBitrateKbps, nil)
}
// PlanDownload picks a healthy proxy for an unbounded file transfer. A
// configured proxy bandwidth cap cannot be reserved accurately without a known
// transfer rate, so capped proxies are excluded instead of being oversubscribed
// during the egress meter's convergence window.
func (p *Planner) PlanDownload(sessionID string) Plan {
if p == nil || p.proxies == nil || sessionID == "" {
return Plan{}
}
p.mu.Lock()
defer p.mu.Unlock()
now := p.now()
p.pruneReservations(now)
delete(p.reserved, sessionID)
var candidates []*Node
for _, node := range p.proxies.Nodes() {
if node == nil || !node.Enabled || !node.Healthy || !p.underCap(node, now) {
continue
}
if node.MaxBandwidthKbps != nil && *node.MaxBandwidthKbps > 0 {
continue
}
candidates = append(candidates, node)
}
if len(candidates) == 0 {
return Plan{}
}
const rrKey = "download"
proxy := candidates[p.rr[rrKey]%len(candidates)]
p.rr[rrKey]++
p.reserved[sessionID] = &reservation{proxyURL: proxy.URL, createdAt: now}
return Plan{ProxyNode: proxy}
}
// PlanSessionWith behaves like PlanSession but restricts transcode-node
// selection to nodes accepted by eligible (nil accepts every node). Capability
// -aware playback planning uses it so a recipe that only some pooled nodes
@@ -180,6 +232,42 @@ func (p *Planner) ReleaseSession(sessionID string) {
p.mu.Unlock()
}
// ReserveTranscodeWork selects the least-loaded healthy transcode node while
// sharing the same health-bridging reservation accounting as playback. Unlike
// a playback session it does not require a proxy partner: the completed file
// is written to the configured shared artifact store and served later.
func (p *Planner) ReserveTranscodeWork(workID string) (*Node, func()) {
if p == nil || p.transcodes == nil || workID == "" {
return nil, func() {}
}
p.mu.Lock()
now := p.now()
p.pruneReservations(now)
reservationID := workID + "-" + uuid.NewString()
var best *Node
for _, node := range p.transcodes.Nodes() {
if node == nil || !node.Enabled || !node.Healthy || !p.underCap(node, now) {
continue
}
if best == nil || p.effectiveJobs(node, now) < p.effectiveJobs(best, now) {
best = node
}
}
if best != nil {
p.reserved[reservationID] = &reservation{transcodeURL: best.URL, createdAt: now}
}
p.mu.Unlock()
if best == nil {
return nil, func() {}
}
var once sync.Once
return best, func() {
once.Do(func() { p.ReleaseSession(reservationID) })
}
}
// groupHealth reports, for every group label present in either pool, whether
// all of its enabled members are healthy. Pools only hold enabled nodes, so
// disabled nodes never count against a group.
+81
View File
@@ -509,3 +509,84 @@ func TestUnknownBitrateAdmittedBelowCap(t *testing.T) {
t.Fatalf("unknown-bitrate stream should be rejected at cap, got %+v", got)
}
}
func TestReserveTranscodeWorkSharesCapacityReservations(t *testing.T) {
capOne := 1
transcodes := NewTranscodePool()
transcodes.SetNodes([]*Node{
{URL: "http://transcode-a", Enabled: true, Healthy: true, MaxJobs: &capOne},
{URL: "http://transcode-b", Enabled: true, Healthy: true, MaxJobs: &capOne},
})
planner := NewPlanner(NewProxyPool(), transcodes)
first, releaseFirst := planner.ReserveTranscodeWork("download-1")
if first == nil || first.URL != "http://transcode-a" {
t.Fatalf("first node = %+v", first)
}
second, releaseSecond := planner.ReserveTranscodeWork("download-2")
if second == nil || second.URL != "http://transcode-b" {
t.Fatalf("second node = %+v", second)
}
if third, _ := planner.ReserveTranscodeWork("download-3"); third != nil {
t.Fatalf("third node = %+v, want no capacity", third)
}
releaseFirst()
third, releaseThird := planner.ReserveTranscodeWork("download-3")
if third == nil || third.URL != "http://transcode-a" {
t.Fatalf("third node after release = %+v", third)
}
releaseSecond()
releaseThird()
}
func TestReserveTranscodeWorkOverlappingAttemptsReleaseOnlyTheirOwnReservation(t *testing.T) {
capTwo := 2
transcodes := NewTranscodePool()
transcodes.SetNodes([]*Node{{URL: "http://transcode-a", Enabled: true, Healthy: true, MaxJobs: &capTwo}})
planner := NewPlanner(NewProxyPool(), transcodes)
first, releaseFirst := planner.ReserveTranscodeWork("same-artifact")
second, releaseSecond := planner.ReserveTranscodeWork("same-artifact")
if first == nil || second == nil {
t.Fatalf("overlapping reservations = first %+v second %+v", first, second)
}
if third, _ := planner.ReserveTranscodeWork("third-attempt"); third != nil {
t.Fatalf("third reservation = %+v, want full node", third)
}
releaseFirst()
third, releaseThird := planner.ReserveTranscodeWork("third-attempt")
if third == nil {
t.Fatal("first release did not free its own reservation")
}
if fourth, _ := planner.ReserveTranscodeWork("fourth-attempt"); fourth != nil {
t.Fatalf("second overlapping reservation was lost: fourth = %+v", fourth)
}
releaseSecond()
releaseThird()
}
func TestPlanDownloadSkipsBandwidthCappedProxiesAndReservesJobCapacity(t *testing.T) {
bandwidthCap := 100_000
jobCap := 1
proxies := NewProxyPool()
proxies.SetNodes([]*Node{
{URL: "http://bandwidth-capped", Enabled: true, Healthy: true, MaxBandwidthKbps: &bandwidthCap},
{URL: "http://uncapped", Enabled: true, Healthy: true, MaxJobs: &jobCap},
})
planner := NewPlanner(proxies, NewTranscodePool())
first := planner.PlanDownload("download-1")
if first.ProxyNode == nil || first.ProxyNode.URL != "http://uncapped" {
t.Fatalf("first plan = %+v", first)
}
if second := planner.PlanDownload("download-2"); second.ProxyNode != nil {
t.Fatalf("second plan = %+v, want job-cap rejection", second)
}
planner.ReleaseSession("download-1")
if second := planner.PlanDownload("download-2"); second.ProxyNode == nil {
t.Fatal("download was not admitted after reservation release")
}
}
+1 -1
View File
@@ -26,7 +26,7 @@ type SessionInfo struct {
UserID string `json:"user_id,omitempty"`
MediaItemID string `json:"media_item_id,omitempty"`
MediaTitle string `json:"media_title,omitempty"`
Type string `json:"type"` // "direct_play", "remux", "transcode"
Type string `json:"type"` // "direct_play", "remux", "transcode", "download_prepare", "download"
CodecVideo string `json:"codec_video,omitempty"`
CodecAudio string `json:"codec_audio,omitempty"`
Resolution string `json:"resolution,omitempty"`
+2 -2
View File
@@ -70,7 +70,7 @@ func PrepareFile(ctx context.Context, opts TranscodeOpts, outputPath string) err
// Resolve a multi-device hw_device list to one concrete GPU for this
// encode. Run blocks until ffmpeg exits, so the deferred release fires at
// exactly the process-exit boundary.
opts.HWAccel = resolveEffectiveTranscodeHWAccel(opts)
opts = normalizeTranscodeOpts(opts)
hwDevice, releaseHWDevice := AcquireHWDevice(opts.HWDevice, opts.HWAccel)
opts.HWDevice = hwDevice
defer releaseHWDevice()
@@ -106,7 +106,7 @@ func PrepareFile(ctx context.Context, opts TranscodeOpts, outputPath string) err
// faststart MP4 instead of HLS segments. Full-file output needs no seek or
// segment-boundary keyframes.
func buildPrepareFileArgs(opts TranscodeOpts, outputPath string) []string {
opts.HWAccel = resolveEffectiveTranscodeHWAccel(opts)
opts = normalizeTranscodeOpts(opts)
isVideoCopy := opts.TargetCodecVideo == "copy"
isAudioCopy := opts.TargetCodecAudio == "copy"
+55
View File
@@ -55,6 +55,61 @@ func TestBuildPrepareFileArgsEmitsFaststartMP4(t *testing.T) {
}
}
func TestBuildPrepareFileArgsSharesHigh10DecodeFallback(t *testing.T) {
tests := []struct {
name string
hwAccel string
want []string
forbidden []string
}{
{
name: "qsv keeps hardware encode with software decode upload",
hwAccel: "qsv",
want: []string{"-c:v h264_qsv", "format=nv12,hwupload,hwmap=derive_device=qsv"},
forbidden: []string{"-hwaccel qsv", "-hwaccel vaapi"},
},
{
name: "vaapi keeps hardware encode with software decode upload",
hwAccel: "vaapi",
want: []string{"-c:v h264_vaapi", "scale=-2:720,format=nv12,hwupload"},
forbidden: []string{"-hwaccel vaapi"},
},
{
name: "nvenc falls back to software encode",
hwAccel: "nvenc",
want: []string{"-c:v libx264", "-vf scale=-2:720"},
forbidden: []string{"-hwaccel cuda", "h264_nvenc", "scale_cuda"},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
args := buildPrepareFileArgs(TranscodeOpts{
InputPath: "/media/high10.mkv",
SourceVideoCodec: "h264",
SourceVideoProfile: "High 10",
SourceVideoBitDepth: 10,
TargetCodecVideo: "h264",
TargetCodecAudio: "aac",
TargetResolution: "720p",
HWAccel: tt.hwAccel,
AudioTrackIndex: -1,
}, "/artifacts/out.mp4")
joined := strings.Join(args, " ")
for _, want := range tt.want {
if !strings.Contains(joined, want) {
t.Fatalf("args missing %q: %s", want, joined)
}
}
for _, forbidden := range tt.forbidden {
if strings.Contains(joined, forbidden) {
t.Fatalf("args unexpectedly contain %q: %s", forbidden, joined)
}
}
})
}
}
func TestResolvePrepareTarget(t *testing.T) {
settings := AdminSettings{TranscodeEnabled: true, Allow4KTranscode: true}
file := &models.MediaFile{CodecVideo: "h264", CodecAudio: "dts", Container: "mkv", Resolution: "1080p"}
+12 -6
View File
@@ -217,8 +217,7 @@ func StartTranscode(ctx context.Context, opts TranscodeOpts) (*TranscodeSession,
if opts.SegmentDuration <= 0 {
opts.SegmentDuration = defaultSegmentDuration
}
opts = resolveSoftwareVideoDecode(opts)
opts.HWAccel = resolveEffectiveTranscodeHWAccel(opts)
opts = normalizeTranscodeOpts(opts)
configuredHWDevices := ParseHWDeviceSet(opts.HWDevice)
reserveHWDeviceOnRestart := configuredHWDevices.Multi() && hwAccelBalancesRenderDevices(opts.HWAccel)
// Resolve a multi-device hw_device list to one concrete GPU. Restarts reuse
@@ -372,12 +371,19 @@ func resolveSoftwareVideoDecode(opts TranscodeOpts) TranscodeOpts {
return opts
}
// normalizeTranscodeOpts resolves source-specific decode safety and the
// configured hardware execution mode in one place. Every FFmpeg entry point
// must pass through this helper so streaming and prepared-file recipes cannot
// disagree about whether a source may use hardware decode or encode.
func normalizeTranscodeOpts(opts TranscodeOpts) TranscodeOpts {
opts = resolveSoftwareVideoDecode(opts)
opts.HWAccel = resolveEffectiveTranscodeHWAccel(opts)
return opts
}
// buildFFmpegArgs constructs the full ffmpeg argument list from TranscodeOpts.
func buildFFmpegArgs(opts TranscodeOpts) []string {
opts = resolveSoftwareVideoDecode(opts)
// Resolve "auto" into a concrete accel method once so all downstream
// helpers (appendHWAccelArgs, appendVideoArgs, etc.) see the real value.
opts.HWAccel = resolveEffectiveTranscodeHWAccel(opts)
opts = normalizeTranscodeOpts(opts)
isVideoCopy := opts.TargetCodecVideo == "copy"
isAudioCopy := opts.TargetCodecAudio == "copy"
+118
View File
@@ -0,0 +1,118 @@
package proxy
import (
"io"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"testing"
"time"
"github.com/Silo-Server/silo-server/internal/config"
"github.com/Silo-Server/silo-server/internal/nodeconfig"
"github.com/Silo-Server/silo-server/internal/streamtoken"
)
func newDownloadProxyServer(t *testing.T, secret string) *Server {
t.Helper()
w := nodeconfig.NewWatcher(nil, nil, nil, nodeconfig.BootstrapOverrides{})
cfg := &config.Config{}
cfg.Auth.JWTSecret = secret
w.SetConfigForTest(cfg)
return NewServer(w, nil)
}
func TestProxyDownloadServesAuthorizedRange(t *testing.T) {
const secret = "download-proxy-secret"
dir := t.TempDir()
path := filepath.Join(dir, "prepared movie.mp4")
if err := os.WriteFile(path, []byte("0123456789"), 0o600); err != nil {
t.Fatal(err)
}
token, err := streamtoken.Sign(streamtoken.Claims{
SessionID: "download-1",
MediaPath: path,
PlayMethod: streamtoken.PlayMethodDownload,
UserID: 7,
ProfileID: "profile-1",
MediaFileID: 42,
}, secret, time.Minute)
if err != nil {
t.Fatal(err)
}
req := httptest.NewRequest(http.MethodGet, "/downloads/file/"+token, nil)
req.Header.Set("Range", "bytes=2-5")
rr := httptest.NewRecorder()
newDownloadProxyServer(t, secret).Handler().ServeHTTP(rr, req)
if rr.Code != http.StatusPartialContent {
t.Fatalf("status = %d, body = %s", rr.Code, rr.Body.String())
}
if rr.Body.String() != "2345" {
t.Fatalf("body = %q, want %q", rr.Body.String(), "2345")
}
if got := rr.Header().Get("Content-Disposition"); !strings.Contains(got, "prepared movie.mp4") {
t.Fatalf("Content-Disposition = %q", got)
}
}
func TestProxyDownloadRejectsPlaybackToken(t *testing.T) {
const secret = "download-proxy-secret"
token, err := streamtoken.Sign(streamtoken.Claims{
SessionID: "playback-1",
MediaPath: "/media/movie.mkv",
PlayMethod: "direct",
}, secret, time.Minute)
if err != nil {
t.Fatal(err)
}
req := httptest.NewRequest(http.MethodGet, "/downloads/file/"+token, nil)
rr := httptest.NewRecorder()
newDownloadProxyServer(t, secret).Handler().ServeHTTP(rr, req)
if rr.Code != http.StatusUnauthorized {
body, _ := io.ReadAll(rr.Result().Body)
t.Fatalf("status = %d, body = %s", rr.Code, body)
}
}
func TestProxyDownloadRejectsExpiredToken(t *testing.T) {
const secret = "download-proxy-secret"
token, err := streamtoken.Sign(streamtoken.Claims{
SessionID: "download-expired",
MediaPath: "/media/movie.mkv",
PlayMethod: streamtoken.PlayMethodDownload,
}, secret, -time.Minute)
if err != nil {
t.Fatal(err)
}
req := httptest.NewRequest(http.MethodHead, "/downloads/file/"+token, nil)
rr := httptest.NewRecorder()
newDownloadProxyServer(t, secret).Handler().ServeHTTP(rr, req)
if rr.Code != http.StatusUnauthorized {
t.Fatalf("status = %d, body = %s", rr.Code, rr.Body.String())
}
}
func TestProxyDownloadReturnsNotFoundForMissingFile(t *testing.T) {
const secret = "download-proxy-secret"
token, err := streamtoken.Sign(streamtoken.Claims{
SessionID: "download-missing",
MediaPath: filepath.Join(t.TempDir(), "gone.mp4"),
PlayMethod: streamtoken.PlayMethodDownload,
}, secret, time.Minute)
if err != nil {
t.Fatal(err)
}
req := httptest.NewRequest(http.MethodGet, "/downloads/file/"+token, nil)
rr := httptest.NewRecorder()
newDownloadProxyServer(t, secret).Handler().ServeHTTP(rr, req)
if rr.Code != http.StatusNotFound {
t.Fatalf("status = %d, body = %s", rr.Code, rr.Body.String())
}
}
+76 -2
View File
@@ -1,18 +1,24 @@
package proxy
import (
"context"
"encoding/json"
"fmt"
"io"
"log/slog"
"mime"
"net/http"
"os"
"path/filepath"
"strconv"
"strings"
"sync"
"time"
"github.com/go-chi/chi/v5"
"github.com/go-chi/cors"
"github.com/Silo-Server/silo-server/internal/downloads"
"github.com/Silo-Server/silo-server/internal/nodeconfig"
"github.com/Silo-Server/silo-server/internal/nodesessions"
"github.com/Silo-Server/silo-server/internal/playback"
@@ -28,6 +34,13 @@ type Server struct {
// subCache stores full-track PGS (.sup) extracts under the transcode dir
// so repeat selections skip the whole-file ffmpeg demux.
subCache *playback.SubtitleCache
// Download limits are node-local once egress is delegated. Rebuild the
// manager when hot-reloaded settings change so existing transfers retain
// their original limiter while new transfers use the new values.
downloadBandwidthMu sync.Mutex
downloadBandwidth *downloads.BandwidthManager
downloadServerBPS int64
downloadUserBPS int64
}
// NewServer creates a new proxy server backed by a config watcher and session
@@ -73,8 +86,7 @@ func (s *Server) Handler() http.Handler {
}))
r.Get("/api/v1/health", s.handleHealth)
r.Group(func(r chi.Router) {
// Everything under /stream counts toward the node's measured
// egress bandwidth.
// Streaming and download bytes count toward the node's measured egress.
r.Use(s.meterEgress)
r.Head("/stream/direct/{token}", s.handleDirectPlay)
r.Get("/stream/direct/{token}", s.handleDirectPlay)
@@ -85,6 +97,8 @@ func (s *Server) Handler() http.Handler {
r.Get("/stream/transcode/{token}/segment/{name}", s.handleTranscodeSegment)
r.Get("/stream/subtitles/{token}/{track}/fonts", s.handleSubtitleFonts)
r.Get("/stream/subtitles/{token}/{track}", s.handleSubtitle)
r.Head("/downloads/file/{token}", s.handleDownloadFile)
r.Get("/downloads/file/{token}", s.handleDownloadFile)
})
// Admin routes — bearer-auth protected.
@@ -153,6 +167,66 @@ func (s *Server) handleDirectPlay(w http.ResponseWriter, r *http.Request) {
http.ServeFile(w, r, claims.MediaPath)
}
func (s *Server) handleDownloadFile(w http.ResponseWriter, r *http.Request) {
claims := s.verifyToken(w, r)
if claims == nil {
return
}
if claims.PlayMethod != streamtoken.PlayMethodDownload || strings.TrimSpace(claims.MediaPath) == "" {
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
// HEAD is a capability/path preflight, not an active transfer. Counting it
// would briefly consume job capacity and could make a health report retire
// the API's reservation before the client starts its GET.
if s.tracker != nil && r.Method != http.MethodHead {
info := sessionInfo(s.tracker, claims, "download")
s.tracker.Track(r.Context(), info)
defer s.tracker.Remove(context.WithoutCancel(r.Context()), claims.SessionID)
}
f, err := os.Open(claims.MediaPath)
if err != nil {
http.NotFound(w, r)
return
}
defer func() { _ = f.Close() }()
stat, err := f.Stat()
if err != nil {
http.Error(w, "download unavailable", http.StatusInternalServerError)
return
}
filename := filepath.Base(claims.MediaPath)
if disposition := mime.FormatMediaType("attachment", map[string]string{"filename": filename}); disposition != "" {
w.Header().Set("Content-Disposition", disposition)
}
w.Header().Set("Content-Type", playback.MimeFromExtension(claims.MediaPath))
reader := io.ReadSeeker(f)
if bandwidth := s.downloadBandwidthManager(); bandwidth != nil {
reader = bandwidth.ThrottledReader(r.Context(), f, claims.UserID)
}
http.ServeContent(w, r, stat.Name(), stat.ModTime(), reader)
}
func (s *Server) downloadBandwidthManager() *downloads.BandwidthManager {
cfg := s.watcher.Config()
if cfg == nil {
return nil
}
serverBPS := cfg.Download.ServerBandwidthBPS
userBPS := cfg.Download.UserBandwidthBPS
s.downloadBandwidthMu.Lock()
defer s.downloadBandwidthMu.Unlock()
if s.downloadBandwidth == nil || serverBPS != s.downloadServerBPS || userBPS != s.downloadUserBPS {
s.downloadBandwidth = downloads.NewBandwidthManager(serverBPS, userBPS)
s.downloadServerBPS = serverBPS
s.downloadUserBPS = userBPS
}
return s.downloadBandwidth
}
func (s *Server) handleRemux(w http.ResponseWriter, r *http.Request) {
claims := s.verifyToken(w, r)
if claims == nil {
+6
View File
@@ -7,6 +7,12 @@ import (
"github.com/golang-jwt/jwt/v5"
)
const (
// PlayMethodDownload identifies a token minted only after the API has
// authorized a file download. Proxy download routes reject playback tokens.
PlayMethodDownload = "download"
)
// Claims holds everything a stateless proxy or transcode node needs
// to serve a streaming session without database access.
//
+106
View File
@@ -0,0 +1,106 @@
package transcodenode
import (
"context"
"os"
"path/filepath"
"strings"
"github.com/Silo-Server/silo-server/internal/models"
)
// InputPathAuthorizer approves a local media input before a node passes it to
// FFmpeg. Implementations must reject protocol URLs and paths outside managed
// library roots.
type InputPathAuthorizer interface {
Allowed(ctx context.Context, path string) (bool, error)
}
type mediaFolderSource interface {
List(ctx context.Context) ([]*models.MediaFolder, error)
}
// MediaRootAuthorizer resolves the deployment's current library roots and
// permits only existing, absolute filesystem paths contained by those roots.
type MediaRootAuthorizer struct {
folders mediaFolderSource
}
// NewMediaRootAuthorizer creates an FFmpeg input authorizer backed by the
// authoritative media-folder repository.
func NewMediaRootAuthorizer(folders mediaFolderSource) *MediaRootAuthorizer {
return &MediaRootAuthorizer{folders: folders}
}
// Allowed reports whether path resolves inside one of the configured media
// roots. Symlinks are resolved on both sides so a link cannot escape a root.
func (a *MediaRootAuthorizer) Allowed(ctx context.Context, path string) (bool, error) {
if a == nil || a.folders == nil || !plainAbsolutePath(path) {
return false, nil
}
folders, err := a.folders.List(ctx)
if err != nil {
return false, err
}
for _, folder := range folders {
if folder == nil {
continue
}
for _, root := range folder.Paths {
if existingPathWithinRoot(root, path) {
return true, nil
}
}
}
return false, nil
}
func plainAbsolutePath(path string) bool {
path = strings.TrimSpace(path)
return path != "" && !strings.ContainsRune(path, '\x00') && filepath.IsAbs(path)
}
func existingPathWithinRoot(root, target string) bool {
if !plainAbsolutePath(root) || !plainAbsolutePath(target) {
return false
}
resolvedRoot, err := filepath.EvalSymlinks(filepath.Clean(root))
if err != nil {
return false
}
resolvedTarget, err := filepath.EvalSymlinks(filepath.Clean(target))
if err != nil {
return false
}
info, err := os.Stat(resolvedTarget)
if err != nil || !info.Mode().IsRegular() {
return false
}
return resolvedPathContained(resolvedRoot, resolvedTarget)
}
// pathWithinRoot validates a not-yet-created output by resolving the root and
// target parent. The caller creates the basename only after this check.
func pathWithinRoot(root, target string) bool {
if !plainAbsolutePath(root) || !plainAbsolutePath(target) {
return false
}
resolvedRoot, err := filepath.EvalSymlinks(filepath.Clean(root))
if err != nil {
return false
}
resolvedParent, err := filepath.EvalSymlinks(filepath.Dir(filepath.Clean(target)))
if err != nil {
return false
}
resolvedTarget := filepath.Join(resolvedParent, filepath.Base(target))
return resolvedPathContained(resolvedRoot, resolvedTarget)
}
func resolvedPathContained(root, target string) bool {
rel, err := filepath.Rel(root, target)
if err != nil {
return false
}
return rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator))
}
@@ -0,0 +1,107 @@
package transcodenode
import (
"context"
"errors"
"os"
"path/filepath"
"runtime"
"testing"
"github.com/Silo-Server/silo-server/internal/models"
)
type staticMediaFolders struct {
folders []*models.MediaFolder
err error
}
func (s staticMediaFolders) List(context.Context) ([]*models.MediaFolder, error) {
return s.folders, s.err
}
func TestMediaRootAuthorizerAllowsOnlyExistingFilesWithinLibraryRoots(t *testing.T) {
root := t.TempDir()
mediaPath := filepath.Join(root, "movie.mkv")
if err := os.WriteFile(mediaPath, []byte("media"), 0o600); err != nil {
t.Fatal(err)
}
authorizer := NewMediaRootAuthorizer(staticMediaFolders{
folders: []*models.MediaFolder{{Paths: []string{root}}},
})
allowed, err := authorizer.Allowed(context.Background(), mediaPath)
if err != nil || !allowed {
t.Fatalf("approved media path: allowed = %v, err = %v", allowed, err)
}
for _, candidate := range []string{
"relative/movie.mkv",
"http://example.test/movie.mkv",
"file:" + mediaPath,
"concat:" + mediaPath + "|" + mediaPath,
filepath.Join(t.TempDir(), "outside.mkv"),
filepath.Join(root, "missing.mkv"),
} {
allowed, err := authorizer.Allowed(context.Background(), candidate)
if err != nil {
t.Fatalf("candidate %q: %v", candidate, err)
}
if allowed {
t.Errorf("candidate %q was allowed", candidate)
}
}
}
func TestMediaRootAuthorizerRejectsSymlinkEscape(t *testing.T) {
if runtime.GOOS == "windows" {
t.Skip("symlink semantics differ on Windows")
}
root := t.TempDir()
outside := filepath.Join(t.TempDir(), "outside.mkv")
if err := os.WriteFile(outside, []byte("media"), 0o600); err != nil {
t.Fatal(err)
}
link := filepath.Join(root, "linked.mkv")
if err := os.Symlink(outside, link); err != nil {
t.Fatal(err)
}
authorizer := NewMediaRootAuthorizer(staticMediaFolders{
folders: []*models.MediaFolder{{Paths: []string{root}}},
})
allowed, err := authorizer.Allowed(context.Background(), link)
if err != nil {
t.Fatal(err)
}
if allowed {
t.Fatal("symlink escape was allowed")
}
}
func TestMediaRootAuthorizerPropagatesRepositoryErrors(t *testing.T) {
wantErr := errors.New("database unavailable")
authorizer := NewMediaRootAuthorizer(staticMediaFolders{err: wantErr})
allowed, err := authorizer.Allowed(context.Background(), "/media/movie.mkv")
if allowed || !errors.Is(err, wantErr) {
t.Fatalf("allowed = %v, err = %v", allowed, err)
}
}
func TestPathWithinRootRejectsSymlinkedOutputParent(t *testing.T) {
if runtime.GOOS == "windows" {
t.Skip("symlink semantics differ on Windows")
}
root := t.TempDir()
outside := t.TempDir()
linkedParent := filepath.Join(root, "linked")
if err := os.Symlink(outside, linkedParent); err != nil {
t.Fatal(err)
}
if pathWithinRoot(root, filepath.Join(linkedParent, "artifact.mp4")) {
t.Fatal("output under symlinked escape was accepted")
}
if !pathWithinRoot(root, filepath.Join(root, "artifact.mp4")) {
t.Fatal("direct child output was rejected")
}
}
+162 -2
View File
@@ -19,6 +19,8 @@ import (
"golang.org/x/sync/singleflight"
"github.com/Silo-Server/silo-server/internal/chapterthumbs"
"github.com/Silo-Server/silo-server/internal/config"
"github.com/Silo-Server/silo-server/internal/downloadprepare"
"github.com/Silo-Server/silo-server/internal/nodeconfig"
"github.com/Silo-Server/silo-server/internal/nodesessions"
"github.com/Silo-Server/silo-server/internal/playback"
@@ -79,11 +81,24 @@ const sessionIdleTTL = 10 * time.Minute
// sessionReapInterval is how often the idle reaper sweeps for stale jobs.
const sessionReapInterval = time.Minute
// Session tracking is monitoring-only and must never make a healthy node's
// control-plane response depend on Redis latency.
const sessionTrackingOperationTimeout = 2 * time.Second
type sessionTracker interface {
Track(context.Context, nodesessions.SessionInfo)
Remove(context.Context, string)
Cleanup(context.Context)
NodeURL() string
NodeName() string
}
// Server is the HTTP handler for transcode mode.
type Server struct {
watcher *nodeconfig.Watcher
tracker *nodesessions.Tracker
tracker sessionTracker
ffmpegSink playback.FFmpegLogSink
inputPaths InputPathAuthorizer
sessions map[string]*playback.TranscodeSession
// lastAccess records, per registered session id, when a manifest or segment
// request last touched the job (registration counts as the first access).
@@ -175,9 +190,13 @@ func (s *Server) restartSessionLocked(ctx context.Context, sessionID string, ses
// NewServer creates a new transcode server.
func NewServer(watcher *nodeconfig.Watcher, tracker *nodesessions.Tracker) *Server {
var trackerImpl sessionTracker
if tracker != nil {
trackerImpl = tracker
}
s := &Server{
watcher: watcher,
tracker: tracker,
tracker: trackerImpl,
sessions: make(map[string]*playback.TranscodeSession),
lastAccess: make(map[string]time.Time),
}
@@ -373,6 +392,12 @@ func (s *Server) SetRecipeStore(store recipeStore) {
s.recipeStore = store
}
// SetInputPathAuthorizer wires the library-root authority used by every node
// endpoint that accepts an FFmpeg input path.
func (s *Server) SetInputPathAuthorizer(authorizer InputPathAuthorizer) {
s.inputPaths = authorizer
}
// Handler returns the chi.Router with all transcode routes.
func (s *Server) Handler() http.Handler {
s.startIdleReaper()
@@ -383,6 +408,7 @@ func (s *Server) Handler() http.Handler {
r.Use(s.requireBearer)
r.Get("/hw-capabilities", s.handleHWCapabilities)
r.Post("/chapter-thumbnails/extract", s.handleChapterThumbnailExtract)
r.Post("/downloads/prepare", s.handleDownloadPrepare)
r.Post("/transcode/start", s.handleStart)
r.Delete("/transcode/{session_id}", s.handleStop)
r.Get("/transcode/{session_id}/master.m3u8", s.handleManifest)
@@ -393,6 +419,88 @@ func (s *Server) Handler() http.Handler {
return r
}
func (s *Server) handleDownloadPrepare(w http.ResponseWriter, r *http.Request) {
var req downloadprepare.Request
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 64<<10)).Decode(&req); err != nil {
http.Error(w, "invalid request body", http.StatusBadRequest)
return
}
if strings.TrimSpace(req.JobID) == "" || strings.TrimSpace(req.InputPath) == "" || strings.TrimSpace(req.OutputPath) == "" {
http.Error(w, "job_id, input_path, and output_path are required", http.StatusBadRequest)
return
}
cfg := s.watcher.Config()
if cfg == nil {
http.Error(w, "node not configured", http.StatusServiceUnavailable)
return
}
if !s.requireApprovedInputPath(w, r, req.InputPath) {
return
}
artifactRoot := config.EffectiveDownloadArtifactDir(cfg.Download.ArtifactDir, cfg.Playback.TranscodeDir)
if err := os.MkdirAll(artifactRoot, 0o755); err != nil {
http.Error(w, "artifact directory unavailable", http.StatusInternalServerError)
return
}
if !pathWithinRoot(artifactRoot, req.OutputPath) || !strings.EqualFold(filepath.Ext(req.OutputPath), ".mp4") {
http.Error(w, "output_path must be an mp4 under the configured artifact directory", http.StatusBadRequest)
return
}
jobCtx := r.Context()
s.activeJobs.Add(1)
defer s.activeJobs.Add(-1)
if s.tracker != nil {
finishTracking := s.trackDownloadPrepare(jobCtx, nodesessions.SessionInfo{
SessionID: "download-" + req.JobID,
NodeURL: s.tracker.NodeURL(),
NodeName: s.tracker.NodeName(),
Type: "download_prepare",
CodecVideo: req.TargetCodecVideo,
CodecAudio: req.TargetCodecAudio,
Resolution: req.TargetResolution,
StartedAt: time.Now().UTC().Format(time.RFC3339),
})
defer finishTracking()
}
opts := req.TranscodeOpts(cfg.Playback.FFmpegPath, cfg.Playback.HWAccel, cfg.Playback.HWDevice, s.ffmpegSink)
if err := playback.PrepareFile(jobCtx, opts, req.OutputPath); err != nil {
if jobCtx.Err() == nil {
slog.ErrorContext(jobCtx, "prepare download artifact", "component", "transcodenode", "job_id", req.JobID, "error", err)
}
http.Error(w, "failed to prepare download artifact", http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]string{"job_id": req.JobID, "status": "ready"})
}
// trackDownloadPrepare runs the monitoring lifecycle off the request path.
// One goroutine owns both operations so Remove can never overtake Track, even
// when the encode completes before Redis responds. finish only signals that
// the job ended; it never waits for either bounded Redis operation.
func (s *Server) trackDownloadPrepare(ctx context.Context, info nodesessions.SessionInfo) func() {
finished := make(chan struct{})
var finishOnce sync.Once
baseCtx := context.WithoutCancel(ctx)
go func() {
trackCtx, cancelTrack := context.WithTimeout(baseCtx, sessionTrackingOperationTimeout)
s.tracker.Track(trackCtx, info)
cancelTrack()
<-finished
removeCtx, cancelRemove := context.WithTimeout(baseCtx, sessionTrackingOperationTimeout)
s.tracker.Remove(removeCtx, info.SessionID)
cancelRemove()
}()
return func() {
finishOnce.Do(func() { close(finished) })
}
}
func (s *Server) handleHealth(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(HealthResponse{
@@ -422,8 +530,15 @@ func (s *Server) handleChapterThumbnailExtract(w http.ResponseWriter, r *http.Re
writeChapterThumbnailError(w, http.StatusBadRequest, "invalid_request", "input_path is required")
return
}
if !s.requireApprovedInputPath(w, r, req.InputPath) {
return
}
cfg := s.watcher.Config()
if cfg == nil {
writeChapterThumbnailError(w, http.StatusServiceUnavailable, "node_unavailable", "node not configured")
return
}
frame, reason, err := chapterthumbs.ExtractFrame(r.Context(), chapterthumbs.FrameExtractOptions{
InputPath: req.InputPath,
SeekSeconds: req.SeekSeconds,
@@ -456,6 +571,10 @@ func writeChapterThumbnailError(w http.ResponseWriter, status int, reason string
func (s *Server) requireBearer(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
cfg := s.watcher.Config()
if cfg == nil {
http.Error(w, "node not configured", http.StatusServiceUnavailable)
return
}
auth := r.Header.Get("Authorization")
if !strings.HasPrefix(auth, "Bearer ") || strings.TrimPrefix(auth, "Bearer ") != cfg.Auth.JWTSecret {
http.Error(w, "unauthorized", http.StatusUnauthorized)
@@ -476,8 +595,15 @@ func (s *Server) handleStart(w http.ResponseWriter, r *http.Request) {
http.Error(w, "session_id and input_path are required", http.StatusBadRequest)
return
}
if !s.requireApprovedInputPath(w, r, req.InputPath) {
return
}
cfg := s.watcher.Config()
if cfg == nil {
http.Error(w, "node not configured", http.StatusServiceUnavailable)
return
}
outputDir := filepath.Join(cfg.Playback.TranscodeDir, req.SessionID)
opts := playback.TranscodeOpts{
@@ -598,6 +724,24 @@ func (s *Server) handleStart(w http.ResponseWriter, r *http.Request) {
})
}
func (s *Server) requireApprovedInputPath(w http.ResponseWriter, r *http.Request, path string) bool {
if s.inputPaths == nil {
http.Error(w, "input path authority unavailable", http.StatusServiceUnavailable)
return false
}
allowed, err := s.inputPaths.Allowed(r.Context(), path)
if err != nil {
slog.ErrorContext(r.Context(), "authorize transcode input path", "component", "transcodenode", "error", err)
http.Error(w, "input path authority unavailable", http.StatusServiceUnavailable)
return false
}
if !allowed {
http.Error(w, "input_path must be an approved absolute media file", http.StatusBadRequest)
return false
}
return true
}
// reconstructFromToken rebuilds a transcode session this node lost to its own
// restart. The proxy forwards the client's verified stream token in the
// X-Silo-Stream-Token header; the token carries the full byte-affecting recipe
@@ -615,6 +759,9 @@ func (s *Server) reconstructFromToken(r *http.Request, sessionID string, request
return nil
}
cfg := s.watcher.Config()
if cfg == nil {
return nil
}
claims, err := streamtoken.Verify(tokenStr, cfg.Auth.JWTSecret)
if err != nil {
slog.WarnContext(r.Context(), "transcode node reconstruct: invalid stream token", "component", "transcodenode", "error", err,
@@ -676,6 +823,16 @@ func (s *Server) reconstructFromToken(r *http.Request, sessionID string, request
// single-flight in reconstructFromToken, so it is the sole writer racing to
// register sessionID. Returns nil if the spawn fails or the slot wait is canceled.
func (s *Server) spawnReconstruct(r *http.Request, sessionID string, requestedSegment int, card playback.RecipeCard) *playback.TranscodeSession {
if s.inputPaths == nil {
slog.ErrorContext(r.Context(), "transcode node reconstruct input authority unavailable", "component", "transcodenode", "session", sessionID)
return nil
}
allowed, err := s.inputPaths.Allowed(r.Context(), card.InputPath)
if err != nil || !allowed {
slog.WarnContext(r.Context(), "transcode node reconstruct input rejected", "component", "transcodenode", "session", sessionID, "error", err)
return nil
}
// Pace the cold-start burst so a node restart that loses many sessions does not
// launch every ffmpeg at once. A client that disconnects while waiting releases
// its slot rather than queueing dead work.
@@ -698,6 +855,9 @@ func (s *Server) spawnReconstruct(r *http.Request, sessionID string, requestedSe
}
cfg := s.watcher.Config()
if cfg == nil {
return nil
}
outputDir := filepath.Join(cfg.Playback.TranscodeDir, sessionID)
opts := card.TranscodeOpts(outputDir, cfg.Playback.FFmpegPath, s.ffmpegSink)
opts.SessionID = sessionID
+216 -2
View File
@@ -9,12 +9,14 @@ import (
"os"
"path/filepath"
"strings"
"sync"
"testing"
"time"
"github.com/go-chi/chi/v5"
"github.com/Silo-Server/silo-server/internal/config"
"github.com/Silo-Server/silo-server/internal/downloadprepare"
"github.com/Silo-Server/silo-server/internal/nodeconfig"
"github.com/Silo-Server/silo-server/internal/nodesessions"
"github.com/Silo-Server/silo-server/internal/playback"
@@ -23,6 +25,57 @@ import (
const testSecret = "node-reconstruct-test-secret"
type allowInputPaths struct{}
func (allowInputPaths) Allowed(context.Context, string) (bool, error) {
return true, nil
}
type blockingSessionTracker struct {
trackStarted chan struct{}
trackRelease chan struct{}
removeStarted chan struct{}
mu sync.Mutex
events []string
trackHasDeadline bool
removeHasDeadline bool
}
func newBlockingSessionTracker() *blockingSessionTracker {
return &blockingSessionTracker{
trackStarted: make(chan struct{}),
trackRelease: make(chan struct{}),
removeStarted: make(chan struct{}),
}
}
func (t *blockingSessionTracker) Track(ctx context.Context, _ nodesessions.SessionInfo) {
_, hasDeadline := ctx.Deadline()
t.mu.Lock()
t.trackHasDeadline = hasDeadline
t.events = append(t.events, "track")
t.mu.Unlock()
close(t.trackStarted)
<-t.trackRelease
t.mu.Lock()
t.events = append(t.events, "track-done")
t.mu.Unlock()
}
func (t *blockingSessionTracker) Remove(ctx context.Context, _ string) {
_, hasDeadline := ctx.Deadline()
t.mu.Lock()
t.removeHasDeadline = hasDeadline
t.events = append(t.events, "remove")
t.mu.Unlock()
close(t.removeStarted)
}
func (*blockingSessionTracker) Cleanup(context.Context) {}
func (*blockingSessionTracker) NodeURL() string { return "http://node" }
func (*blockingSessionTracker) NodeName() string { return "node" }
// newTestServer builds a transcode Server whose config carries a known JWT secret
// so reconstructFromToken can verify forwarded stream tokens. The tracker is left
// nil: the guard-rejection cases never reach the spawn/track path.
@@ -34,8 +87,9 @@ func newTestServer(t *testing.T) *Server {
cfg.Playback.TranscodeDir = t.TempDir()
w.SetConfigForTest(cfg)
return &Server{
watcher: w,
sessions: make(map[string]*playback.TranscodeSession),
watcher: w,
inputPaths: allowInputPaths{},
sessions: make(map[string]*playback.TranscodeSession),
}
}
@@ -71,6 +125,166 @@ func TestHandleStartRequireReadyRejectsExitedFFmpeg(t *testing.T) {
}
}
func TestHandleDownloadPrepareWritesConfiguredSharedArtifact(t *testing.T) {
server := newTestServer(t)
artifactDir := t.TempDir()
server.watcher.Config().Download.ArtifactDir = artifactDir
ffmpegPath := filepath.Join(t.TempDir(), "ffmpeg.sh")
if err := os.WriteFile(ffmpegPath, []byte("#!/bin/sh\nfor last; do :; done\ntouch \"$last\"\n"), 0o755); err != nil {
t.Fatal(err)
}
server.watcher.Config().Playback.FFmpegPath = ffmpegPath
server.watcher.Config().Playback.HWAccel = "none"
outputPath := filepath.Join(artifactDir, "artifact.mp4")
body, err := json.Marshal(downloadprepare.Request{
JobID: "artifact-1",
InputPath: "/media/movie.mkv",
OutputPath: outputPath,
TargetCodecVideo: "copy",
TargetCodecAudio: "copy",
AudioTrackIndex: -1,
})
if err != nil {
t.Fatal(err)
}
req := httptest.NewRequest(http.MethodPost, "/downloads/prepare", bytes.NewReader(body))
rr := httptest.NewRecorder()
server.handleDownloadPrepare(rr, req)
if rr.Code != http.StatusOK {
t.Fatalf("status = %d, body = %s", rr.Code, rr.Body.String())
}
if _, err := os.Stat(outputPath); err != nil {
t.Fatalf("prepared output: %v", err)
}
if got := server.activeJobs.Load(); got != 0 {
t.Fatalf("active jobs after prepare = %d, want 0", got)
}
}
func TestHandleDownloadPrepareTrackingDoesNotBlockAndRemovesAfterTrack(t *testing.T) {
server := newTestServer(t)
tracker := newBlockingSessionTracker()
server.tracker = tracker
artifactDir := t.TempDir()
server.watcher.Config().Download.ArtifactDir = artifactDir
ffmpegPath := filepath.Join(t.TempDir(), "ffmpeg.sh")
if err := os.WriteFile(ffmpegPath, []byte("#!/bin/sh\nfor last; do :; done\ntouch \"$last\"\n"), 0o755); err != nil {
t.Fatal(err)
}
server.watcher.Config().Playback.FFmpegPath = ffmpegPath
server.watcher.Config().Playback.HWAccel = "none"
body, err := json.Marshal(downloadprepare.Request{
JobID: "artifact-tracking",
InputPath: "/media/movie.mkv",
OutputPath: filepath.Join(artifactDir, "artifact.mp4"),
TargetCodecVideo: "copy",
TargetCodecAudio: "copy",
AudioTrackIndex: -1,
})
if err != nil {
t.Fatal(err)
}
req := httptest.NewRequest(http.MethodPost, "/downloads/prepare", bytes.NewReader(body))
rr := httptest.NewRecorder()
handlerDone := make(chan struct{})
go func() {
server.handleDownloadPrepare(rr, req)
close(handlerDone)
}()
select {
case <-tracker.trackStarted:
case <-time.After(time.Second):
t.Fatal("tracking did not start")
}
select {
case <-handlerDone:
case <-time.After(250 * time.Millisecond):
close(tracker.trackRelease)
<-handlerDone
t.Fatal("download prepare waited for session tracking")
}
if rr.Code != http.StatusOK {
close(tracker.trackRelease)
t.Fatalf("status = %d, body = %s", rr.Code, rr.Body.String())
}
close(tracker.trackRelease)
select {
case <-tracker.removeStarted:
case <-time.After(time.Second):
t.Fatal("tracking cleanup was not scheduled")
}
tracker.mu.Lock()
defer tracker.mu.Unlock()
if !tracker.trackHasDeadline || !tracker.removeHasDeadline {
t.Fatalf("tracking deadlines: track=%t remove=%t", tracker.trackHasDeadline, tracker.removeHasDeadline)
}
if got := strings.Join(tracker.events, ","); got != "track,track-done,remove" {
t.Fatalf("tracking order = %q", got)
}
}
func TestHandleDownloadPrepareRejectsOutputOutsideArtifactRoot(t *testing.T) {
server := newTestServer(t)
server.watcher.Config().Download.ArtifactDir = t.TempDir()
body, err := json.Marshal(downloadprepare.Request{
JobID: "artifact-2",
InputPath: "/media/movie.mkv",
OutputPath: filepath.Join(t.TempDir(), "escape.mp4"),
TargetCodecVideo: "h264",
TargetCodecAudio: "aac",
})
if err != nil {
t.Fatal(err)
}
req := httptest.NewRequest(http.MethodPost, "/downloads/prepare", bytes.NewReader(body))
rr := httptest.NewRecorder()
server.handleDownloadPrepare(rr, req)
if rr.Code != http.StatusBadRequest {
t.Fatalf("status = %d, body = %s", rr.Code, rr.Body.String())
}
}
func TestHandleDownloadPrepareRejectsUnavailableConfig(t *testing.T) {
server := newTestServer(t)
server.watcher.SetConfigForTest(nil)
body := []byte(`{"job_id":"artifact-3","input_path":"/media/movie.mkv","output_path":"/artifacts/movie.mp4"}`)
req := httptest.NewRequest(http.MethodPost, "/downloads/prepare", bytes.NewReader(body))
rr := httptest.NewRecorder()
server.handleDownloadPrepare(rr, req)
if rr.Code != http.StatusServiceUnavailable {
t.Fatalf("status = %d, body = %s", rr.Code, rr.Body.String())
}
}
func TestHandleStartRejectsUnapprovedInputPath(t *testing.T) {
server := newTestServer(t)
server.inputPaths = NewMediaRootAuthorizer(staticMediaFolders{})
body := []byte(`{"session_id":"unsafe-input","input_path":"http://example.test/movie.mkv"}`)
req := httptest.NewRequest(http.MethodPost, "/transcode/start", bytes.NewReader(body))
rr := httptest.NewRecorder()
server.handleStart(rr, req)
if rr.Code != http.StatusBadRequest {
t.Fatalf("status = %d, body = %s", rr.Code, rr.Body.String())
}
}
func TestDownloadPrepareRouteRequiresNodeBearer(t *testing.T) {
server := newTestServer(t)
req := httptest.NewRequest(http.MethodPost, "/downloads/prepare", strings.NewReader(`{}`))
rr := httptest.NewRecorder()
server.Handler().ServeHTTP(rr, req)
if rr.Code != http.StatusUnauthorized {
t.Fatalf("status = %d, body = %s", rr.Code, rr.Body.String())
}
}
func TestHandleStartDistinctReplacementFailurePreservesPredecessor(t *testing.T) {
server := newTestServer(t)
server.sessions["public-session"] = &playback.TranscodeSession{}