diff --git a/cmd/silo/main.go b/cmd/silo/main.go index 60515b8d..0c5e8cd5 100644 --- a/cmd/silo/main.go +++ b/cmd/silo/main.go @@ -756,6 +756,10 @@ func main() { defer func() { _ = apiRedisClient.Close() }() } + // One process-local registry is shared by every download-class serving + // surface and the admin view. It is intentionally separate from live streams. + transferMonitoring := newTransferRegistryComposition() + // Central-side kill switch: written by the async enforcer, admin terminate, // and account revocation; edges enforce it via Redis pub/sub + poll. Memory- // only when Redis is absent (single-node integrated). The Postgres durable @@ -824,6 +828,7 @@ func main() { configWatcher.RequestReload() }, } + transferMonitoring.wireAPI(&deps) accessGroupStore := access.NewGroupStore(pool) audiobooksService := audiobooks.New(&audiobooksSettingsAdapter{repo: settingsRepo}) absCompatEnabled, err := audiobooksService.ABSCompatEnabled(appCtx) @@ -2287,6 +2292,7 @@ func main() { SessionSyncer: deps.SessionSyncer, Revocation: streamRevocation, } + transferMonitoring.wireABS(&absHDeps) absH := audiobooksService.BuildABSHandler(absHDeps) deps.ABSHandler = absH } @@ -2543,6 +2549,7 @@ func main() { RecipeNodeStore: noderecipe.NewStore(apiRedisClient, 0), SessionSyncer: deps.SessionSyncer, } + transferMonitoring.wireJellycompat(&compatDeps) // Wire direct dependencies when DB is available. if deps.DB != nil { diff --git a/cmd/silo/transfer_wiring.go b/cmd/silo/transfer_wiring.go new file mode 100644 index 00000000..e437223c --- /dev/null +++ b/cmd/silo/transfer_wiring.go @@ -0,0 +1,30 @@ +package main + +import ( + "github.com/Silo-Server/silo-server/internal/api" + "github.com/Silo-Server/silo-server/internal/audiobooks" + "github.com/Silo-Server/silo-server/internal/jellycompat" + "github.com/Silo-Server/silo-server/internal/transfers" +) + +// transferRegistryComposition owns the one process-local registry and keeps +// independently constructed protocol surfaces from accidentally diverging. +type transferRegistryComposition struct { + registry *transfers.Registry +} + +func newTransferRegistryComposition() *transferRegistryComposition { + return &transferRegistryComposition{registry: transfers.New()} +} + +func (c *transferRegistryComposition) wireAPI(deps *api.Dependencies) { + deps.TransferRegistry = c.registry +} + +func (c *transferRegistryComposition) wireJellycompat(deps *jellycompat.Dependencies) { + deps.TransferRegistry = c.registry +} + +func (c *transferRegistryComposition) wireABS(deps *audiobooks.ABSHandlerDeps) { + deps.Transfers = c.registry +} diff --git a/cmd/silo/transfer_wiring_test.go b/cmd/silo/transfer_wiring_test.go new file mode 100644 index 00000000..0147944a --- /dev/null +++ b/cmd/silo/transfer_wiring_test.go @@ -0,0 +1,30 @@ +package main + +import ( + "testing" + + "github.com/Silo-Server/silo-server/internal/api" + "github.com/Silo-Server/silo-server/internal/audiobooks" + "github.com/Silo-Server/silo-server/internal/jellycompat" +) + +func TestTransferRegistryCompositionSharesOneInstance(t *testing.T) { + composition := newTransferRegistryComposition() + var apiDeps api.Dependencies + var compatDeps jellycompat.Dependencies + var absDeps audiobooks.ABSHandlerDeps + + composition.wireAPI(&apiDeps) + composition.wireJellycompat(&compatDeps) + composition.wireABS(&absDeps) + + if apiDeps.TransferRegistry == nil { + t.Fatal("API/native/admin transfer registry is nil") + } + if apiDeps.TransferRegistry != compatDeps.TransferRegistry { + t.Fatal("jellycompat received a different transfer registry") + } + if apiDeps.TransferRegistry != absDeps.Transfers { + t.Fatal("ABS received a different transfer registry") + } +} diff --git a/docs/architecture/playback-paths-monitoring-kill-matrix.md b/docs/architecture/playback-paths-monitoring-kill-matrix.md index daf590a5..e1a87e4b 100644 --- a/docs/architecture/playback-paths-monitoring-kill-matrix.md +++ b/docs/architecture/playback-paths-monitoring-kill-matrix.md @@ -84,9 +84,9 @@ bytes but withholds/falsifies progress must still be counted. | jellycompat | direct | ✅ transport-count shield (`streams.go:129`) | ✅ edge tracker, owner+route carried | | jellycompat | remux | ✅ transport-count shield | ✅ edge tracker, owner+route carried | | jellycompat | transcode | ✅ per-segment transport marker around `ServeFile` (`streams.go:543`) | ✅ edge tracker, owner+route carried | -| ABS | authenticated file/download | ⚠️ download-class exemption; no session/monitor record | same | +| ABS | authenticated file/download | ✅ separate in-memory transfer record; cap-exempt | same process only | | ABS | public playback track | ✅ native-session transport marker + metered bytes | same | -| ABS | public RSS feed file | ⚠️ public download-class pour; no session/monitor record | same | +| ABS | public RSS feed file | ✅ separate in-memory transfer record; cap-exempt | same process only | - **Existence is server-observed on every cell.** Integrated: a session is unreapable while `activeTransportCount > 0`, and every byte-serving path now @@ -139,14 +139,22 @@ Captured per live stream (`streammonitor.LiveStream` / `nodesessions.SessionInfo stream tracked by both the central manager and the edge serving it shows as ONE row — matching the enforcer's count. A `node_id` filter targets an edge, so integrated sessions appear only in the unfiltered listing. +- **Download visibility is a separate sibling collection.** The same unfiltered + admin response includes `transfers`, sourced from a bounded process-local + registry. Transfer rows contain a unique per-pour id, optional download-row + correlation id, owner/profile, media file, route, client metadata, timestamps, + and server-observed bytes. A `node_id` filter excludes them because they belong + to this API process, not an edge. The byte timestamp advances in coarse metered + chunks (1 MiB or final close), and multi-replica deployments see only the + registry of the process answering the request. - **Downloads are an intentional exemption — with asymmetries.** `/downloads/*` (native), compat `/Items/{id}/Download`, and ABS authenticated file/download - serves transfer full media with no - session/monitor record and do NOT count against the live-stream cap. The native - route requires a quota-checked download row (`internal/downloads` - concurrency/period limits); the compat and ABS routes are covered by NO quota - today. Unifying all three under one quota and one download monitor record is - explicit follow-up work. All three routes arm the shared `WatchAndCut`, so a + and RSS feed serves transfer full media without a live-stream session and do NOT + count against the live-stream cap. They now create separate, in-memory transfer + records while a pour is active. Native download quotas run only when a download + row is created; neither `ServeDownload` nor `ServeDirect` gates a pour, and + compat/ABS routes have no shared download quota. All routes arm the shared + `WatchAndCut`, so a per-user stream revocation cuts an in-flight download pour, and the same hook deletes every compat login so reconnects need re-auth. Documented at the handlers. @@ -338,12 +346,11 @@ so a killed direct-play stops on its next request at worst. native and a jellycompat single-long-GET direct-play, admin-terminate it, confirm the connection drops within ~5s rather than only on the next request. - [x] **VERIFY-2 — download routes vs the stream cap — RESOLVED (documented exemption).** - `/downloads/*` (native) and compat `/Items/{id}/Download` serve full media with no - session/monitor record; they are intentionally exempt from the live-stream cap and - governed by the separate download concurrency/period quota. Documented at both - handlers and in the monitoring data model above. (Note: they also do not consult - the kill switch — a per-user *stream* revocation does not stop an in-flight - download. If a ban should also cut downloads, guard those handlers separately.) + Native, compat, and ABS download-class routes remain intentionally exempt from + the live-stream cap but are visible in the sibling, process-local `transfers` + array. Native concurrency/period checks happen only at row creation, not at + serve time. A per-user revocation is a cutoff: it refuses/cuts older pours but + deliberately allows a new authenticated pour started after the cutoff. - [x] **VERIFY-4 — transcode buffer-ahead evasion — RESOLVED.** Unpaused transcodes use a bounded 10-minute integrated grace measured from the same activity resolution advanced by server-observed bytes/transports; paused @@ -437,8 +444,10 @@ Remaining work: enforcer re-revokes it every pass and the admin view overcounts). Needs a node-side idle sweep tied to real serve activity (`MarkServed` now provides the signal). -6. **Unified download quota and monitor:** compat `/Items/{id}/Download` and ABS - authenticated file/download routes are exempt from the stream cap and covered - by no download quota. Bring them under the native download policy with one - quota and one monitor-record model; today's controls are authentication, - library access, and the in-flight user-revocation cut. +6. **Download controls phase 2:** add per-download kill and a standing per-user + download block (a user revocation is only a cutoff and permits newer + credentials/requests). Separately unify native, compat, and ABS under a shared + download quota and rolling volume budget. Also resolve the pre-existing native + completion semantics (`ServeContent` cannot report write failure), correlate + ABS playback sessions with duplicate `abs_file_stream` transfer rows, and make + transfer visibility multi-replica rather than process-local. diff --git a/docs/architecture/stream-abuse-matrix.md b/docs/architecture/stream-abuse-matrix.md index a63d7f12..644dbaf1 100644 --- a/docs/architecture/stream-abuse-matrix.md +++ b/docs/architecture/stream-abuse-matrix.md @@ -108,11 +108,11 @@ Jellyfin-compat surface is unthrottled (see Category E). | # | Abuse story | Detection | Enforcement | One-line verdict | |---|---|---|---|---| -| 17 | Rip whole library via **native download** endpoints | ⚠️ | ⚠️ | 3-concurrent gate only; **no volume cap**; loop create→complete defeats it | -| 18 | Rip via **compat `/Items/{id}/Download`** (Infuse) | ❌ | ❌ | **No quota at all** — auth + library filter only. Widest hole | -| 18a | Rip via authenticated **ABS file/download** route | ❌ | ⚠️ | Download-class and unquota'd/unmonitored, but user cutoff refuses and cuts pours | +| 17 | Rip whole library via **native download** endpoints | ⚠️ | ⚠️ | Active pours/bytes visible; creation-time gate only; **no volume cap** | +| 18 | Rip via **compat `/Items/{id}/Download`** (Infuse) | ⚠️ | ❌ | Active pour visible, but **no quota at all** | +| 18a | Rip via authenticated **ABS file/download** route | ⚠️ | ⚠️ | Active pour visible and user-cuttable; no quota | | 18b | Pull an **ABS public playback track** | ✅ | ✅ | Native session id is monitored/metered; session and owner kills land | -| 18c | Pull a **public ABS RSS feed file** | ❌ | ⚠️ | No live record; owner cutoff uses feed creation time and cuts in-flight pours | +| 18c | Pull a **public ABS RSS feed file** | ⚠️ | ⚠️ | Active pour visible; owner cutoff uses feed creation time | | 19 | Rip via **sequential direct-play GETs** (one at a time) | ⚠️ | ❌ | Cap counts concurrency, not volume; stays at 1 forever | | 20 | Admin **stops an in-progress rip** mid-transfer | — | ⚠️ | Branch adds `WatchAndCut` on downloads (best-effort); admin must issue the revoke | @@ -254,21 +254,24 @@ everything. **Detection ❌ (at mint) / Enforcement ❌ (unless it becomes over- ### Category D — Ripping / bulk data exfiltration **D17. Rip via native download endpoints.** -Bounded only by `download.max_concurrent_per_user` (default **3**, -`internal/downloads/limiter.go`). `download.max_per_period` and both bandwidth +Download record creation is bounded by `download.max_concurrent_per_user` +(default **3**, `internal/downloads/limiter.go`). Serving and re-fetching an +existing row is not gated. `download.max_per_period` and both bandwidth knobs default to **0 = unlimited** — and the bandwidth managers are *rate* throttles (token buckets), never *volume* caps. The concurrency check runs at record creation, so a `create → complete → create` loop pulls the entire library 3 files -at a time forever. **No total-volume cap exists anywhere.** **Detection ⚠️ -(per-download rows, no cumulative signal) / Enforcement ⚠️ (weak concurrency gate).** +at a time forever. Active pours and server-observed bytes are visible through the +process-local admin `transfers` array, but disappear at completion and are not a +cumulative signal. **No total-volume cap exists anywhere.** **Detection ⚠️ / +Enforcement ⚠️ (creation-time gate only).** **D18. Rip via compat `/Items/{id}/Download`. — WIDEST HOLE.** `HandleDownload` (`jellycompat/streams.go`) requires only a compat session and the per-item library-access filter, then serves the raw file with full Range support. -**No download row, no quota, no bandwidth throttle, no session/monitor record, no -byte cap.** An Infuse-style client can GET/Range every original file back-to-back, -unlimited, and the server keeps no record beyond generic HTTP logs. The branch's own -doc comment admits this and files it as an open follow-up. **Detection ❌ / +**No download row, no quota, no bandwidth throttle, and no byte cap.** An +Infuse-style client can GET/Range every original file back-to-back. The active +pour and served bytes are visible in the process-local admin transfer registry, +but there is no durable history or automated consumer. **Detection ⚠️ / Enforcement ❌.** **D19. Rip via sequential direct-play GETs.** @@ -279,14 +282,13 @@ attributable session, but nothing correlates sequential sessions into "this user ripping." **Detection ⚠️ / Enforcement ❌.** **D20. Admin stops an in-progress rip mid-transfer.** -On `main`, impossible — `Refuse` only 403s *new* requests; an open multi-GB GET -keeps pouring. The branch adds `streamrevoke.Store.WatchAndCut` (checks on entry + +`streamrevoke.Store.WatchAndCut` (checks on entry + every 5s, forces the socket via `SetWriteDeadline`) and arms it on native downloads (`api/handlers/downloads.go`) and compat download/direct-play (`jellycompat/streams.go`). So a **user revocation now cuts an in-flight download** — best-effort (no-op if the writer chain lacks write-deadline support, then stops on next request). The admin still has to *decide* to revoke; nothing auto-detects the -rip. **Enforcement ⚠️ (branch improves `main`'s ❌ to best-effort cut).** +rip. **Enforcement ⚠️.** ### Category E — API / DB load abuse (branch is orthogonal) @@ -390,11 +392,15 @@ visible, no CPU/process signal) / Enforcement ❌ (no aggregate cap).** ## Recommended follow-ups (prioritized by exposure) -1. **Unify download-class accounting:** bring compat - `/Items/{id}/Download` and ABS `/items/{id}/file/{ino}` (including - `/download`) under the same quota and monitor record as native downloads. -2. **Add a per-user *volume* budget** (D17/D19): a rolling bytes-per-period cap that - spans downloads *and* direct-play, since concurrency caps can't bound a rip. +1. **Download controls phase 2:** add per-download kill and a standing per-user + download block. The existing user revocation is a cutoff, not a ban: it cuts + older pours but deliberately allows new ones. +2. **Unify download quotas and add a per-user *volume* budget** (D17/D19): cover + native serving, compat, and ABS with a rolling bytes-per-period cap that spans + downloads and direct-play, since concurrency caps cannot bound a rip. + Also deferred: correct native completion semantics after partial/write-failed + responses, correlate duplicate ABS playback-session and `abs_file_stream` + views, and aggregate transfer visibility across API replicas. 3. **Extend rate limiting to the compat surface + `/auth/refresh`** and add a connection cap / `LimitListener` (E25–E28). 4. **Add a per-node concurrent-transcode cap** and wire `nodepool.MaxJobs` defaults diff --git a/internal/api/handlers/downloads.go b/internal/api/handlers/downloads.go index 2e8fafc4..d65bb441 100644 --- a/internal/api/handlers/downloads.go +++ b/internal/api/handlers/downloads.go @@ -13,10 +13,13 @@ import ( apimw "github.com/Silo-Server/silo-server/internal/api/middleware" "github.com/Silo-Server/silo-server/internal/catalog" + "github.com/Silo-Server/silo-server/internal/clientip" "github.com/Silo-Server/silo-server/internal/downloads" "github.com/Silo-Server/silo-server/internal/httpstream" "github.com/Silo-Server/silo-server/internal/playback" "github.com/Silo-Server/silo-server/internal/streamrevoke" + "github.com/Silo-Server/silo-server/internal/transfers" + "github.com/google/uuid" ) // DownloadService is the interface that the download handler depends on. A @@ -33,8 +36,8 @@ type DownloadService interface { UpdateSubscription(ctx context.Context, userID int, profileID, deviceID, id string, patch downloads.SubscriptionPatch, filter catalog.AccessFilter) (*downloads.SubscriptionResult, error) DeleteSubscription(ctx context.Context, userID int, profileID, deviceID, id string) error SyncSubscriptions(ctx context.Context, userID int, profileID, deviceID string, filter catalog.AccessFilter) (int, error) - ServeDirect(ctx context.Context, w http.ResponseWriter, r *http.Request, userID, fileID int, format string, filter catalog.AccessFilter) error - ServeFile(ctx context.Context, w http.ResponseWriter, r *http.Request, userID int, profileID, deviceID, downloadID string, filter catalog.AccessFilter) error + ServeDirect(ctx context.Context, w http.ResponseWriter, r *http.Request, userID, fileID int, format string, filter catalog.AccessFilter, transfer transfers.Transfer) error + ServeFile(ctx context.Context, w http.ResponseWriter, r *http.Request, userID int, profileID, deviceID, downloadID string, filter catalog.AccessFilter, transfer transfers.Transfer) error List(ctx context.Context, userID int, profileID, deviceID string) ([]*downloads.Download, error) Delete(ctx context.Context, userID int, profileID, deviceID, downloadID string) error PatchStatus(ctx context.Context, userID int, profileID, deviceID, downloadID, status string) error @@ -48,6 +51,7 @@ type DownloadService interface { type DownloadHandler struct { svc DownloadService revocation *streamrevoke.Store + transfers *transfers.Registry } // NewDownloadHandler creates a new DownloadHandler. @@ -57,15 +61,22 @@ func NewDownloadHandler(svc DownloadService) *DownloadHandler { // SetRevocationStore wires the stream kill switch so a per-user revocation cuts // an IN-FLIGHT download pour. Downloads stay exempt from the live-stream cap -// (they have their own quota), but a user whose sessions were just revoked must -// not keep pulling a multi-GB file on a pre-revocation connection. Optional; -// nil keeps prior behavior. +// (creation-time download quotas do not gate serving), but a user whose sessions +// were just revoked must not keep pulling a multi-GB file on a pre-revocation +// connection. Optional; nil keeps prior behavior. func (h *DownloadHandler) SetRevocationStore(store *streamrevoke.Store) { if h != nil { h.revocation = store } } +// SetTransferRegistry wires process-local monitoring for active file pours. +func (h *DownloadHandler) SetTransferRegistry(registry *transfers.Registry) { + if h != nil { + h.transfers = registry + } +} + // cutDownloadOnUserRevocation arms the shared in-flight cut for a download pour. // Sessionless (downloads have no stream session), so only user-kind kills apply; // the request entry time predates any future revocation, which is exactly the @@ -408,11 +419,10 @@ func (h *DownloadHandler) HandlePatchDownload(w http.ResponseWriter, r *http.Req // HandleDownloadFile handles GET /downloads/{id}/file. // // This is an offline download, NOT a live stream: it creates no SessionManager -// session and no monitor record, so it is intentionally EXEMPT from the -// concurrent-stream cap and invisible to streammonitor. Downloads are bounded by -// the separate download concurrency/period quota (see the download service's -// ErrConcurrentLimitReached / ErrPeriodLimitReached). See the coverage matrix -// "downloads" note before making downloads count against the live-stream cap. +// session and no live-stream monitor record, so it is intentionally EXEMPT from the +// concurrent-stream cap and never enters streammonitor. Active pours are exposed +// separately through the in-memory transfer registry; creation-time download +// quotas do not gate this serving request. func (h *DownloadHandler) HandleDownloadFile(w http.ResponseWriter, r *http.Request) { userID := apimw.GetUserID(r.Context()) if userID == 0 { @@ -429,16 +439,20 @@ func (h *DownloadHandler) HandleDownloadFile(w http.ResponseWriter, r *http.Requ return } - // In-flight kill switch: a user revocation cuts this pour mid-transfer. - stop := h.cutDownloadOnUserRevocation(w, userID) - defer stop() - - profileID, deviceID, _, _ := managedIdentity(r) + profileID, deviceID, deviceName, _ := managedIdentity(r) filter := requestAccessFilter(r) // 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 { + transfer := h.newTransfer(r, userID, profileID, deviceName, "native_download") + defer h.transfers.End(transfer.ID) + metered := playback.NewSessionMeteredWriter(sw, h.transfers, transfer.ID) + defer func() { _ = metered.Close() }() + // In-flight kill switch: a user revocation cuts this pour mid-transfer. + stop := h.cutDownloadOnUserRevocation(metered, userID) + defer stop() + + if err := h.svc.ServeFile(r.Context(), metered, r, userID, profileID, deviceID, id, filter, transfer); err != nil { if errors.Is(err, downloads.ErrNotFound) || errors.Is(err, catalog.ErrItemNotFound) { writeError(w, http.StatusNotFound, "not_found", "Download not found") return @@ -481,17 +495,37 @@ func (h *DownloadHandler) HandleDirectDownload(w http.ResponseWriter, r *http.Re return } + filter := requestAccessFilter(r) + profileID, _, deviceName, _ := managedIdentity(r) + transfer := h.newTransfer(r, userID, profileID, deviceName, "native_direct") + defer h.transfers.End(transfer.ID) + metered := playback.NewSessionMeteredWriter(w, h.transfers, transfer.ID) + defer func() { _ = metered.Close() }() // In-flight kill switch: a user revocation cuts this pour mid-transfer. - stop := h.cutDownloadOnUserRevocation(w, userID) + stop := h.cutDownloadOnUserRevocation(metered, userID) defer stop() - filter := requestAccessFilter(r) - if err := h.svc.ServeDirect(r.Context(), w, r, userID, fileID, r.URL.Query().Get("format"), filter); err != nil { + if err := h.svc.ServeDirect(r.Context(), metered, r, userID, fileID, r.URL.Query().Get("format"), filter, transfer); err != nil { h.writeDownloadError(w, err) return } } +func (h *DownloadHandler) newTransfer(r *http.Request, userID int, profileID, clientName, route string) transfers.Transfer { + if clientName == "" { + clientName = r.UserAgent() + } + return transfers.Transfer{ + ID: uuid.NewString(), + UserID: userID, + ProfileID: profileID, + Route: route, + ClientIP: clientip.FromContext(r.Context()), + ClientName: clientName, + StartedAt: time.Now(), + } +} + // 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 diff --git a/internal/api/handlers/downloads_test.go b/internal/api/handlers/downloads_test.go index 7f72312f..7e6e6dd7 100644 --- a/internal/api/handlers/downloads_test.go +++ b/internal/api/handlers/downloads_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "net/http" "net/http/httptest" "testing" @@ -14,6 +15,7 @@ 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/transfers" ) // fakeDownloadService is a programmable DownloadService for handler tests. It @@ -54,6 +56,9 @@ type fakeDownloadService struct { gotSubtitleRef string gotDirectFormat string gotDirectFileID int + gotTransfer transfers.Transfer + transferRegistry *transfers.Registry + beginTransfer bool // Season download + series-monitoring (subscription) fakes. season []*downloads.Download @@ -154,9 +159,14 @@ func (f *fakeDownloadService) SyncSubscriptions(_ context.Context, userID int, p return f.syncRegistered, nil } -func (f *fakeDownloadService) ServeDirect(_ context.Context, w http.ResponseWriter, _ *http.Request, _, fileID int, format string, _ catalog.AccessFilter) error { +func (f *fakeDownloadService) ServeDirect(_ context.Context, w http.ResponseWriter, _ *http.Request, _, fileID int, format string, _ catalog.AccessFilter, transfer transfers.Transfer) error { f.gotDirectFileID = fileID f.gotDirectFormat = format + f.gotTransfer = transfer + if f.beginTransfer && f.transferRegistry != nil { + transfer.MediaFileID = fileID + _ = f.transferRegistry.Begin(transfer) + } if f.directErr != nil { return f.directErr } @@ -164,8 +174,13 @@ func (f *fakeDownloadService) ServeDirect(_ context.Context, w http.ResponseWrit return nil } -func (f *fakeDownloadService) ServeFile(_ context.Context, w http.ResponseWriter, _ *http.Request, userID int, profileID, deviceID, downloadID string, _ catalog.AccessFilter) error { +func (f *fakeDownloadService) ServeFile(_ context.Context, w http.ResponseWriter, _ *http.Request, userID int, profileID, deviceID, downloadID string, _ catalog.AccessFilter, transfer transfers.Transfer) error { f.gotServe = identityCall{userID, profileID, deviceID, downloadID} + f.gotTransfer = transfer + if f.beginTransfer && f.transferRegistry != nil { + transfer.DownloadID = downloadID + _ = f.transferRegistry.Begin(transfer) + } if f.serveErr != nil { return f.serveErr } @@ -523,6 +538,39 @@ func TestHandleDirectDownloadMissingFileID(t *testing.T) { } } +func TestDownloadTransferResolutionFailureNeverRegisters(t *testing.T) { + registry := transfers.New() + svc := &fakeDownloadService{directErr: catalog.ErrItemNotFound} + h := NewDownloadHandler(svc) + h.SetTransferRegistry(registry) + rec := httptest.NewRecorder() + h.HandleDirectDownload(rec, downloadTestRequest(http.MethodGet, "/direct-download?file_id=42", nil, 7, "p1", "")) + + if got := registry.Snapshot(); len(got) != 0 { + t.Fatalf("failed resolution left transfers: %+v", got) + } + if svc.gotTransfer.ID == "" || svc.gotTransfer.Route != "native_direct" { + t.Fatalf("transfer context = %+v", svc.gotTransfer) + } +} + +func TestDownloadTransferServingErrorStillDeregisters(t *testing.T) { + registry := transfers.New() + svc := &fakeDownloadService{ + directErr: errors.New("write failed"), + transferRegistry: registry, + beginTransfer: true, + } + h := NewDownloadHandler(svc) + h.SetTransferRegistry(registry) + rec := httptest.NewRecorder() + h.HandleDirectDownload(rec, downloadTestRequest(http.MethodGet, "/direct-download?file_id=42", nil, 7, "p1", "")) + + if got := registry.Snapshot(); len(got) != 0 { + t.Fatalf("serving error left transfers: %+v", got) + } +} + func withChiParams(r *http.Request, params map[string]string) *http.Request { rctx := chi.NewRouteContext() for k, v := range params { diff --git a/internal/api/handlers/nodes.go b/internal/api/handlers/nodes.go index 016a0300..b4e64d5f 100644 --- a/internal/api/handlers/nodes.go +++ b/internal/api/handlers/nodes.go @@ -18,6 +18,7 @@ import ( "github.com/Silo-Server/silo-server/internal/nodesessions" "github.com/Silo-Server/silo-server/internal/playback" "github.com/Silo-Server/silo-server/internal/streammonitor" + "github.com/Silo-Server/silo-server/internal/transfers" "github.com/go-chi/chi/v5" "github.com/redis/go-redis/v9" ) @@ -74,6 +75,12 @@ type NodeListEnabled interface { ListEnabled(ctx context.Context, nodeType string) ([]*nodepool.Node, error) } +// TransferSnapshotSource is the narrow process-local monitoring view needed by +// the admin endpoint. +type TransferSnapshotSource interface { + Snapshot() []transfers.Transfer +} + // NodeHandler handles CRUD operations and health checks for stream nodes. type NodeHandler struct { repo NodeRepository @@ -88,6 +95,7 @@ type NodeHandler struct { // single-node streams (which never write Redis). Optional; nil in edge modes. sessionMgr *playback.SessionManager localNodeName string + transfers TransferSnapshotSource } // SetLocalSessionSource wires the in-process session manager so HandleListSessions @@ -97,6 +105,12 @@ func (h *NodeHandler) SetLocalSessionSource(sm *playback.SessionManager, nodeNam h.localNodeName = nodeName } +// SetTransferSource wires process-local download monitoring into the unfiltered +// admin response. Transfers do not belong to an edge node or live-stream source. +func (h *NodeHandler) SetTransferSource(source TransferSnapshotSource) { + h.transfers = source +} + // NewNodeHandler creates a new NodeHandler. func NewNodeHandler(repo NodeRepository, proxyPool *nodepool.ProxyPool, transcodePool *nodepool.TranscodePool, lister NodeListEnabled, eventBus cache.EventBus, redisClient *redis.Client, jwtSecret string) *NodeHandler { return &NodeHandler{ @@ -419,9 +433,14 @@ func (h *NodeHandler) HandleListSessions(w http.ResponseWriter, r *http.Request) } type sessionsResponse struct { - Sessions []json.RawMessage `json:"sessions"` + Sessions []json.RawMessage `json:"sessions"` + Transfers []transfers.Transfer `json:"transfers"` } - writeJSON(w, http.StatusOK, sessionsResponse{Sessions: sessions}) + activeTransfers := []transfers.Transfer{} + if nodeFilter == "" && h.transfers != nil { + activeTransfers = h.transfers.Snapshot() + } + writeJSON(w, http.StatusOK, sessionsResponse{Sessions: sessions, Transfers: activeTransfers}) } // reloadPools refreshes the in-memory proxy and transcode pools from the database. diff --git a/internal/api/handlers/nodes_test.go b/internal/api/handlers/nodes_test.go new file mode 100644 index 00000000..720bd588 --- /dev/null +++ b/internal/api/handlers/nodes_test.go @@ -0,0 +1,100 @@ +package handlers + +import ( + "bytes" + "context" + "encoding/json" + "net/http/httptest" + "testing" + "time" + + "github.com/Silo-Server/silo-server/internal/nodesessions" + "github.com/Silo-Server/silo-server/internal/playback" + "github.com/Silo-Server/silo-server/internal/transfers" +) + +const ( + testTransferID = "transfer-1" + testProfileID = "profile-1" + testLocalNode = "local" +) + +func TestHandleListSessionsAddsSiblingTransfersWithoutChangingSessionShape(t *testing.T) { + sm := playback.NewSessionManager(0, 0) + ctx := playback.WithClientInfo(context.Background(), playback.ClientInfo{Name: "native client"}) + session, err := sm.StartSessionWithContext(ctx, 42, testProfileID, 9, playback.PlayDirect, false) + if err != nil { + t.Fatalf("StartSessionWithContext: %v", err) + } + registry := transfers.New() + started := time.Date(2026, 7, 29, 12, 0, 0, 0, time.UTC) + if err := registry.Begin(transfers.Transfer{ + ID: testTransferID, + UserID: 42, + ProfileID: testProfileID, + MediaFileID: 9, + Route: "native_direct", + StartedAt: started, + }); err != nil { + t.Fatalf("Begin: %v", err) + } + + h := NewNodeHandler(nil, nil, nil, nil, nil, nil, "") + h.SetLocalSessionSource(sm, testLocalNode) + h.SetTransferSource(registry) + rec := httptest.NewRecorder() + h.HandleListSessions(rec, httptest.NewRequest("GET", "/admin/nodes/sessions", nil)) + + var body struct { + Sessions []json.RawMessage `json:"sessions"` + Transfers []transfers.Transfer `json:"transfers"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil { + t.Fatalf("decode response: %v", err) + } + if len(body.Sessions) != 1 { + t.Fatalf("sessions = %s", body.Sessions) + } + expectedSession, err := json.Marshal(nodesessions.SessionInfo{ + SessionID: session.ID, + NodeName: testLocalNode, + AuthUserID: session.UserID, + ProfileID: session.ProfileID, + Type: string(session.PlayMethod), + Route: session.Origin(), + MediaFileID: session.MediaFileID, + ClientName: session.ClientName, + StartedAt: session.StartedAt.UTC().Format(time.RFC3339), + LastServedAt: session.LastActivityAt.UTC().Format(time.RFC3339), + }) + if err != nil { + t.Fatalf("marshal expected session: %v", err) + } + if !bytes.Equal(body.Sessions[0], expectedSession) { + t.Fatalf("session JSON changed:\n got %s\nwant %s", body.Sessions[0], expectedSession) + } + if len(body.Transfers) != 1 || body.Transfers[0].ID != testTransferID { + t.Fatalf("transfers = %+v", body.Transfers) + } +} + +func TestHandleListSessionsNodeFilterExcludesLocalTransfers(t *testing.T) { + registry := transfers.New() + if err := registry.Begin(transfers.Transfer{ID: testTransferID}); err != nil { + t.Fatalf("Begin: %v", err) + } + h := NewNodeHandler(nil, nil, nil, nil, nil, nil, "") + h.SetTransferSource(registry) + rec := httptest.NewRecorder() + h.HandleListSessions(rec, httptest.NewRequest("GET", "/admin/nodes/sessions?node_id=12", nil)) + + var body struct { + Transfers []transfers.Transfer `json:"transfers"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil { + t.Fatalf("decode response: %v", err) + } + if body.Transfers == nil || len(body.Transfers) != 0 { + t.Fatalf("filtered transfers = %+v, want non-nil empty array", body.Transfers) + } +} diff --git a/internal/api/router.go b/internal/api/router.go index 0222ce2d..db8cd1d5 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -73,6 +73,7 @@ import ( "github.com/Silo-Server/silo-server/internal/subtitles/subsource" "github.com/Silo-Server/silo-server/internal/taskmanager" "github.com/Silo-Server/silo-server/internal/taskmanager/repository" + "github.com/Silo-Server/silo-server/internal/transfers" "github.com/Silo-Server/silo-server/internal/usercollections" "github.com/Silo-Server/silo-server/internal/userstore" "github.com/Silo-Server/silo-server/internal/watchstate" @@ -172,6 +173,7 @@ type Dependencies struct { // RevocationStore is the stream kill switch (may be nil). Admin terminate // writes to it and integrated-mode serve paths consult it. RevocationStore *streamrevoke.Store + TransferRegistry *transfers.Registry OnServerSettingUpdated func(ctx context.Context, key, value string) RequestServerRestart func(ctx context.Context) error ServerRestartStatus *handlers.ServerRestartStatusTracker @@ -1624,6 +1626,7 @@ func NewRouter(deps Dependencies) chi.Router { // devices sync on app open / background refresh; there is no server // background worker. downloadSvc.SetSubscriptions(downloads.NewSubscriptionRepository(deps.DB)) + downloadSvc.SetTransferRegistry(deps.TransferRegistry) downloadHandler = handlers.NewDownloadHandler(downloadSvc) if profileHandler != nil { // Profiles may live outside Postgres (sqlite userdb backend), so @@ -1639,6 +1642,7 @@ func NewRouter(deps Dependencies) chi.Router { // live-stream cap; this only closes the "revoked user keeps pulling a // multi-GB file on a pre-revocation connection" hole). downloadHandler.SetRevocationStore(deps.RevocationStore) + downloadHandler.SetTransferRegistry(deps.TransferRegistry) var policyHandler *handlers.PolicyHandler if deps.PolicySystem != nil && deps.DB != nil { @@ -2940,6 +2944,7 @@ func NewRouter(deps Dependencies) chi.Router { } nodeHandler := handlers.NewNodeHandler(deps.NodeRepo, deps.ProxyPool, deps.TranscodePool, deps.NodeRepo, deps.EventBus, deps.RedisClient, jwtSecret) nodeHandler.SetLocalSessionSource(deps.SessionMgr, deps.NodeID) + nodeHandler.SetTransferSource(deps.TransferRegistry) r.Route("/nodes", func(r chi.Router) { r.Get("/", nodeHandler.HandleListNodes) r.Post("/", nodeHandler.HandleCreateNode) diff --git a/internal/audiobooks/abs/file_handler.go b/internal/audiobooks/abs/file_handler.go index cc8f27b6..b6f4f193 100644 --- a/internal/audiobooks/abs/file_handler.go +++ b/internal/audiobooks/abs/file_handler.go @@ -3,6 +3,7 @@ package abs import ( "crypto/md5" "encoding/hex" + "log/slog" "net/http" "path/filepath" "strconv" @@ -10,8 +11,10 @@ import ( "time" "github.com/go-chi/chi/v5" + "github.com/google/uuid" "github.com/Silo-Server/silo-server/internal/playback" + "github.com/Silo-Server/silo-server/internal/transfers" ) const publicTrackSessionTTL = 24 * time.Hour @@ -54,9 +57,10 @@ func trackInoFor(contentID string, fileIdx int) string { // browser save-to-disk / mobile offline-save behaviour. // // This is a download-class route: like native and Jellyfin-compatible -// downloads it is exempt from the live-stream cap and streammonitor, and it is -// not yet covered by a download quota. It is still revocable before and during -// the file pour. +// downloads it is exempt from the live-stream cap and never enters +// streammonitor. Its active pour is exposed through the separate process-local +// transfer registry. It is not yet covered by a download quota and remains +// revocable before and during the file pour. func (h *Handler) handleFileStream(w http.ResponseWriter, r *http.Request) { a, ok := absAuthFrom(r) if !ok || a.UserID == "" { @@ -128,9 +132,34 @@ func (h *Handler) handleFileStream(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", ct) } - stop := h.deps.Revocation.WatchAndCut(w, "", userID, a.IssuedAt) - defer stop() - if err := playback.ServeDirectPlay(w, r, mediaFile.FilePath); err != nil { + route := "abs_file_stream" + if strings.HasSuffix(r.URL.Path, "/download") { + route = "abs_file_download" + } + transfer := transfers.Transfer{ + ID: uuid.NewString(), + UserID: userID, + ProfileID: a.ProfileID, + MediaFileID: mediaFile.ID, + Route: route, + ClientIP: requestClientIP(r), + ClientName: r.UserAgent(), + StartedAt: time.Now(), + } + if h.deps.Transfers != nil { + if err := h.deps.Transfers.Begin(transfer); err != nil { + slog.DebugContext(r.Context(), "ABS transfer not monitored", "component", "audiobooks", "transfer_id", transfer.ID, "error", err) + } + } + defer h.deps.Transfers.End(transfer.ID) + metered := playback.NewSessionMeteredWriter(w, h.deps.Transfers, transfer.ID) + defer func() { _ = metered.Close() }() + + if h.deps.Revocation != nil { + stop := h.deps.Revocation.WatchAndCut(metered, "", userID, a.IssuedAt) + defer stop() + } + if err := playback.ServeDirectPlay(metered, r, mediaFile.FilePath); err != nil { // ServeDirectPlay has already written an error response; just log. return } diff --git a/internal/audiobooks/abs/handler.go b/internal/audiobooks/abs/handler.go index d1b41d0b..6eda55a7 100644 --- a/internal/audiobooks/abs/handler.go +++ b/internal/audiobooks/abs/handler.go @@ -22,6 +22,7 @@ import ( "github.com/Silo-Server/silo-server/internal/catalog" "github.com/Silo-Server/silo-server/internal/models" "github.com/Silo-Server/silo-server/internal/streamrevoke" + "github.com/Silo-Server/silo-server/internal/transfers" ) // --------------------------------------------------------------------------- @@ -281,6 +282,8 @@ type Dependencies struct { // Revocation makes ABS byte-serving routes obey the shared stream kill // switch. Nil preserves compatibility for isolated tests. Revocation *streamrevoke.Store + // Transfers tracks file pours separately from native playback sessions. + Transfers *transfers.Registry // NativeSessionSyncer flushes native session-manager state into the shared // admin live-session table after ABS play/sync/close events. NativeSessionSyncer PlaybackSessionSyncer diff --git a/internal/audiobooks/abs/rss_feeds_handler.go b/internal/audiobooks/abs/rss_feeds_handler.go index c8a49f3e..54f02b3b 100644 --- a/internal/audiobooks/abs/rss_feeds_handler.go +++ b/internal/audiobooks/abs/rss_feeds_handler.go @@ -13,9 +13,12 @@ import ( "time" "github.com/go-chi/chi/v5" + "github.com/google/uuid" "github.com/oklog/ulid/v2" "github.com/Silo-Server/silo-server/internal/models" + "github.com/Silo-Server/silo-server/internal/playback" + "github.com/Silo-Server/silo-server/internal/transfers" ) var slugRe = regexp.MustCompile(`^[a-z0-9-]{4,64}$`) @@ -243,7 +246,28 @@ func (h *Handler) handlePublicFeedFile(w http.ResponseWriter, r *http.Request) { http.Error(w, "file not found", http.StatusNotFound) return } - stop := h.deps.Revocation.WatchAndCut(w, "", userID, f.CreatedAt) - defer stop() - http.ServeFile(w, r, mf.FilePath) + transfer := transfers.Transfer{ + ID: uuid.NewString(), + UserID: userID, + ProfileID: f.ProfileID, + MediaFileID: mf.ID, + Route: "abs_feed", + ClientIP: requestClientIP(r), + ClientName: r.UserAgent(), + StartedAt: time.Now(), + } + if h.deps.Transfers != nil { + if err := h.deps.Transfers.Begin(transfer); err != nil { + slog.DebugContext(r.Context(), "ABS feed transfer not monitored", "component", "audiobooks", "transfer_id", transfer.ID, "error", err) + } + } + defer h.deps.Transfers.End(transfer.ID) + metered := playback.NewSessionMeteredWriter(w, h.deps.Transfers, transfer.ID) + defer func() { _ = metered.Close() }() + + if h.deps.Revocation != nil { + stop := h.deps.Revocation.WatchAndCut(metered, "", userID, f.CreatedAt) + defer stop() + } + http.ServeFile(metered, r, mf.FilePath) } diff --git a/internal/audiobooks/service.go b/internal/audiobooks/service.go index d8a52b2f..3c0c37ba 100644 --- a/internal/audiobooks/service.go +++ b/internal/audiobooks/service.go @@ -13,6 +13,7 @@ import ( "github.com/Silo-Server/silo-server/internal/recommendations" "github.com/Silo-Server/silo-server/internal/scanner" "github.com/Silo-Server/silo-server/internal/streamrevoke" + "github.com/Silo-Server/silo-server/internal/transfers" "github.com/jackc/pgx/v5/pgxpool" ) @@ -62,6 +63,7 @@ type ABSHandlerDeps struct { SessionMgr *playback.SessionManager SessionSyncer abs.PlaybackSessionSyncer Revocation *streamrevoke.Store + Transfers *transfers.Registry } // absAuthAdapter is the narrow slice of internal/auth that BuildABSHandler @@ -177,6 +179,7 @@ func (s *Service) BuildABSHandler(deps ABSHandlerDeps) *abs.Handler { NativeSessions: deps.SessionMgr, NativeSessionSyncer: deps.SessionSyncer, Revocation: deps.Revocation, + Transfers: deps.Transfers, CoverResolver: func(ctx context.Context, path, variant string) string { if deps.Detail == nil { return "" diff --git a/internal/downloads/service.go b/internal/downloads/service.go index b10957b0..388bf262 100644 --- a/internal/downloads/service.go +++ b/internal/downloads/service.go @@ -19,6 +19,7 @@ import ( "github.com/Silo-Server/silo-server/internal/idgen" "github.com/Silo-Server/silo-server/internal/models" "github.com/Silo-Server/silo-server/internal/playback" + "github.com/Silo-Server/silo-server/internal/transfers" ) // FileResolver looks up media files by various keys. @@ -90,6 +91,7 @@ type Service struct { groupProvider access.GroupPolicyProvider itemAccess ItemAccessChecker settings SettingsReader + transfers *transfers.Registry // Offline-manifest dependencies (Phase 2); nil until SetOfflineDeps wires them. manifest *ManifestBuilder @@ -142,6 +144,11 @@ func (s *Service) SetSubscriptions(subRepo *SubscriptionRepository) { s.subRepo = subRepo } +// SetTransferRegistry wires process-local monitoring for active file pours. +func (s *Service) SetTransferRegistry(registry *transfers.Registry) { + s.transfers = registry +} + // SetGroupPolicyProvider wires access-group policy composition into download // checks. Nil keeps legacy user-row behavior. func (s *Service) SetGroupPolicyProvider(provider access.GroupPolicyProvider) { @@ -846,7 +853,7 @@ 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 { +func (s *Service) ServeDirect(ctx context.Context, w http.ResponseWriter, r *http.Request, userID, fileID int, format string, filter catalog.AccessFilter, transfer transfers.Transfer) error { if _, _, err := s.downloadConfigForUser(ctx, userID, ""); err != nil { return err } @@ -866,6 +873,8 @@ func (s *Service) ServeDirect(ctx context.Context, w http.ResponseWriter, r *htt if !catalog.FileAllowedByAccess(file, filter) { return catalog.ErrItemNotFound } + transfer.MediaFileID = file.ID + s.beginTransfer(ctx, transfer) return s.serveLocalFile(ctx, w, r, file.FilePath, userID) } @@ -873,14 +882,14 @@ func (s *Service) ServeDirect(ctx context.Context, w http.ResponseWriter, r *htt // authorize on (user, profile, device) and re-check per-profile content access // 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 { +func (s *Service) ServeFile(ctx context.Context, w http.ResponseWriter, r *http.Request, userID int, profileID, deviceID, downloadID string, filter catalog.AccessFilter, transfer transfers.Transfer) error { // 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) + return s.serveManaged(ctx, w, r, userID, profileID, deviceID, downloadID, filter, transfer) } dl, err := s.repo.GetByID(ctx, downloadID) @@ -909,7 +918,7 @@ func (s *Service) ServeFile(ctx context.Context, w http.ResponseWriter, r *http. } } - if err := s.serveDownloadBytes(ctx, w, r, dl, userID, filter); err != nil { + if err := s.serveDownloadBytes(ctx, w, r, dl, userID, filter, transfer); err != nil { if dl.Format == FormatOriginal { if updateErr := s.repo.UpdateStatus(ctx, dl.ID, StatusFailed, 0, nil); updateErr != nil { slog.ErrorContext(ctx, "failed to mark download as failed", "component", "downloads", "download_id", dl.ID, "error", updateErr) @@ -927,7 +936,7 @@ func (s *Service) ServeFile(ctx context.Context, w http.ResponseWriter, r *http. // 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 { +func (s *Service) serveManaged(ctx context.Context, w http.ResponseWriter, r *http.Request, userID int, profileID, deviceID, downloadID string, filter catalog.AccessFilter, transfer transfers.Transfer) error { if profileID == "" { return ErrProfileRequired } @@ -943,7 +952,7 @@ func (s *Service) serveManaged(ctx context.Context, w http.ResponseWriter, r *ht if err := s.itemAccess.EnsureAccessible(ctx, dl.ContentID, filter); err != nil { return err } - return s.serveDownloadBytes(ctx, w, r, dl, userID, filter) + return s.serveDownloadBytes(ctx, w, r, dl, userID, filter, transfer) } // PatchStatus lets a client confirm a managed entry's local state @@ -1044,7 +1053,7 @@ func (s *Service) resolveFile(ctx context.Context, req CreateRequest) (*models.M // original rows. Both paths mirror playback's per-file authorization // (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 { +func (s *Service) serveDownloadBytes(ctx context.Context, w http.ResponseWriter, r *http.Request, dl *Download, userID int, filter catalog.AccessFilter, transfer transfers.Transfer) error { file, err := s.fileRepo.GetByID(ctx, dl.MediaFileID) if err != nil { return fmt.Errorf("loading media file: %w", err) @@ -1070,6 +1079,9 @@ func (s *Service) serveDownloadBytes(ctx context.Context, w http.ResponseWriter, if !catalog.FileAllowedByAccess(&served, filter) { return catalog.ErrItemNotFound } + transfer.DownloadID = dl.ID + transfer.MediaFileID = dl.MediaFileID + s.beginTransfer(ctx, transfer) return s.serveLocalFile(ctx, w, r, artifact.OutputPath, userID) } if file.MissingSince != nil { @@ -1078,9 +1090,21 @@ func (s *Service) serveDownloadBytes(ctx context.Context, w http.ResponseWriter, if !catalog.FileAllowedByAccess(file, filter) { return catalog.ErrItemNotFound } + transfer.DownloadID = dl.ID + transfer.MediaFileID = dl.MediaFileID + s.beginTransfer(ctx, transfer) return s.serveLocalFile(ctx, w, r, file.FilePath, userID) } +func (s *Service) beginTransfer(ctx context.Context, transfer transfers.Transfer) { + if s.transfers == nil { + return + } + if err := s.transfers.Begin(transfer); err != nil { + slog.DebugContext(ctx, "download transfer not monitored", "component", "downloads", "transfer_id", transfer.ID, "error", err) + } +} + func (s *Service) serveLocalFile(ctx context.Context, w http.ResponseWriter, r *http.Request, path string, userID int) error { f, err := os.Open(path) if err != nil { diff --git a/internal/jellycompat/handlers_playback.go b/internal/jellycompat/handlers_playback.go index 0eb8d992..156bc1f1 100644 --- a/internal/jellycompat/handlers_playback.go +++ b/internal/jellycompat/handlers_playback.go @@ -29,6 +29,7 @@ import ( "github.com/Silo-Server/silo-server/internal/streamtoken" "github.com/Silo-Server/silo-server/internal/subtitles" "github.com/Silo-Server/silo-server/internal/transcodenode" + "github.com/Silo-Server/silo-server/internal/transfers" "github.com/Silo-Server/silo-server/internal/userstore" "github.com/Silo-Server/silo-server/internal/watchsync" ) @@ -212,6 +213,8 @@ type PlaybackHandler struct { // an admin terminate / over-cap / account revocation actually stops the bytes — // the multi-node redirect path is already guarded by the proxy. Optional. Revocation *streamrevoke.Store + // Transfers tracks download-class pours without entering live-stream limits. + Transfers *transfers.Registry } // recipeNodePutter persists and removes a remote transcode's reconstruction diff --git a/internal/jellycompat/router.go b/internal/jellycompat/router.go index 4378665d..c36ca078 100644 --- a/internal/jellycompat/router.go +++ b/internal/jellycompat/router.go @@ -109,6 +109,7 @@ func NewRouter(deps Dependencies) chi.Router { playbackHandler.NodePlanner = deps.NodePlanner playbackHandler.JWTSecret = deps.JWTSecret playbackHandler.Revocation = deps.RevocationStore + playbackHandler.Transfers = deps.TransferRegistry // Compat transcode reconstruct is driven by the recipe carried in the durable // compat playback store (jellycompat_playback_sessions); no separate native // recipe table is needed. diff --git a/internal/jellycompat/server.go b/internal/jellycompat/server.go index 945614d6..8af41ba3 100644 --- a/internal/jellycompat/server.go +++ b/internal/jellycompat/server.go @@ -19,6 +19,7 @@ import ( "github.com/Silo-Server/silo-server/internal/secret" "github.com/Silo-Server/silo-server/internal/streamrevoke" "github.com/Silo-Server/silo-server/internal/subtitles" + "github.com/Silo-Server/silo-server/internal/transfers" "github.com/Silo-Server/silo-server/internal/userstore" "github.com/Silo-Server/silo-server/internal/watchstate" "github.com/Silo-Server/silo-server/internal/watchsync" @@ -34,6 +35,8 @@ type Dependencies struct { // RevocationStore is the shared stream kill switch consulted by local serving. RevocationStore *streamrevoke.Store + // TransferRegistry exposes download-class pours separately from live streams. + TransferRegistry *transfers.Registry // LiveConfig returns the current hot-reloaded config. May be nil (tests, // worker modes); read through CurrentConfig(), which falls back to Config. LiveConfig func() *config.Config diff --git a/internal/jellycompat/streams.go b/internal/jellycompat/streams.go index 55b726b3..0a690599 100644 --- a/internal/jellycompat/streams.go +++ b/internal/jellycompat/streams.go @@ -21,10 +21,12 @@ import ( "github.com/go-chi/chi/v5" + "github.com/Silo-Server/silo-server/internal/clientip" "github.com/Silo-Server/silo-server/internal/models" "github.com/Silo-Server/silo-server/internal/nodepool" "github.com/Silo-Server/silo-server/internal/playback" "github.com/Silo-Server/silo-server/internal/subtitles" + "github.com/Silo-Server/silo-server/internal/transfers" "github.com/Silo-Server/silo-server/internal/watchsync" ) @@ -180,8 +182,9 @@ func (h *PlaybackHandler) HandleVideoStream(w http.ResponseWriter, r *http.Reque // this route must exist. // HandleDownload serves a full file for offline/download clients (e.g. Infuse // with CanDownload). By design it is NOT a live stream: it creates no -// SessionManager session and no monitor record, so it is EXEMPT from the -// concurrent-stream cap and invisible to streammonitor. NOTE: unlike the native +// SessionManager session and never enters streammonitor, so it is EXEMPT from the +// concurrent-stream cap. Its active pour is exposed through the separate, +// process-local transfer registry. NOTE: unlike the native // /downloads/{id}/file route, this route is NOT covered by the download // concurrency/period quota either — it requires no download row, only compat // auth. Access control is compat authentication plus the library access filter; @@ -226,21 +229,41 @@ func (h *PlaybackHandler) HandleDownload(w http.ResponseWriter, r *http.Request) return } file, err := h.fileResolver.GetByID(r.Context(), version.FileID) - if err != nil { + if err != nil || file == nil { writeError(w, http.StatusNotFound, "NotFound", "Media file not found") return } + client := playback.ClientInfoFromContext(r.Context()) + transfer := transfers.Transfer{ + ID: uuidNewString(), + UserID: session.StreamAppUserID, + ProfileID: session.ProfileID, + MediaFileID: file.ID, + Route: "jellycompat_download", + ClientIP: clientip.FromContext(r.Context()), + ClientName: firstNonEmpty(client.Name, client.UserAgent), + StartedAt: time.Now(), + } + if h.Transfers != nil { + if err := h.Transfers.Begin(transfer); err != nil { + slog.DebugContext(r.Context(), "compat transfer not monitored", "component", "jellycompat", "transfer_id", transfer.ID, "error", err) + } + } + defer h.Transfers.End(transfer.ID) + metered := playback.NewSessionMeteredWriter(w, h.Transfers, transfer.ID) + defer func() { _ = metered.Close() }() + // In-flight kill switch: sessionless pour, so only user-kind kills apply; // the entry time predates any future revocation (the user-kill cutoff), so a // revocation issued mid-transfer hangs this connection up. if h.Revocation != nil { - stop := h.Revocation.WatchAndCut(w, "", session.StreamAppUserID, time.Now()) + stop := h.Revocation.WatchAndCut(metered, "", session.StreamAppUserID, transfer.StartedAt) defer stop() } w.Header().Set("Content-Disposition", "attachment; filename*=UTF-8''"+url.PathEscape(filepath.Base(file.FilePath))) - _ = playback.ServeDirectPlay(w, r, file.FilePath) + _ = playback.ServeDirectPlay(metered, r, file.FilePath) } // HandleMasterManifest serves the compat-owned HLS manifest route. diff --git a/internal/transfers/registry.go b/internal/transfers/registry.go new file mode 100644 index 00000000..8710e778 --- /dev/null +++ b/internal/transfers/registry.go @@ -0,0 +1,183 @@ +// Package transfers tracks active download-class file pours in memory. +package transfers + +import ( + "errors" + "log/slog" + "math" + "sort" + "strings" + "sync" + "time" + "unicode/utf8" +) + +const ( + defaultMaxEntries = 10_000 + fullWarningInterval = time.Minute + maxDownloadIDLength = 256 + maxProfileIDLength = 256 + maxRouteLength = 64 + maxClientIPLength = 128 + maxClientNameLength = 256 +) + +var ( + ErrRegistryFull = errors.New("transfer registry is full") + ErrInvalidID = errors.New("transfer id is required") + ErrDuplicateID = errors.New("transfer id is already active") +) + +// Transfer describes one active HTTP file pour. DownloadID is correlation +// metadata only: ID uniquely identifies this request, including concurrent +// Range requests for the same download row. +type Transfer struct { + ID string `json:"id"` + DownloadID string `json:"download_id"` + UserID int `json:"user_id"` + ProfileID string `json:"profile_id"` + MediaFileID int `json:"media_file_id"` + Route string `json:"route"` + ClientIP string `json:"client_ip"` + ClientName string `json:"client_name"` + StartedAt time.Time `json:"started_at"` + LastServedAt time.Time `json:"last_served_at"` + BytesServed int64 `json:"bytes_served"` +} + +// Options customizes a Registry. Zero values use production defaults. +type Options struct { + MaxEntries int + Now func() time.Time +} + +// Registry is a bounded, process-local collection of active transfers. +type Registry struct { + mu sync.RWMutex + items map[string]Transfer + maxEntries int + now func() time.Time + lastFullWarning time.Time +} + +// New creates a Registry with production limits. +func New() *Registry { + return NewWithOptions(Options{}) +} + +// NewWithOptions creates a Registry with explicit limits, primarily for tests. +func NewWithOptions(opts Options) *Registry { + if opts.MaxEntries <= 0 { + opts.MaxEntries = defaultMaxEntries + } + if opts.Now == nil { + opts.Now = time.Now + } + return &Registry{ + items: make(map[string]Transfer), + maxEntries: opts.MaxEntries, + now: opts.Now, + } +} + +// Begin records an active transfer. Strings derived from request metadata are +// normalized and clamped so the entry limit also provides a useful memory bound. +func (r *Registry) Begin(t Transfer) error { + if r == nil { + return nil + } + t.ID = strings.TrimSpace(t.ID) + if t.ID == "" { + return ErrInvalidID + } + t.DownloadID = clamp(t.DownloadID, maxDownloadIDLength) + t.ProfileID = clamp(t.ProfileID, maxProfileIDLength) + t.Route = clamp(t.Route, maxRouteLength) + t.ClientIP = clamp(t.ClientIP, maxClientIPLength) + t.ClientName = clamp(t.ClientName, maxClientNameLength) + if t.StartedAt.IsZero() { + t.StartedAt = r.now() + } + t.StartedAt = t.StartedAt.UTC() + t.LastServedAt = t.StartedAt + t.BytesServed = 0 + + r.mu.Lock() + defer r.mu.Unlock() + if _, exists := r.items[t.ID]; exists { + return ErrDuplicateID + } + if len(r.items) >= r.maxEntries { + now := r.now() + if r.lastFullWarning.IsZero() || now.Sub(r.lastFullWarning) >= fullWarningInterval { + r.lastFullWarning = now + slog.Warn("transfer registry full; serving without monitoring", "component", "transfers", "max_entries", r.maxEntries) + } + return ErrRegistryFull + } + r.items[t.ID] = t + return nil +} + +// AddServedBytes implements playback.ServedBytesRecorder. Updates for unknown +// transfers are deliberately discarded, and non-positive values are ignored. +// The metered writer reports in coarse chunks (currently 1 MiB and on Close), +// so LastServedAt is intentionally a coarse monitoring timestamp. +func (r *Registry) AddServedBytes(id string, n int64) error { + if r == nil || n <= 0 { + return nil + } + r.mu.Lock() + defer r.mu.Unlock() + t, ok := r.items[id] + if !ok { + return nil + } + if n > math.MaxInt64-t.BytesServed { + t.BytesServed = math.MaxInt64 + } else { + t.BytesServed += n + } + t.LastServedAt = r.now().UTC() + r.items[id] = t + return nil +} + +// End removes an active transfer. It is idempotent. +func (r *Registry) End(id string) { + if r == nil { + return + } + r.mu.Lock() + delete(r.items, id) + r.mu.Unlock() +} + +// Snapshot returns a stable copy of the active transfers. +func (r *Registry) Snapshot() []Transfer { + if r == nil { + return nil + } + r.mu.RLock() + out := make([]Transfer, 0, len(r.items)) + for _, t := range r.items { + out = append(out, t) + } + r.mu.RUnlock() + sort.Slice(out, func(i, j int) bool { + if out[i].StartedAt.Equal(out[j].StartedAt) { + return out[i].ID < out[j].ID + } + return out[i].StartedAt.Before(out[j].StartedAt) + }) + return out +} + +func clamp(value string, maxRunes int) string { + value = strings.TrimSpace(value) + if utf8.RuneCountInString(value) <= maxRunes { + return value + } + runes := []rune(value) + return string(runes[:maxRunes]) +} diff --git a/internal/transfers/registry_test.go b/internal/transfers/registry_test.go new file mode 100644 index 00000000..8182d6fa --- /dev/null +++ b/internal/transfers/registry_test.go @@ -0,0 +1,274 @@ +package transfers + +import ( + "errors" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/Silo-Server/silo-server/internal/playback" +) + +const testPourID = "pour" + +func TestRegistryLifecycleAndAccounting(t *testing.T) { + now := time.Date(2026, 7, 29, 12, 0, 0, 0, time.UTC) + r := NewWithOptions(Options{Now: func() time.Time { return now }}) + started := now.Add(-time.Minute) + if err := r.Begin(Transfer{ID: "pour-1", UserID: 7, DownloadID: "download-1", StartedAt: started}); err != nil { + t.Fatalf("Begin: %v", err) + } + + got := r.Snapshot() + if len(got) != 1 || got[0].LastServedAt != started || got[0].BytesServed != 0 { + t.Fatalf("initial snapshot = %+v", got) + } + if err := r.AddServedBytes("pour-1", 25); err != nil { + t.Fatalf("AddServedBytes: %v", err) + } + got = r.Snapshot() + if got[0].BytesServed != 25 || got[0].LastServedAt != now { + t.Fatalf("accounted snapshot = %+v", got) + } + + r.End("pour-1") + r.End("pour-1") + if got := r.Snapshot(); len(got) != 0 { + t.Fatalf("snapshot after End = %+v", got) + } +} + +func TestRegistryUnknownAndNonPositiveUpdatesAreNoOps(t *testing.T) { + r := New() + if err := r.AddServedBytes("missing", 20); err != nil { + t.Fatalf("unknown AddServedBytes: %v", err) + } + if len(r.Snapshot()) != 0 { + t.Fatal("unknown update created an entry") + } + if err := r.Begin(Transfer{ID: testPourID}); err != nil { + t.Fatalf("Begin: %v", err) + } + for _, n := range []int64{0, -1, -100} { + if err := r.AddServedBytes(testPourID, n); err != nil { + t.Fatalf("AddServedBytes(%d): %v", n, err) + } + } + if got := r.Snapshot()[0].BytesServed; got != 0 { + t.Fatalf("BytesServed = %d, want 0", got) + } +} + +func TestRegistryConcurrentPoursForSameDownloadDoNotCollide(t *testing.T) { + r := New() + for _, id := range []string{"range-a", "range-b"} { + if err := r.Begin(Transfer{ID: id, DownloadID: "same-download"}); err != nil { + t.Fatalf("Begin(%s): %v", id, err) + } + } + _ = r.AddServedBytes("range-a", 10) + _ = r.AddServedBytes("range-b", 20) + got := r.Snapshot() + if len(got) != 2 || got[0].ID == got[1].ID { + t.Fatalf("snapshot = %+v", got) + } +} + +func TestRegistryConcurrentLifecycle(t *testing.T) { + r := NewWithOptions(Options{MaxEntries: 256}) + var wg sync.WaitGroup + for i := 0; i < 200; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + id := time.Unix(0, int64(i)+1).String() + if err := r.Begin(Transfer{ID: id}); err != nil { + t.Errorf("Begin(%d): %v", i, err) + return + } + for j := 0; j < 10; j++ { + _ = r.AddServedBytes(id, 1) + _ = r.Snapshot() + } + r.End(id) + }(i) + } + wg.Wait() + if got := r.Snapshot(); len(got) != 0 { + t.Fatalf("leaked transfers = %d", len(got)) + } +} + +func TestRegistryCapHoldsUnderRacingBegin(t *testing.T) { + const cap = 7 + r := NewWithOptions(Options{MaxEntries: cap}) + var wg sync.WaitGroup + var accepted atomic.Int64 + for i := 0; i < 100; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + err := r.Begin(Transfer{ID: time.Unix(0, int64(i)+1).String()}) + if err == nil { + accepted.Add(1) + } else if !errors.Is(err, ErrRegistryFull) { + t.Errorf("Begin(%d): %v", i, err) + } + }(i) + } + wg.Wait() + if got := accepted.Load(); got != cap { + t.Fatalf("accepted = %d, want %d", got, cap) + } + if got := len(r.Snapshot()); got != cap { + t.Fatalf("snapshot size = %d, want %d", got, cap) + } +} + +func TestRegistryClampsAndNormalizesRequestStrings(t *testing.T) { + r := New() + long := strings.Repeat("界", 400) + if err := r.Begin(Transfer{ + ID: " pour ", + DownloadID: " " + long + " ", + ProfileID: long, + Route: long, + ClientIP: long, + ClientName: long, + }); err != nil { + t.Fatalf("Begin: %v", err) + } + got := r.Snapshot()[0] + if got.ID != testPourID { + t.Fatalf("ID = %q", got.ID) + } + for name, tc := range map[string]struct { + value string + max int + }{ + "download": {got.DownloadID, maxDownloadIDLength}, + "profile": {got.ProfileID, maxProfileIDLength}, + "route": {got.Route, maxRouteLength}, + "ip": {got.ClientIP, maxClientIPLength}, + "client": {got.ClientName, maxClientNameLength}, + } { + if n := len([]rune(tc.value)); n != tc.max { + t.Errorf("%s length = %d, want %d", name, n, tc.max) + } + } +} + +func TestMeterCloseFlushesSubMiBTailBeforeRegistryEnd(t *testing.T) { + var accountingCalls atomic.Int64 + r := NewWithOptions(Options{Now: func() time.Time { + accountingCalls.Add(1) + return time.Now() + }}) + if err := r.Begin(Transfer{ID: testPourID, StartedAt: time.Now()}); err != nil { + t.Fatalf("Begin: %v", err) + } + + func() { + defer r.End(testPourID) // registered first, runs last + metered := playback.NewSessionMeteredWriter(httptest.NewRecorder(), r, testPourID) + defer func() { _ = metered.Close() }() // registered second, runs first + if _, err := metered.Write([]byte("smaller than one MiB")); err != nil { + t.Fatalf("Write: %v", err) + } + }() + + if got := accountingCalls.Load(); got != 1 { + t.Fatalf("accounting calls = %d, want 1; tail was not flushed before End", got) + } +} + +func TestDirectPlayByteAccounting(t *testing.T) { + payload := []byte("0123456789abcdef") + path := filepath.Join(t.TempDir(), "sample.mp4") + if err := os.WriteFile(path, payload, 0o600); err != nil { + t.Fatalf("WriteFile: %v", err) + } + + tests := []struct { + name string + method string + headers map[string]string + writer func() http.ResponseWriter + wantBytes int64 + wantStatus int + }{ + {name: "full body", method: http.MethodGet, writer: func() http.ResponseWriter { return httptest.NewRecorder() }, wantBytes: int64(len(payload)), wantStatus: http.StatusOK}, + {name: "range", method: http.MethodGet, headers: map[string]string{"Range": "bytes=2-5"}, writer: func() http.ResponseWriter { return httptest.NewRecorder() }, wantBytes: 4, wantStatus: http.StatusPartialContent}, + {name: "head", method: http.MethodHead, writer: func() http.ResponseWriter { return httptest.NewRecorder() }, wantStatus: http.StatusOK}, + {name: "conditional", method: http.MethodGet, headers: map[string]string{"If-Modified-Since": time.Now().Add(time.Hour).UTC().Format(http.TimeFormat)}, writer: func() http.ResponseWriter { return httptest.NewRecorder() }, wantStatus: http.StatusNotModified}, + {name: "client disconnect", method: http.MethodGet, writer: func() http.ResponseWriter { return &disconnectWriter{limit: 5, header: make(http.Header)} }, wantBytes: 5, wantStatus: http.StatusOK}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + r := New() + if err := r.Begin(Transfer{ID: testPourID}); err != nil { + t.Fatalf("Begin: %v", err) + } + base := tt.writer() + metered := playback.NewSessionMeteredWriter(base, r, testPourID) + req := httptest.NewRequest(tt.method, "/file", nil) + for key, value := range tt.headers { + req.Header.Set(key, value) + } + if err := playback.ServeDirectPlay(metered, req, path); err != nil { + t.Fatalf("ServeDirectPlay: %v", err) + } + if err := metered.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + got := r.Snapshot()[0] + if got.BytesServed != tt.wantBytes { + t.Fatalf("BytesServed = %d, want %d", got.BytesServed, tt.wantBytes) + } + switch w := base.(type) { + case *httptest.ResponseRecorder: + if w.Code != tt.wantStatus { + t.Fatalf("status = %d, want %d", w.Code, tt.wantStatus) + } + case *disconnectWriter: + if w.status != tt.wantStatus { + t.Fatalf("status = %d, want %d", w.status, tt.wantStatus) + } + } + }) + } +} + +type disconnectWriter struct { + header http.Header + status int + limit int +} + +func (w *disconnectWriter) Header() http.Header { return w.header } + +func (w *disconnectWriter) WriteHeader(status int) { + if w.status == 0 { + w.status = status + } +} + +func (w *disconnectWriter) Write(p []byte) (int, error) { + if w.status == 0 { + w.status = http.StatusOK + } + n := min(len(p), w.limit) + w.limit -= n + if n < len(p) || w.limit == 0 { + return n, io.ErrClosedPipe + } + return n, nil +}