feat(podcasts): concurrent download pool — fair share, tickets, admin tunables
Replaces the sequential worker with a download-manager-style pool (§0 item 4). Model: the user's connection is the limit; semantics are fair-share ACROSS feeds, not goroutine-per-book (a global FIFO would make Book B wait behind Book A's 100 files). - Atomic claim: GetNextQueuedEpisode + SetPodcastEpisodeDownloading was a read-then-write race under concurrency; ClaimQueuedPodcastEpisode converts queued→downloading in ONE guarded statement (two claimants, one winner — the loser gets ErrNoRows). SetPodcastEpisodeDownloading retired. - Queue tickets: download_queue_pos BIGINT fed by a Postgres SEQUENCE stamped at enqueue — the enqueue loop's ordering IS the claim order (the foundation §0 item 5 builds on; previously all rows in a batch shared one updated_at, so within-feed order was arbitrary). - Pool: up to podcast_download_concurrency (default 4, live range 1–16) claim-loop goroutines per drain round; the setting re-reads every round so admin changes take effect without restart. - Fair share: claim prefers the feed least-recently-served (pickFeedToServe; never-served first), earliest ticket within a feed — books proceed in parallel while chapters land in order (start listening while the rest download). - Bandwidth: shared token bucket across all workers (podcast_download_speed_mbps, default 0 = unlimited); take() reserves bandwidth and callers sleep outside the lock. - Settings: new float setting type end to end (registry defaults + clamping, save validation, admin UI number-with-step input) + the two schema seeds; worker wires the registry via SetSettings (tests run on registered defaults). - Tests: unit coverage for the bucket reservation math + fair-share pick; integration TestPodcastDownloadPool pins ticket order = enqueue order, claim exclusivity, and a full 3-episode pool drain. Dev-DB note: the column/sequence/settings were applied live (idempotent DDL, same statements the schema runs at next boot) since the running container predates them.
This commit is contained in:
John O'Keefe committed
1 parent
e3675536aa
commit
0f3c4a8cec
14 files changed
+893
-264
No files matched your search
@@ -1,11 +1,21 @@
|
||||
// Podcast download worker (plan §7 / D6 / D8 / D19): ONE sequential
|
||||
// worker (day one) driving the episode state machine
|
||||
// Podcast download worker (plan §7 / D6 / D8 / D19; pool since §0 item
|
||||
// 4): a download-manager-style pool driving the episode state machine
|
||||
//
|
||||
// pending → queued → downloading → downloaded
|
||||
// ↘ failed (+ download_error)
|
||||
//
|
||||
// queued rows are the DB-backed worklist — the state is persisted so
|
||||
// the episode list ALWAYS renders truth (no in-memory queue to lose).
|
||||
// Each queue ticket (download_queue_pos, a SEQUENCE nextval at
|
||||
// enqueue) makes the enqueue loop's ordering the download order.
|
||||
//
|
||||
// Concurrency: up to podcast_download_concurrency episodes in flight
|
||||
// server-wide (default 4, admin-tunable live 1–16). Fair share: the
|
||||
// claim prefers the feed least-recently-served, so several books
|
||||
// download in parallel while chapters within a book land in ticket
|
||||
// order (start listening while the rest catch up). Bandwidth: an
|
||||
// optional shared token bucket (podcast_download_speed_mbps, 0 =
|
||||
// unlimited) paces all workers together.
|
||||
//
|
||||
// Idempotency: a download request for an episode whose file already
|
||||
// exists and probes cleanly re-marks it downloaded without re-fetching
|
||||
@@ -29,10 +39,10 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -48,25 +58,64 @@ const maxAutoDownloadAttempts = 5
|
||||
|
||||
// PodcastDownloadWorker processes queued episode downloads.
|
||||
type PodcastDownloadWorker struct {
|
||||
db *database.Queries
|
||||
svc *PodcastService
|
||||
wake chan struct{}
|
||||
db *database.Queries
|
||||
svc *PodcastService
|
||||
settings *database.SettingsRegistry
|
||||
wake chan struct{}
|
||||
|
||||
// Fair-share bookkeeping (§0 item 4): feed UUID string → last
|
||||
// time the pool served it. Claim prefers the smallest value;
|
||||
// never-served feeds (missing key) go first. Guarded by fairMu.
|
||||
fairMu sync.Mutex
|
||||
lastServed map[string]time.Time
|
||||
|
||||
// Shared bandwidth bucket across all pool workers (§0 item 4).
|
||||
limiter bandwidthLimiter
|
||||
|
||||
// Live byte-progress slots for in-flight downloads (issue #1).
|
||||
// Keyed by episode UUID string; guarded by progMu.
|
||||
progMu sync.Mutex
|
||||
progMu sync.Mutex
|
||||
progress map[string]*episodeProgress
|
||||
}
|
||||
|
||||
func NewPodcastDownloadWorker(db *database.Queries, svc *PodcastService) *PodcastDownloadWorker {
|
||||
return &PodcastDownloadWorker{
|
||||
db: db,
|
||||
svc: svc,
|
||||
wake: make(chan struct{}, 1),
|
||||
progress: make(map[string]*episodeProgress),
|
||||
db: db,
|
||||
svc: svc,
|
||||
wake: make(chan struct{}, 1),
|
||||
lastServed: make(map[string]time.Time),
|
||||
progress: make(map[string]*episodeProgress),
|
||||
}
|
||||
}
|
||||
|
||||
// SetSettings wires the live settings registry (admin tunables:
|
||||
// concurrency + bandwidth cap). Without it the worker runs on the
|
||||
// registered defaults (4 in flight, unlimited speed) — the test
|
||||
// harness constructs workers without settings.
|
||||
func (w *PodcastDownloadWorker) SetSettings(r *database.SettingsRegistry) {
|
||||
w.settings = r
|
||||
}
|
||||
|
||||
// concurrency re-reads the live setting each drain cycle.
|
||||
func (w *PodcastDownloadWorker) concurrency() int {
|
||||
if w.settings == nil {
|
||||
return 4 // registered default
|
||||
}
|
||||
if n := w.settings.PodcastDownloadConcurrency(); n >= 1 {
|
||||
return n
|
||||
}
|
||||
return 4
|
||||
}
|
||||
|
||||
// refillLimiter re-syncs the shared bucket's rate (bytes/sec) from the
|
||||
// live setting; 0 = unlimited.
|
||||
func (w *PodcastDownloadWorker) refillLimiter() {
|
||||
if w.settings == nil {
|
||||
return
|
||||
}
|
||||
w.limiter.setRate(int64(w.settings.PodcastDownloadSpeedMbps() * 125000)) // Mbps → bytes/sec
|
||||
}
|
||||
|
||||
// Wake nudges the loop (new work queued).
|
||||
func (w *PodcastDownloadWorker) Wake() {
|
||||
select {
|
||||
@@ -93,21 +142,129 @@ func (w *PodcastDownloadWorker) Start(ctx context.Context) {
|
||||
}()
|
||||
}
|
||||
|
||||
// drain processes queued episodes until the queue is empty.
|
||||
// DrainNow runs one bounded pool round synchronously — the test hook
|
||||
// and the future admin "retry now" seam (normally the wake channel
|
||||
// drives rounds).
|
||||
func (w *PodcastDownloadWorker) DrainNow(ctx context.Context) {
|
||||
w.drain(ctx)
|
||||
}
|
||||
|
||||
// drain runs one pool round: up to concurrency() claim-loop goroutines
|
||||
// each claim (fair-share) → download → settle, until the queue is
|
||||
// empty. The cap setting is re-read every round — live-tunable.
|
||||
func (w *PodcastDownloadWorker) drain(ctx context.Context) {
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
acted, err := w.processNext(ctx)
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
w.refillLimiter()
|
||||
limit := w.concurrency()
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < limit; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
ep, ok := w.claimNext(ctx)
|
||||
if !ok {
|
||||
return // queue drained (or nothing claimable)
|
||||
}
|
||||
w.runOne(ctx, ep)
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
// claimNext atomically claims the next queued episode, preferring the
|
||||
// feed least-recently-served (fair share — books proceed in parallel)
|
||||
// and the earliest queue ticket within a feed (chapters land in
|
||||
// enqueue order). Returns ok=false when nothing is claimable.
|
||||
func (w *PodcastDownloadWorker) claimNext(ctx context.Context) (database.MediaItems, bool) {
|
||||
actx, err := w.svc.adminCtx(ctx)
|
||||
if err != nil {
|
||||
return database.MediaItems{}, false
|
||||
}
|
||||
cands, err := w.db.ListQueuedPodcastEpisodes(actx)
|
||||
if err != nil || len(cands) == 0 {
|
||||
return database.MediaItems{}, false
|
||||
}
|
||||
|
||||
// Group per feed preserving ticket order (the SQL scan is ordered).
|
||||
perFeed := make(map[pgtype.UUID][]pgtype.UUID, 8)
|
||||
for _, c := range cands {
|
||||
perFeed[c.SourceFeedID] = append(perFeed[c.SourceFeedID], c.ID)
|
||||
}
|
||||
|
||||
w.fairMu.Lock()
|
||||
defer w.fairMu.Unlock()
|
||||
pickFeed := pickFeedToServe(perFeed, w.lastServed)
|
||||
if !pickFeed.Valid {
|
||||
return database.MediaItems{}, false
|
||||
}
|
||||
|
||||
// Claim the head ticket of the chosen feed; if another claimant
|
||||
// won it meanwhile (retry raced), fall through the feed's tickets.
|
||||
for _, id := range perFeed[pickFeed] {
|
||||
ep, err := w.db.ClaimQueuedPodcastEpisode(actx, id)
|
||||
if err != nil {
|
||||
log.Printf("[PODCAST-DL] worker step: %v", err)
|
||||
return
|
||||
continue // claimed elsewhere / state moved on
|
||||
}
|
||||
if !acted {
|
||||
return
|
||||
w.lastServed[pickFeed.String()] = time.Now()
|
||||
return ep, true
|
||||
}
|
||||
return database.MediaItems{}, false
|
||||
}
|
||||
|
||||
// pickFeedToServe chooses the feed least-recently-served (never-served
|
||||
// first); among never-served feeds the first encountered wins (any is
|
||||
// equally fair). Unit-testable seam for the fair-share rule.
|
||||
func pickFeedToServe(perFeed map[pgtype.UUID][]pgtype.UUID, lastServed map[string]time.Time) pgtype.UUID {
|
||||
var pick pgtype.UUID
|
||||
var best time.Time
|
||||
haveNever := false
|
||||
for feed := range perFeed {
|
||||
last, seen := lastServed[feed.String()]
|
||||
if !seen {
|
||||
if !haveNever {
|
||||
pick, haveNever = feed, true
|
||||
}
|
||||
continue
|
||||
}
|
||||
if haveNever {
|
||||
continue // a never-served feed already won
|
||||
}
|
||||
if !pick.Valid || last.Before(best) {
|
||||
pick, best = feed, last
|
||||
}
|
||||
}
|
||||
return pick
|
||||
}
|
||||
|
||||
// runOne downloads a claimed episode and settles the row either way
|
||||
// (failed/rejected: state + reason persist; metadata/file facts
|
||||
// untouched — invariant 7, write-after-success only).
|
||||
func (w *PodcastDownloadWorker) runOne(ctx context.Context, ep database.MediaItems) {
|
||||
actx, err := w.svc.adminCtx(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if err := w.runPipeline(actx, ep); err != nil {
|
||||
_ = w.db.MarkPodcastEpisodeFailed(actx, database.MarkPodcastEpisodeFailedParams{
|
||||
ID: ep.ID,
|
||||
DownloadError: pgtypeText(err.Error()),
|
||||
Column3: false, // auto failure: never resets the attempt budget
|
||||
})
|
||||
log.Printf("[PODCAST-DL] episode %s failed: %v", ep.ID, err)
|
||||
}
|
||||
}
|
||||
|
||||
// AdminCtx exposes the worker's first-admin identity (tests use it for
|
||||
// worker-scope DB statements).
|
||||
func (w *PodcastDownloadWorker) AdminCtx(ctx context.Context) (context.Context, error) {
|
||||
return w.svc.adminCtx(ctx)
|
||||
}
|
||||
|
||||
// QueueEpisode marks one episode queued (manual download / retry path).
|
||||
@@ -228,33 +385,6 @@ func (w *PodcastDownloadWorker) ApplyAutoPolicies(ctx context.Context, feedID pg
|
||||
return queued, nil
|
||||
}
|
||||
|
||||
// processNext claims and processes one queued episode. Returns false
|
||||
// when the queue was empty.
|
||||
func (w *PodcastDownloadWorker) processNext(ctx context.Context) (bool, error) {
|
||||
actx, err := w.svc.adminCtx(ctx)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
ep, err := w.db.GetNextQueuedEpisode(actx)
|
||||
if err != nil {
|
||||
return false, nil // empty queue (or read failure — logged upstream)
|
||||
}
|
||||
|
||||
_ = w.db.SetPodcastEpisodeDownloading(actx, ep.ID)
|
||||
|
||||
if err := w.runPipeline(actx, ep); err != nil {
|
||||
// failed/rejected: state + reason persist; metadata/file facts
|
||||
// untouched (invariant 7 — write-after-success only).
|
||||
_ = w.db.MarkPodcastEpisodeFailed(actx, database.MarkPodcastEpisodeFailedParams{
|
||||
ID: ep.ID,
|
||||
DownloadError: pgtypeText(err.Error()),
|
||||
Column3: false, // auto failure: never resets the attempt budget
|
||||
})
|
||||
log.Printf("[PODCAST-DL] episode %s failed: %v", ep.ID, err)
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// runPipeline: fetch enclosure (SSRF/size guards) → temp file → MIME/
|
||||
// size validation → ffprobe reconcile → move into place → SUCCESS-only
|
||||
// row update.
|
||||
@@ -398,6 +528,42 @@ func (w *PodcastDownloadWorker) EpisodeProgress(episodeID pgtype.UUID) (received
|
||||
return p.Received, p.Total, true
|
||||
}
|
||||
|
||||
// ---- shared bandwidth bucket (§0 item 4) ----
|
||||
// One leaky-bucket-with-reservations limiter across ALL pool workers:
|
||||
// each caller reserves the next slot of bandwidth and sleeps outside
|
||||
// the lock — reservation order = transmission order, and N workers
|
||||
// share one cap instead of N caps.
|
||||
|
||||
type bandwidthLimiter struct {
|
||||
mu sync.Mutex
|
||||
rate int64 // bytes per second; <=0 = unlimited
|
||||
next time.Time // when the currently-reserved budget runs out
|
||||
}
|
||||
|
||||
func (l *bandwidthLimiter) setRate(bytesPerSec int64) {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
l.rate = bytesPerSec
|
||||
}
|
||||
|
||||
// take reserves n bytes of bandwidth and returns how long the caller
|
||||
// must wait before sending them (0 when free flow).
|
||||
func (l *bandwidthLimiter) take(n int64) time.Duration {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
if l.rate <= 0 {
|
||||
return 0
|
||||
}
|
||||
now := time.Now()
|
||||
if l.next.Before(now) {
|
||||
l.next = now
|
||||
}
|
||||
d := time.Duration(float64(n) / float64(l.rate) * float64(time.Second))
|
||||
start := l.next // my window begins when earlier reservations end
|
||||
l.next = start.Add(d)
|
||||
return start.Sub(now)
|
||||
}
|
||||
|
||||
// streamEnclosure performs the SSRF-guarded enclosure download straight
|
||||
// to tmpPath while updating the progress slot. Returns bytes received
|
||||
// and the authoritative total (0 when unknown). Scheme and size guards
|
||||
@@ -456,6 +622,15 @@ func (w *PodcastDownloadWorker) streamEnclosure(ctx context.Context, ep database
|
||||
if received > maxEnclosureCapBytes {
|
||||
return received, total, fmt.Errorf("rejected: enclosure exceeds %d byte cap", maxEnclosureCapBytes)
|
||||
}
|
||||
// Shared bandwidth cap: pace against the server-wide
|
||||
// bucket (0 rate = unlimited, take returns 0).
|
||||
if wait := w.limiter.take(int64(n)); wait > 0 {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return received, total, fmt.Errorf("download failed: cancelled")
|
||||
case <-time.After(wait):
|
||||
}
|
||||
}
|
||||
w.addProgress(ep.ID, int64(n))
|
||||
if _, werr := out.Write(buf[:n]); werr != nil {
|
||||
return received, total, fmt.Errorf("write: %w", werr)
|
||||
@@ -496,13 +671,13 @@ func (w *PodcastDownloadWorker) markDownloaded(ctx context.Context, ep database.
|
||||
}
|
||||
|
||||
_, err = w.db.MarkPodcastEpisodeDownloaded(ctx, database.MarkPodcastEpisodeDownloadedParams{
|
||||
ID: ep.ID,
|
||||
FilePath: ep.FilePath,
|
||||
FileSize: pgtypeInt8(info.Size()),
|
||||
MimeType: pgtypeText(strings.TrimSpace(pickMimeType(ep, absPath))),
|
||||
ID: ep.ID,
|
||||
FilePath: ep.FilePath,
|
||||
FileSize: pgtypeInt8(info.Size()),
|
||||
MimeType: pgtypeText(strings.TrimSpace(pickMimeType(ep, absPath))),
|
||||
CoverImagePath: coverRel,
|
||||
AudioMetadata: audioPayloadJSONB(payload),
|
||||
ChapterCount: chapterCount,
|
||||
AudioMetadata: audioPayloadJSONB(payload),
|
||||
ChapterCount: chapterCount,
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("mark downloaded: %w", err)
|
||||
|
||||
Reference in new issue
Block a user