feat(downloads): monitor in-flight download pours in memory
Six routes poured full media with no monitor record and no byte measurement: the two native download routes, the Jellyfin-compat download, both ABS file variants, and the public ABS RSS feed file. They were the last invisible bytes on the server. - internal/transfers: a process-local, in-memory registry of active pours. Bounded (10k entries) with rate-limited "full" warnings, all request-derived strings normalized and length-clamped (ABS and jellycompat carry no bounded client name, so those fields are header-derived and untrusted), overflow-safe byte accumulation, deterministic snapshots, nil-safe throughout. No I/O on any path, and no persistence: a pour dies with the process, so durable rows would only need reaping after a crash. - Reuses the existing playback.SessionMeteredWriter via ServedBytesRecorder rather than adding a second writer — that writer is where the sendfile (ReadFrom) and kill-cut (Unwrap) hazards live and both have regressed before. - Deliberately NOT plumbed through streammonitor/streamenforcer, which are untouched. Downloads stay off the live-stream path by construction: there is no type, field or collection through which one can reach the enforcer, so a download can never be counted against max_streams or trimmed as an over-cap stream. - Admin visibility is a sibling `transfers` array on the existing /admin/nodes/sessions response; the `sessions` array is byte-for-byte unchanged. Transfers appear only in the unfiltered listing, since a node_id filter targets an edge and these are process-local. - Per-pour ids are unique, never the download id: concurrent and repeated GET/Range requests against one download row are legitimate and would collide. The id is minted in the handler (where the revocation watcher is armed) and the registry entry is opened in the service only after file/artifact resolution succeeds, so a failed auth or lookup never registers a transfer. - Defer order is an invariant at every call site: End is registered before the meter's Close so LIFO flushes the tail first. Reversed, a final sub-1MiB flush lands on an unknown id and is silently lost. Pinned by a test. No schema change and no migration: downloads.bytes_sent keeps its documented meaning (a lifecycle marker set to file_size on completion, not a live counter) and is untouched. Phase 2 — killing a single download, and a standing per-user download block — is deliberately deferred. A user revocation already cuts in-flight download pours; it is a cutoff, so it does not refuse new ones. The unique per-pour id exists so phase 2 only changes "" to that id at each WatchAndCut site. Part of the stream monitoring & kill-switch epic.
This commit is contained in:
CoffeeKnyte
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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 ""
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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])
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user