haeaeae/silo-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(playback): fail closed on transfer-registry saturation

Browse Source 
`transfers.Registry` caps at 10,000 entries. Past that `Begin` returned
`ErrRegistryFull`, and every call site logged at Debug and served the file
anyway with its byte updates discarded. With no connection cap anywhere, one
actor could pin the registry and blind download-class monitoring for everyone —
which is the monitoring the abuse controls depend on.

Per decision A7, saturation now fails closed and is unreachable by a single
actor in the first place:

- A per-user concurrent-transfer cap (`playback.max_user_concurrent_transfers`,
  default 24) checked before the global limit, so one actor exhausts its own
  budget rather than the shared registry. 24 leaves headroom for
  multi-connection downloaders, which open 4-8 sockets per file and take a
  transfer id per concurrent Range request. `MaxPerUser` is a *pointer* option
  because a plain int cannot distinguish "unset, use the default" from
  "explicitly 0, unlimited".
- All five download-class call sites refuse rather than serve unmonitored:
  429 `transfer_limit_exceeded` for the per-user cap, 503
  `monitoring_unavailable` for a full registry, both with `Retry-After`.
  The handoff listed four sites; `internal/api/handlers/ebook_reader.go` was
  the missing fifth.
- The ABS file handler now admits the transfer *before* setting
  Content-Disposition and the audio Content-Type, so a rejection is not
  mislabeled as an audio attachment.

Two correctness fixes fall out of doing this properly:

- `End` now decrements the per-user count only when it actually removed an
  entry, and drops the map entry at zero so neither counts nor warning
  timestamps grow unbounded.
- The two download handlers registered `defer transfers.End(transfer.ID)`
  *before* the service called `Begin`, so a duplicate id would have removed
  another request's live record. `Begin` moves into the handler beside its
  `End`, and the service keeps its late DownloadID/MediaFileID enrichment
  through a new `Registry.Annotate`.

Saturation is logged at Warn inside the registry, rate-limited per user and
carrying the route and user id, rather than once per rejected request at each
call site. A7 asked for Warn at the call sites, but an actor parked at its cap
would then generate unbounded warning traffic — a log-amplification vector of
its own. No information is lost: route and user id are already on the Transfer.

429/503 are new failure statuses on existing endpoints, not repurposed ones.
silo-android and silo-apple need follow-up to retry with backoff on
`Retry-After`; the cap and the fail-closed behavior are advertised on
`GET /downloads/capability` so clients can feature-detect instead of probing.

The jellycompat call site is verified by reading only: that test package does
not compile on origin/main (pre-existing, unrelated).

Part 3 of 3 for the Batch 4 liveness/replica work.

Part of #305
This commit is contained in:
CoffeeKnyte
2026-07-31 00:46:14 +00:00
parent 40f117bd2d
commit cd9e358bbc
19 changed files with 697 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/handlers/downloads.go
+48 -16
View File
@@ -152,14 +152,16 @@ type batchManifestsResponse struct {
// downloadCapabilityResponse is the GET /downloads/capability payload clients
// use for feature detection instead of introspecting admin settings.
type downloadCapabilityResponse struct {
Enabled bool `json:"enabled"`
DownloadAllowed bool `json:"download_allowed"`
QualityPresets []string `json:"quality_presets"`
TranscodeEnabled bool `json:"transcode_enabled"`
TranscodeUserAllowed bool `json:"transcode_user_allowed"`
SeasonDownload bool `json:"season_download"`
SeriesMonitoring bool `json:"series_monitoring"`
MonitoringModes []string `json:"monitoring_modes,omitempty"`
Enabled bool `json:"enabled"`
DownloadAllowed bool `json:"download_allowed"`
QualityPresets []string `json:"quality_presets"`
TranscodeEnabled bool `json:"transcode_enabled"`
TranscodeUserAllowed bool `json:"transcode_user_allowed"`
SeasonDownload bool `json:"season_download"`
SeriesMonitoring bool `json:"series_monitoring"`
MonitoringModes []string `json:"monitoring_modes,omitempty"`
MaxUserConcurrentTransfers int `json:"max_user_concurrent_transfers"`
TransferMonitoringFailClosed bool `json:"transfer_monitoring_fail_closed"`
}
func toDownloadResponse(d *downloads.Download) downloadResponse {
@@ -216,14 +218,16 @@ func (h *DownloadHandler) HandleCapability(w http.ResponseWriter, r *http.Reques
}
writeJSON(w, http.StatusOK, downloadCapabilityResponse{
Enabled: capInfo.Enabled,
DownloadAllowed: capInfo.DownloadAllowed,
QualityPresets: capInfo.QualityPresets,
TranscodeEnabled: capInfo.TranscodeEnabled,
TranscodeUserAllowed: capInfo.TranscodeUserAllowed,
SeasonDownload: capInfo.SeasonDownload,
SeriesMonitoring: capInfo.SeriesMonitoring,
MonitoringModes: capInfo.MonitoringModes,
Enabled: capInfo.Enabled,
DownloadAllowed: capInfo.DownloadAllowed,
QualityPresets: capInfo.QualityPresets,
TranscodeEnabled: capInfo.TranscodeEnabled,
TranscodeUserAllowed: capInfo.TranscodeUserAllowed,
SeasonDownload: capInfo.SeasonDownload,
SeriesMonitoring: capInfo.SeriesMonitoring,
MonitoringModes: capInfo.MonitoringModes,
MaxUserConcurrentTransfers: h.transfers.MaxPerUser(),
TransferMonitoringFailClosed: true,
})
}
@@ -447,6 +451,9 @@ func (h *DownloadHandler) HandleDownloadFile(w http.ResponseWriter, r *http.Requ
// the write deadline with progress instead.
sw := httpstream.NewRollingDeadlineWriterCtx(r.Context(), w)
transfer := h.newTransfer(r, userID, profileID, deviceName, "native_download")
if !h.beginTransfer(w, r, transfer) {
return
}
defer h.transfers.End(transfer.ID)
metered := playback.NewSessionMeteredWriter(sw, h.transfers, transfer.ID)
defer func() { _ = metered.Close() }()
@@ -502,6 +509,9 @@ func (h *DownloadHandler) HandleDirectDownload(w http.ResponseWriter, r *http.Re
startedAt := time.Now()
r = r.WithContext(httpstream.WithCutLatch(r.Context(), &httpstream.CutLatch{}))
transfer := h.newTransfer(r, userID, profileID, deviceName, "native_direct")
if !h.beginTransfer(w, r, transfer) {
return
}
defer h.transfers.End(transfer.ID)
metered := playback.NewSessionMeteredWriter(w, h.transfers, transfer.ID)
defer func() { _ = metered.Close() }()
@@ -515,6 +525,28 @@ func (h *DownloadHandler) HandleDirectDownload(w http.ResponseWriter, r *http.Re
}
}
func (h *DownloadHandler) beginTransfer(w http.ResponseWriter, r *http.Request, transfer transfers.Transfer) bool {
if err := h.transfers.Begin(transfer); err != nil {
slog.DebugContext(r.Context(), "download transfer rejected",
"component", "api",
"transfer_id", transfer.ID,
"error", err,
)
switch {
case errors.Is(err, transfers.ErrUserTransferLimit):
w.Header().Set("Retry-After", "5")
writeError(w, http.StatusTooManyRequests, "transfer_limit_exceeded", "Concurrent transfer limit exceeded")
case errors.Is(err, transfers.ErrRegistryFull):
w.Header().Set("Retry-After", "5")
writeError(w, http.StatusServiceUnavailable, "monitoring_unavailable", "Transfer monitoring unavailable")
default:
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to monitor transfer")
}
return false
}
return true
}
func (h *DownloadHandler) newTransfer(r *http.Request, userID int, profileID, clientName, route string) transfers.Transfer {
if clientName == "" {
clientName = r.UserAgent()
internal/api/handlers/downloads_test.go
+86 -2
View File
@@ -165,7 +165,7 @@ func (f *fakeDownloadService) ServeDirect(_ context.Context, w http.ResponseWrit
f.gotTransfer = transfer
if f.beginTransfer && f.transferRegistry != nil {
transfer.MediaFileID = fileID
_ = f.transferRegistry.Begin(transfer)
f.transferRegistry.Annotate(transfer.ID, "", fileID)
}
if f.directErr != nil {
return f.directErr
@@ -179,7 +179,7 @@ func (f *fakeDownloadService) ServeFile(_ context.Context, w http.ResponseWriter
f.gotTransfer = transfer
if f.beginTransfer && f.transferRegistry != nil {
transfer.DownloadID = downloadID
_ = f.transferRegistry.Begin(transfer)
f.transferRegistry.Annotate(transfer.ID, downloadID, 0)
}
if f.serveErr != nil {
return f.serveErr
@@ -281,6 +281,7 @@ func TestHandleCapability(t *testing.T) {
TranscodeUserAllowed: true,
}}
h := NewDownloadHandler(svc)
h.SetTransferRegistry(transfers.New())
rec := httptest.NewRecorder()
h.HandleCapability(rec, downloadTestRequest(http.MethodGet, "/downloads/capability", nil, 7, "", ""))
@@ -298,6 +299,9 @@ func TestHandleCapability(t *testing.T) {
if len(resp.QualityPresets) != 1 || resp.QualityPresets[0] != downloads.QualityOriginal {
t.Fatalf("quality presets = %v, want [original]", resp.QualityPresets)
}
if resp.MaxUserConcurrentTransfers != 24 || !resp.TransferMonitoringFailClosed {
t.Fatalf("transfer capability = %+v", resp)
}
}
func TestHandleCapabilityUnauthorized(t *testing.T) {
@@ -538,6 +542,86 @@ func TestHandleDirectDownloadMissingFileID(t *testing.T) {
}
}
func TestDownloadTransferAdmissionFailsClosed(t *testing.T) {
for _, tc := range []struct {
name string
registry func(t *testing.T) *transfers.Registry
wantStatus int
wantCode string
}{
{
name: "per-user cap",
registry: func(t *testing.T) *transfers.Registry {
t.Helper()
limit := 1
r := transfers.NewWithOptions(transfers.Options{MaxPerUser: &limit})
if err := r.Begin(transfers.Transfer{ID: "existing", UserID: 7}); err != nil {
t.Fatal(err)
}
return r
},
wantStatus: http.StatusTooManyRequests,
wantCode: "transfer_limit_exceeded",
},
{
name: "global cap",
registry: func(t *testing.T) *transfers.Registry {
t.Helper()
unlimited := 0
r := transfers.NewWithOptions(transfers.Options{MaxEntries: 1, MaxPerUser: &unlimited})
if err := r.Begin(transfers.Transfer{ID: "existing", UserID: 8}); err != nil {
t.Fatal(err)
}
return r
},
wantStatus: http.StatusServiceUnavailable,
wantCode: "monitoring_unavailable",
},
} {
for _, route := range []string{"managed", "direct"} {
for _, method := range []string{http.MethodGet, http.MethodHead, "range"} {
t.Run(tc.name+"/"+route+"/"+method, func(t *testing.T) {
registry := tc.registry(t)
svc := &fakeDownloadService{}
h := NewDownloadHandler(svc)
h.SetTransferRegistry(registry)
requestMethod := method
if method == "range" {
requestMethod = http.MethodGet
}
var req *http.Request
if route == "managed" {
req = withChiID(downloadTestRequest(requestMethod, "/downloads/dl1/file", nil, 7, "p1", "dev1"), "dl1")
} else {
req = downloadTestRequest(requestMethod, "/direct-download?file_id=42", nil, 7, "p1", "")
}
if method == "range" {
req.Header.Set("Range", "bytes=0-1")
}
rec := httptest.NewRecorder()
if route == "managed" {
h.HandleDownloadFile(rec, req)
} else {
h.HandleDirectDownload(rec, req)
}
if rec.Code != tc.wantStatus {
t.Fatalf("status = %d, want %d; body=%s", rec.Code, tc.wantStatus, rec.Body.String())
}
if rec.Header().Get("Retry-After") != "5" {
t.Fatalf("Retry-After = %q, want 5", rec.Header().Get("Retry-After"))
}
if !bytes.Contains(rec.Body.Bytes(), []byte(tc.wantCode)) || bytes.Contains(rec.Body.Bytes(), []byte("served")) {
t.Fatalf("body = %q, want error code and no served bytes", rec.Body.String())
}
if svc.gotTransfer.ID != "" {
t.Fatalf("service was called with transfer %+v", svc.gotTransfer)
}
})
}
}
}
}
func TestDownloadTransferResolutionFailureNeverRegisters(t *testing.T) {
registry := transfers.New()
svc := &fakeDownloadService{directErr: catalog.ErrItemNotFound}
internal/api/handlers/ebook_reader.go
+12 -1
View File
@@ -165,11 +165,22 @@ func (h *EbookReaderHandler) HandleReadFile(w http.ResponseWriter, r *http.Reque
}
if h.Transfers != nil {
if err := h.Transfers.Begin(transfer); err != nil {
slog.DebugContext(r.Context(), "ebook transfer not monitored",
slog.DebugContext(r.Context(), "ebook transfer rejected",
"component", "ebooks",
"transfer_id", transfer.ID,
"error", err,
)
switch {
case errors.Is(err, transfers.ErrUserTransferLimit):
w.Header().Set("Retry-After", "5")
writeError(w, http.StatusTooManyRequests, "transfer_limit_exceeded", "Concurrent transfer limit exceeded")
case errors.Is(err, transfers.ErrRegistryFull):
w.Header().Set("Retry-After", "5")
writeError(w, http.StatusServiceUnavailable, "monitoring_unavailable", "Transfer monitoring unavailable")
default:
writeError(w, http.StatusInternalServerError, "internal_error", "Failed to monitor transfer")
}
return
}
}
defer h.Transfers.End(transfer.ID)
internal/api/handlers/ebook_reader_test.go
+74
View File
@@ -210,6 +210,80 @@ func TestEbookReaderServesEbookInlineWithRangeSupport(t *testing.T) {
}
}
func TestEbookReaderTransferAdmissionFailsClosed(t *testing.T) {
for _, tc := range []struct {
name string
registry func(t *testing.T) *transfers.Registry
wantStatus int
wantCode string
}{
{
name: "per-user cap",
registry: func(t *testing.T) *transfers.Registry {
t.Helper()
limit := 1
r := transfers.NewWithOptions(transfers.Options{MaxPerUser: &limit})
if err := r.Begin(transfers.Transfer{ID: "existing", UserID: 1}); err != nil {
t.Fatal(err)
}
return r
},
wantStatus: http.StatusTooManyRequests,
wantCode: "transfer_limit_exceeded",
},
{
name: "global cap",
registry: func(t *testing.T) *transfers.Registry {
t.Helper()
unlimited := 0
r := transfers.NewWithOptions(transfers.Options{MaxEntries: 1, MaxPerUser: &unlimited})
if err := r.Begin(transfers.Transfer{ID: "existing", UserID: 2}); err != nil {
t.Fatal(err)
}
return r
},
wantStatus: http.StatusServiceUnavailable,
wantCode: "monitoring_unavailable",
},
} {
for _, method := range []string{http.MethodGet, http.MethodHead, "range"} {
t.Run(tc.name+"/"+method, func(t *testing.T) {
registry := tc.registry(t)
handler := NewEbookReaderHandler(&MediaFileAuthorizer{
FileResolver: stubMediaFileResolver{file: &models.MediaFile{
ID: 42, ContentID: "ebook-1", FilePath: "must-not-be-served.epub", Container: "epub", BaseType: "ebook",
}},
ItemAccess: stubItemAccessChecker{},
})
handler.Transfers = registry
requestMethod := method
if method == "range" {
requestMethod = http.MethodGet
}
req := withEbookReaderRouteParams(
newEbookReaderAuthRequest(requestMethod, "/ebooks/ebook-1/files/42/read"),
"ebook-1",
"42",
)
if method == "range" {
req.Header.Set("Range", "bytes=0-1")
}
rec := httptest.NewRecorder()
handler.HandleReadFile(rec, req)
if rec.Code != tc.wantStatus {
t.Fatalf("status = %d, want %d; body=%s", rec.Code, tc.wantStatus, rec.Body.String())
}
if rec.Header().Get("Retry-After") != "5" {
t.Fatalf("Retry-After = %q, want 5", rec.Header().Get("Retry-After"))
}
if !strings.Contains(rec.Body.String(), tc.wantCode) || strings.Contains(rec.Body.String(), "must-not-be-served") {
t.Fatalf("body = %q, want error code and no served bytes", rec.Body.String())
}
})
}
}
}
type blockingEbookWriter struct {
header http.Header
reached chan struct{}