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:
1 parent
e3675536aa
commit
0f3c4a8cec
14 files changed
+874
-245
No files matched your search
+4
-2
@@ -179,10 +179,12 @@ func main() {
|
|||||||
matchingHandler := handlers.NewMatchingHandler(queries, connManager)
|
matchingHandler := handlers.NewMatchingHandler(queries, connManager)
|
||||||
jobsHandler := handlers.NewJobsHandler(queries, worker)
|
jobsHandler := handlers.NewJobsHandler(queries, worker)
|
||||||
|
|
||||||
// Podcast subsystem: fetcher service + sequential download worker +
|
// Podcast subsystem: fetcher service + concurrent download worker
|
||||||
// poller. One instance trio per server (per-feed locks dedupe).
|
// (§0 item 4: pool, fair share, admin tunables) + poller. One
|
||||||
|
// instance trio per server (per-feed locks dedupe).
|
||||||
podcastService := services.NewPodcastService(queries)
|
podcastService := services.NewPodcastService(queries)
|
||||||
podcastWorker := services.NewPodcastDownloadWorker(queries, podcastService)
|
podcastWorker := services.NewPodcastDownloadWorker(queries, podcastService)
|
||||||
|
podcastWorker.SetSettings(registry)
|
||||||
podcastHandler := handlers.NewPodcastHandler(queries, podcastService, podcastWorker)
|
podcastHandler := handlers.NewPodcastHandler(queries, podcastService, podcastWorker)
|
||||||
|
|
||||||
e := echo.New()
|
e := echo.New()
|
||||||
|
|||||||
@@ -761,6 +761,126 @@ func TestPodcastReverseOrder(t *testing.T) {
|
|||||||
"position 3 first — a messy-date feed reorders by position, not dates")
|
"position 3 first — a messy-date feed reorders by position, not dates")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ---- §0 item 4: concurrent pool — tickets, exclusivity, drain ----
|
||||||
|
|
||||||
|
func TestPodcastDownloadPool(t *testing.T) {
|
||||||
|
setup := setupTestServer(t)
|
||||||
|
if _, err := exec.LookPath("ffmpeg"); err != nil {
|
||||||
|
if _, err := os.Stat("/usr/bin/ffmpeg"); err != nil {
|
||||||
|
t.Skip("ffmpeg unavailable: worker pipeline test needs runtime-generated audio")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
lib, _ := createPodcastsLibrary(t, setup, "test Pool Podcasts")
|
||||||
|
|
||||||
|
mp3Path := filepath.Join(t.TempDir(), "tone.mp3")
|
||||||
|
gen := exec.Command("ffmpeg", "-y", "-v", "error", "-f", "lavfi",
|
||||||
|
"-i", "sine=frequency=440:duration=1", "-b:a", "32k", mp3Path)
|
||||||
|
out, err := gen.CombinedOutput()
|
||||||
|
require.NoError(t, err, "ffmpeg generate: %s", string(out))
|
||||||
|
audio, err := os.ReadFile(mp3Path)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
mux := http.NewServeMux()
|
||||||
|
mux.HandleFunc("/audio/", func(w http.ResponseWriter, _ *http.Request) {
|
||||||
|
w.Header().Set("Content-Type", "audio/mpeg")
|
||||||
|
_, _ = w.Write(audio)
|
||||||
|
})
|
||||||
|
mux.HandleFunc("/feed.xml", func(w http.ResponseWriter, _ *http.Request) {
|
||||||
|
w.Header().Set("Content-Type", "application/rss+xml")
|
||||||
|
_, _ = w.Write([]byte(podcastFixture("Pool Feed", map[string]string{
|
||||||
|
"pl-1": "Pool One", "pl-2": "Pool Two", "pl-3": "Pool Three",
|
||||||
|
})))
|
||||||
|
})
|
||||||
|
srv := swapFeedServer(t, mux.ServeHTTP)
|
||||||
|
|
||||||
|
libID := uuid.UUID(lib.ID.Bytes).String()
|
||||||
|
resp, body := podcastAuthed(t, setup, setup.Token, "POST", "/api/podcasts/feeds",
|
||||||
|
fmt.Sprintf(`{"library_id":%q,"url":%q}`, libID, srv.URL+"/feed.xml"))
|
||||||
|
require.Equal(t, http.StatusCreated, resp, string(body))
|
||||||
|
|
||||||
|
adminCtx := adminSeedCtx(t, setup)
|
||||||
|
feeds, err := setup.DB.ListPodcastFeedsForPolling(adminCtx)
|
||||||
|
require.NoError(t, err)
|
||||||
|
var feedRow database.PodcastFeeds
|
||||||
|
for _, f := range feeds {
|
||||||
|
if f.Title == "Pool Feed" {
|
||||||
|
feedRow = f
|
||||||
|
}
|
||||||
|
}
|
||||||
|
require.NotZero(t, feedRow.ID, "feed must exist after add")
|
||||||
|
eps, err := setup.DB.ListPodcastEpisodes(adminCtx, feedRow.ID)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.Len(t, eps, 3)
|
||||||
|
|
||||||
|
// Point enclosures at the test server (admin-declared write).
|
||||||
|
adminScope := database.NewScopedDB(setup.DBPool)
|
||||||
|
_, err = adminScope.Exec(rls.WithUser(context.Background(), adminCtxValue(t, setup)),
|
||||||
|
`UPDATE media_items SET enclosure_url = $2 WHERE source_feed_id = $1`,
|
||||||
|
feedRow.ID, srv.URL+"/audio/ep.mp3")
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
worker := services.NewPodcastDownloadWorker(setup.DB, services.NewPodcastService(setup.DB))
|
||||||
|
actx, err := worker.AdminCtx(context.Background())
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
// Queue in a KNOWN order: tickets (download_queue_pos SEQUENCE) make
|
||||||
|
// the enqueue loop's ordering the claim order — the §0 item 5
|
||||||
|
// foundation ("getting the last chapter first doesn't help me").
|
||||||
|
ordered := make([]database.MediaItems, 0, len(eps))
|
||||||
|
for _, wantPos := range []int32{1, 2, 3} {
|
||||||
|
for _, ep := range eps {
|
||||||
|
if ep.FeedPosition.Int32 == wantPos {
|
||||||
|
ordered = append(ordered, ep)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for _, ep := range ordered {
|
||||||
|
_, err := setup.DB.SetPodcastEpisodeDownloadQueued(actx, ep.ID)
|
||||||
|
require.NoError(t, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
queued, err := setup.DB.ListQueuedPodcastEpisodes(actx)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.Len(t, queued, 3)
|
||||||
|
for i, ep := range ordered {
|
||||||
|
assert.Equal(t, ep.ID, queued[i].ID,
|
||||||
|
"PIN: ticket order = enqueue order (claim claims in enqueue order)")
|
||||||
|
}
|
||||||
|
|
||||||
|
t.Run("claim_is_exclusive", func(t *testing.T) {
|
||||||
|
first, err := setup.DB.ClaimQueuedPodcastEpisode(actx, queued[0].ID)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, "downloading", first.DownloadState)
|
||||||
|
// A second claimant racing the same row gets NOTHING — the
|
||||||
|
// single-statement guard (queued → downloading) is the
|
||||||
|
// SKIP-LOCKED-equivalent guarantee.
|
||||||
|
_, err = setup.DB.ClaimQueuedPodcastEpisode(actx, queued[0].ID)
|
||||||
|
assert.Error(t, err, "second claim of a downloading row must not match")
|
||||||
|
|
||||||
|
// Hand it back for the drain below (enqueue only takes
|
||||||
|
// pending|failed — a stuck 'downloading' needs the manual reset,
|
||||||
|
// exactly what the crash-recovery path does).
|
||||||
|
require.NoError(t, setup.DB.SetPodcastEpisodePending(actx, queued[0].ID))
|
||||||
|
_, err = setup.DB.SetPodcastEpisodeDownloadQueued(actx, queued[0].ID)
|
||||||
|
require.NoError(t, err)
|
||||||
|
})
|
||||||
|
|
||||||
|
worker.DrainNow(context.Background())
|
||||||
|
require.Eventually(t, func() bool {
|
||||||
|
rowset, err := setup.DB.ListPodcastEpisodes(adminCtx, feedRow.ID)
|
||||||
|
if err != nil {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
for _, row := range rowset {
|
||||||
|
if row.DownloadState != "downloaded" {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return len(rowset) == 3
|
||||||
|
}, 30*time.Second, 250*time.Millisecond, "all 3 episodes must download through the pool")
|
||||||
|
}
|
||||||
|
|
||||||
// ---- issue #8: page search hits EPISODE titles/descriptions too ----
|
// ---- issue #8: page search hits EPISODE titles/descriptions too ----
|
||||||
|
|
||||||
func TestPodcastPageSearchEpisodes(t *testing.T) {
|
func TestPodcastPageSearchEpisodes(t *testing.T) {
|
||||||
|
|||||||
@@ -102,7 +102,9 @@ INSERT INTO system_settings (setting_key, setting_value, description, setting_ty
|
|||||||
('worker_pool_size', '3', 'Number of background worker goroutines', 'int', '1', '100', TRUE, 'performance'),
|
('worker_pool_size', '3', 'Number of background worker goroutines', 'int', '1', '100', TRUE, 'performance'),
|
||||||
('worker_queue_cap', '100', 'Background worker job queue capacity', 'int', '1', '10000', TRUE, 'performance'),
|
('worker_queue_cap', '100', 'Background worker job queue capacity', 'int', '1', '10000', TRUE, 'performance'),
|
||||||
-- podcasts
|
-- podcasts
|
||||||
('podcast_poll_interval_seconds', '3600', 'How often to poll podcast feeds for new episodes (seconds; complete feeds back off)', 'int', '300', '604800', FALSE, 'podcasts')
|
('podcast_poll_interval_seconds', '3600', 'How often to poll podcast feeds for new episodes (seconds; complete feeds back off)', 'int', '300', '604800', FALSE, 'podcasts'),
|
||||||
|
('podcast_download_concurrency', '4', 'Podcast episode downloads in flight server-wide (the connection is the limit; fair-share spreads them across feeds)', 'int', '1', '16', FALSE, 'podcasts'),
|
||||||
|
('podcast_download_speed_mbps', '0', 'Shared bandwidth cap for podcast downloads in megabits/second (0 = unlimited)', 'float', '0', '10000', FALSE, 'podcasts')
|
||||||
ON CONFLICT (setting_key) DO NOTHING;
|
ON CONFLICT (setting_key) DO NOTHING;
|
||||||
|
|
||||||
-- Create refresh_tokens table
|
-- Create refresh_tokens table
|
||||||
@@ -1989,7 +1991,16 @@ ALTER TABLE media_items
|
|||||||
-- list renders this inline (D8/D19) — never Processing Issues.
|
-- list renders this inline (D8/D19) — never Processing Issues.
|
||||||
ADD COLUMN IF NOT EXISTS download_state VARCHAR(20) NOT NULL DEFAULT 'pending',
|
ADD COLUMN IF NOT EXISTS download_state VARCHAR(20) NOT NULL DEFAULT 'pending',
|
||||||
ADD COLUMN IF NOT EXISTS download_error TEXT,
|
ADD COLUMN IF NOT EXISTS download_error TEXT,
|
||||||
ADD COLUMN IF NOT EXISTS download_attempts INT NOT NULL DEFAULT 0;
|
ADD COLUMN IF NOT EXISTS download_attempts INT NOT NULL DEFAULT 0,
|
||||||
|
-- Monotonic queue ticket (Postgres SEQUENCE nextval at enqueue
|
||||||
|
-- time): the download pool claims in ticket order, so the enqueue
|
||||||
|
-- loop's ordering IS the download order (effective listening order
|
||||||
|
-- — §0 items 4/5). NULL for rows never queued.
|
||||||
|
ADD COLUMN IF NOT EXISTS download_queue_pos BIGINT;
|
||||||
|
|
||||||
|
-- Queue ticket source: atomic across concurrent enqueues (per-feed
|
||||||
|
-- locks serialize within a feed; the sequence covers across feeds).
|
||||||
|
CREATE SEQUENCE IF NOT EXISTS podcast_queue_seq;
|
||||||
|
|
||||||
-- Identity + ordering + worker-worklist indexes (all partial: scan
|
-- Identity + ordering + worker-worklist indexes (all partial: scan
|
||||||
-- rows never touch these columns).
|
-- rows never touch these columns).
|
||||||
@@ -1999,6 +2010,8 @@ CREATE INDEX IF NOT EXISTS idx_media_items_feed_pos
|
|||||||
ON media_items(source_feed_id, feed_position) WHERE source_feed_id IS NOT NULL;
|
ON media_items(source_feed_id, feed_position) WHERE source_feed_id IS NOT NULL;
|
||||||
CREATE INDEX IF NOT EXISTS idx_media_items_feed_undownloaded
|
CREATE INDEX IF NOT EXISTS idx_media_items_feed_undownloaded
|
||||||
ON media_items(source_feed_id) WHERE source_feed_id IS NOT NULL AND download_state <> 'downloaded';
|
ON media_items(source_feed_id) WHERE source_feed_id IS NOT NULL AND download_state <> 'downloaded';
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_media_items_queue
|
||||||
|
ON media_items(download_queue_pos) WHERE source = 'podcast_episode' AND download_state = 'queued';
|
||||||
|
|
||||||
|
|
||||||
-- ============================================================
|
-- ============================================================
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
// Code generated by sqlc. DO NOT EDIT.
|
// Code generated by sqlc. DO NOT EDIT.
|
||||||
// versions:
|
// versions:
|
||||||
// sqlc v1.31.1
|
// sqlc v1.31.0
|
||||||
|
|
||||||
package database
|
package database
|
||||||
|
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
// Code generated by sqlc. DO NOT EDIT.
|
// Code generated by sqlc. DO NOT EDIT.
|
||||||
// versions:
|
// versions:
|
||||||
// sqlc v1.31.1
|
// sqlc v1.31.0
|
||||||
|
|
||||||
package database
|
package database
|
||||||
|
|
||||||
@@ -355,6 +355,7 @@ type MediaItems struct {
|
|||||||
DownloadState string `db:"download_state" json:"download_state"`
|
DownloadState string `db:"download_state" json:"download_state"`
|
||||||
DownloadError pgtype.Text `db:"download_error" json:"download_error"`
|
DownloadError pgtype.Text `db:"download_error" json:"download_error"`
|
||||||
DownloadAttempts int32 `db:"download_attempts" json:"download_attempts"`
|
DownloadAttempts int32 `db:"download_attempts" json:"download_attempts"`
|
||||||
|
DownloadQueuePos pgtype.Int8 `db:"download_queue_pos" json:"download_queue_pos"`
|
||||||
}
|
}
|
||||||
|
|
||||||
type MediaNotes struct {
|
type MediaNotes struct {
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
// Code generated by sqlc. DO NOT EDIT.
|
// Code generated by sqlc. DO NOT EDIT.
|
||||||
// versions:
|
// versions:
|
||||||
// sqlc v1.31.1
|
// sqlc v1.31.0
|
||||||
|
|
||||||
package database
|
package database
|
||||||
|
|
||||||
@@ -32,6 +32,11 @@ type Querier interface {
|
|||||||
BulkUpdateFormatGroups(ctx context.Context) error
|
BulkUpdateFormatGroups(ctx context.Context) error
|
||||||
BulkUpdateProgressFromSync(ctx context.Context, arg BulkUpdateProgressFromSyncParams) ([]interface{}, error)
|
BulkUpdateProgressFromSync(ctx context.Context, arg BulkUpdateProgressFromSyncParams) ([]interface{}, error)
|
||||||
CheckForProgressConflicts(ctx context.Context, arg CheckForProgressConflictsParams) (int64, error)
|
CheckForProgressConflicts(ctx context.Context, arg CheckForProgressConflictsParams) (int64, error)
|
||||||
|
// Atomic claim (§0 item 4): converts queued -> downloading in ONE
|
||||||
|
// statement. Two claimants racing on the same row: exactly one UPDATE
|
||||||
|
// matches download_state='queued'; the loser gets no rows (ErrNoRows).
|
||||||
|
// This is the SKIP-LOCKED guarantee without a second round-trip.
|
||||||
|
ClaimQueuedPodcastEpisode(ctx context.Context, id pgtype.UUID) (MediaItems, error)
|
||||||
// Cleanup expired OPDS tokens
|
// Cleanup expired OPDS tokens
|
||||||
CleanupExpiredOpdsTokens(ctx context.Context) error
|
CleanupExpiredOpdsTokens(ctx context.Context) error
|
||||||
CleanupExpiredRefreshTokens(ctx context.Context, dollar_1 float64) error
|
CleanupExpiredRefreshTokens(ctx context.Context, dollar_1 float64) error
|
||||||
@@ -446,8 +451,6 @@ type Querier interface {
|
|||||||
// GetMediaScanState reads one item's manifest (per-item rescan paths
|
// GetMediaScanState reads one item's manifest (per-item rescan paths
|
||||||
// and the manifest integration tests).
|
// and the manifest integration tests).
|
||||||
GetMediaScanState(ctx context.Context, mediaItemID pgtype.UUID) (MediaScanState, error)
|
GetMediaScanState(ctx context.Context, mediaItemID pgtype.UUID) (MediaScanState, error)
|
||||||
// Sequential download worker's claim source (single worker, day one).
|
|
||||||
GetNextQueuedEpisode(ctx context.Context) (MediaItems, error)
|
|
||||||
GetNextRetryTime(ctx context.Context) (interface{}, error)
|
GetNextRetryTime(ctx context.Context) (interface{}, error)
|
||||||
GetNotStartedItems(ctx context.Context, arg GetNotStartedItemsParams) ([]MediaItems, error)
|
GetNotStartedItems(ctx context.Context, arg GetNotStartedItemsParams) ([]MediaItems, error)
|
||||||
// Get OPDS token
|
// Get OPDS token
|
||||||
@@ -656,6 +659,11 @@ type Querier interface {
|
|||||||
// Item-side projections COALESCE to keep them non-nullable strings;
|
// Item-side projections COALESCE to keep them non-nullable strings;
|
||||||
// the display file path falls back to the issue's own for row-less rows.
|
// the display file path falls back to the issue's own for row-less rows.
|
||||||
ListProcessingIssuesByLibrary(ctx context.Context, libraryID pgtype.UUID) ([]ListProcessingIssuesByLibraryRow, error)
|
ListProcessingIssuesByLibrary(ctx context.Context, libraryID pgtype.UUID) ([]ListProcessingIssuesByLibraryRow, error)
|
||||||
|
// Fair-share candidate scan (§0 item 4): the download pool claims in
|
||||||
|
// TICKET order (download_queue_pos = SEQUENCE nextval at enqueue — the
|
||||||
|
// enqueue loop's ordering IS the download order, item 5). The worker
|
||||||
|
// groups these per feed and prefers the least-recently-served feed.
|
||||||
|
ListQueuedPodcastEpisodes(ctx context.Context) ([]ListQueuedPodcastEpisodesRow, error)
|
||||||
ListSyncConflictsByMediaItem(ctx context.Context, arg ListSyncConflictsByMediaItemParams) ([]SyncConflicts, error)
|
ListSyncConflictsByMediaItem(ctx context.Context, arg ListSyncConflictsByMediaItemParams) ([]SyncConflicts, error)
|
||||||
ListSyncConflictsByUser(ctx context.Context, userID pgtype.UUID) ([]ListSyncConflictsByUserRow, error)
|
ListSyncConflictsByUser(ctx context.Context, userID pgtype.UUID) ([]ListSyncConflictsByUserRow, error)
|
||||||
// Download All + policy union worklist.
|
// Download All + policy union worklist.
|
||||||
@@ -668,6 +676,9 @@ type Querier interface {
|
|||||||
ListUsers(ctx context.Context) ([]ListUsersRow, error)
|
ListUsers(ctx context.Context) ([]ListUsersRow, error)
|
||||||
// First consecutive scan that cannot find the file on disk.
|
// First consecutive scan that cannot find the file on disk.
|
||||||
MarkMediaItemMissing(ctx context.Context, id pgtype.UUID) error
|
MarkMediaItemMissing(ctx context.Context, id pgtype.UUID) error
|
||||||
|
// SetPodcastEpisodeDownloading RETIRED (§0 item 4): the pair
|
||||||
|
// GetNextQueuedEpisode + SetPodcastEpisodeDownloading raced under
|
||||||
|
// concurrency; ClaimQueuedPodcastEpisode does both atomically.
|
||||||
// SUCCESS-only write (invariant 7): the file is on disk and probed.
|
// SUCCESS-only write (invariant 7): the file is on disk and probed.
|
||||||
// ffprobe facts (real duration via audio_metadata, precise size/mime)
|
// ffprobe facts (real duration via audio_metadata, precise size/mime)
|
||||||
// reconcile the feed's declared values.
|
// reconcile the feed's declared values.
|
||||||
@@ -753,10 +764,11 @@ type Querier interface {
|
|||||||
SetDeviceSyncEnabled(ctx context.Context, arg SetDeviceSyncEnabledParams) (Devices, error)
|
SetDeviceSyncEnabled(ctx context.Context, arg SetDeviceSyncEnabledParams) (Devices, error)
|
||||||
// Library Visibility queries
|
// Library Visibility queries
|
||||||
SetLibraryVisibility(ctx context.Context, arg SetLibraryVisibilityParams) (LibraryVisibility, error)
|
SetLibraryVisibility(ctx context.Context, arg SetLibraryVisibilityParams) (LibraryVisibility, error)
|
||||||
// Worker claim: pending|failed -> queued. Attempts incremented at
|
// Enqueue: pending|failed -> queued, stamped with the NEXT QUEUE TICKET
|
||||||
// claim time so a crashed worker still counts the try.
|
// (§0 items 4/5) so claim order = enqueue order (the effective
|
||||||
|
// listening order the enqueue loop applied). Attempts increment at
|
||||||
|
// enqueue so a crashed worker still counts the try.
|
||||||
SetPodcastEpisodeDownloadQueued(ctx context.Context, id pgtype.UUID) (int64, error)
|
SetPodcastEpisodeDownloadQueued(ctx context.Context, id pgtype.UUID) (int64, error)
|
||||||
SetPodcastEpisodeDownloading(ctx context.Context, id pgtype.UUID) error
|
|
||||||
// Manual retry (same endpoint as download): reset to pending so the
|
// Manual retry (same endpoint as download): reset to pending so the
|
||||||
// worker re-claims it; the reason text clears with the fresh attempt.
|
// worker re-claims it; the reason text clears with the fresh attempt.
|
||||||
SetPodcastEpisodePending(ctx context.Context, id pgtype.UUID) error
|
SetPodcastEpisodePending(ctx context.Context, id pgtype.UUID) error
|
||||||
|
|||||||
+249
-162
File diff suppressed because it is too large.
Load diff
@@ -4862,10 +4862,13 @@ SELECT * FROM media_items
|
|||||||
WHERE id = $1 AND source = 'podcast_episode';
|
WHERE id = $1 AND source = 'podcast_episode';
|
||||||
|
|
||||||
-- name: SetPodcastEpisodeDownloadQueued :execrows
|
-- name: SetPodcastEpisodeDownloadQueued :execrows
|
||||||
-- Worker claim: pending|failed -> queued. Attempts incremented at
|
-- Enqueue: pending|failed -> queued, stamped with the NEXT QUEUE TICKET
|
||||||
-- claim time so a crashed worker still counts the try.
|
-- (§0 items 4/5) so claim order = enqueue order (the effective
|
||||||
|
-- listening order the enqueue loop applied). Attempts increment at
|
||||||
|
-- enqueue so a crashed worker still counts the try.
|
||||||
UPDATE media_items
|
UPDATE media_items
|
||||||
SET download_state = 'queued',
|
SET download_state = 'queued',
|
||||||
|
download_queue_pos = nextval('podcast_queue_seq'),
|
||||||
download_error = NULL,
|
download_error = NULL,
|
||||||
download_attempts = download_attempts + 1,
|
download_attempts = download_attempts + 1,
|
||||||
updated_at = NOW()
|
updated_at = NOW()
|
||||||
@@ -4873,10 +4876,9 @@ WHERE id = $1
|
|||||||
AND source = 'podcast_episode'
|
AND source = 'podcast_episode'
|
||||||
AND download_state IN ('pending', 'failed');
|
AND download_state IN ('pending', 'failed');
|
||||||
|
|
||||||
-- name: SetPodcastEpisodeDownloading :exec
|
-- SetPodcastEpisodeDownloading RETIRED (§0 item 4): the pair
|
||||||
UPDATE media_items
|
-- GetNextQueuedEpisode + SetPodcastEpisodeDownloading raced under
|
||||||
SET download_state = 'downloading', updated_at = NOW()
|
-- concurrency; ClaimQueuedPodcastEpisode does both atomically.
|
||||||
WHERE id = $1 AND source = 'podcast_episode';
|
|
||||||
|
|
||||||
-- name: MarkPodcastEpisodeDownloaded :one
|
-- name: MarkPodcastEpisodeDownloaded :one
|
||||||
-- SUCCESS-only write (invariant 7): the file is on disk and probed.
|
-- SUCCESS-only write (invariant 7): the file is on disk and probed.
|
||||||
@@ -4964,12 +4966,26 @@ WHERE s.user_id = sqlc.arg('user_id')
|
|||||||
OR COALESCE(f.publisher, '') ILIKE '%' || sqlc.arg('search_query')::text || '%')
|
OR COALESCE(f.publisher, '') ILIKE '%' || sqlc.arg('search_query')::text || '%')
|
||||||
ORDER BY COALESCE(s.display_name, f.title) ASC;
|
ORDER BY COALESCE(s.display_name, f.title) ASC;
|
||||||
|
|
||||||
-- name: GetNextQueuedEpisode :one
|
-- name: ListQueuedPodcastEpisodes :many
|
||||||
-- Sequential download worker's claim source (single worker, day one).
|
-- Fair-share candidate scan (§0 item 4): the download pool claims in
|
||||||
SELECT * FROM media_items
|
-- TICKET order (download_queue_pos = SEQUENCE nextval at enqueue — the
|
||||||
|
-- enqueue loop's ordering IS the download order, item 5). The worker
|
||||||
|
-- groups these per feed and prefers the least-recently-served feed.
|
||||||
|
SELECT id, source_feed_id FROM media_items
|
||||||
WHERE source = 'podcast_episode' AND download_state = 'queued'
|
WHERE source = 'podcast_episode' AND download_state = 'queued'
|
||||||
ORDER BY updated_at ASC
|
ORDER BY download_queue_pos ASC NULLS LAST;
|
||||||
LIMIT 1;
|
|
||||||
|
-- name: ClaimQueuedPodcastEpisode :one
|
||||||
|
-- Atomic claim (§0 item 4): converts queued -> downloading in ONE
|
||||||
|
-- statement. Two claimants racing on the same row: exactly one UPDATE
|
||||||
|
-- matches download_state='queued'; the loser gets no rows (ErrNoRows).
|
||||||
|
-- This is the SKIP-LOCKED guarantee without a second round-trip.
|
||||||
|
UPDATE media_items
|
||||||
|
SET download_state = 'downloading', updated_at = NOW()
|
||||||
|
WHERE id = $1
|
||||||
|
AND source = 'podcast_episode'
|
||||||
|
AND download_state = 'queued'
|
||||||
|
RETURNING media_items.*;
|
||||||
|
|
||||||
-- name: CountQueuedEpisodes :one
|
-- name: CountQueuedEpisodes :one
|
||||||
SELECT COUNT(*) FROM media_items
|
SELECT COUNT(*) FROM media_items
|
||||||
|
|||||||
@@ -25,6 +25,7 @@ import (
|
|||||||
// SettingType enumerates the value types stored in system_settings.setting_type.
|
// SettingType enumerates the value types stored in system_settings.setting_type.
|
||||||
const (
|
const (
|
||||||
SettingTypeInt = "int"
|
SettingTypeInt = "int"
|
||||||
|
SettingTypeFloat = "float"
|
||||||
SettingTypeBool = "bool"
|
SettingTypeBool = "bool"
|
||||||
SettingTypeString = "string"
|
SettingTypeString = "string"
|
||||||
SettingTypeStringList = "string_list"
|
SettingTypeStringList = "string_list"
|
||||||
@@ -83,6 +84,8 @@ var SettingDefaults = []SettingDefault{
|
|||||||
{Key: "worker_pool_size", Value: "3", Type: SettingTypeInt, Min: "1", Max: "100", RequiresRestart: true, Category: "performance", Group: "Worker Pool", Description: "Number of background worker goroutines"},
|
{Key: "worker_pool_size", Value: "3", Type: SettingTypeInt, Min: "1", Max: "100", RequiresRestart: true, Category: "performance", Group: "Worker Pool", Description: "Number of background worker goroutines"},
|
||||||
{Key: "worker_queue_cap", Value: "100", Type: SettingTypeInt, Min: "1", Max: "10000", RequiresRestart: true, Category: "performance", Group: "Worker Pool", Description: "Background worker job queue capacity"},
|
{Key: "worker_queue_cap", Value: "100", Type: SettingTypeInt, Min: "1", Max: "10000", RequiresRestart: true, Category: "performance", Group: "Worker Pool", Description: "Background worker job queue capacity"},
|
||||||
{Key: "podcast_poll_interval_seconds", Value: "3600", Type: SettingTypeInt, Min: "300", Max: "604800", Category: "podcasts", Group: "Podcasts", Description: "How often to poll podcast feeds for new episodes (seconds; complete feeds back off)"},
|
{Key: "podcast_poll_interval_seconds", Value: "3600", Type: SettingTypeInt, Min: "300", Max: "604800", Category: "podcasts", Group: "Podcasts", Description: "How often to poll podcast feeds for new episodes (seconds; complete feeds back off)"},
|
||||||
|
{Key: "podcast_download_concurrency", Value: "4", Type: SettingTypeInt, Min: "1", Max: "16", Category: "podcasts", Group: "Podcasts", Description: "Podcast episode downloads in flight server-wide (the connection is the limit; fair-share spreads them across feeds)"},
|
||||||
|
{Key: "podcast_download_speed_mbps", Value: "0", Type: SettingTypeFloat, Min: "0", Max: "10000", Category: "podcasts", Group: "Podcasts", Description: "Shared bandwidth cap for podcast downloads in megabits/second (0 = unlimited)"},
|
||||||
}
|
}
|
||||||
|
|
||||||
// defaultBy indexes SettingDefaults by key for O(1) lookup.
|
// defaultBy indexes SettingDefaults by key for O(1) lookup.
|
||||||
@@ -173,6 +176,26 @@ func (r *SettingsRegistry) getInt(key string) int {
|
|||||||
return n
|
return n
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (r *SettingsRegistry) getFloat(key string) float64 {
|
||||||
|
d := defaultBy[key]
|
||||||
|
v := r.raw(key)
|
||||||
|
f, err := strconv.ParseFloat(v, 64)
|
||||||
|
if err != nil {
|
||||||
|
f, _ = strconv.ParseFloat(d.Value, 64)
|
||||||
|
}
|
||||||
|
if d.Min != "" {
|
||||||
|
if mn, err := strconv.ParseFloat(d.Min, 64); err == nil && f < mn {
|
||||||
|
f = mn
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if d.Max != "" {
|
||||||
|
if mx, err := strconv.ParseFloat(d.Max, 64); err == nil && f > mx {
|
||||||
|
f = mx
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return f
|
||||||
|
}
|
||||||
|
|
||||||
func (r *SettingsRegistry) getBool(key string) bool {
|
func (r *SettingsRegistry) getBool(key string) bool {
|
||||||
v := r.raw(key)
|
v := r.raw(key)
|
||||||
b, err := strconv.ParseBool(v)
|
b, err := strconv.ParseBool(v)
|
||||||
@@ -189,13 +212,24 @@ func (r *SettingsRegistry) ScanPollInterval() time.Duration {
|
|||||||
return time.Duration(r.getInt("scan_poll_interval_seconds")) * time.Second
|
return time.Duration(r.getInt("scan_poll_interval_seconds")) * time.Second
|
||||||
}
|
}
|
||||||
|
|
||||||
// PodcastPollInterval is the feed-poller tick (live setting; the poller
|
// PodcastPollInterval is how often the poller checks feeds, as a
|
||||||
// re-reads it every cycle — writes go through the settings API path,
|
// duration (live setting; the poller re-reads it every cycle — writes
|
||||||
// AGENTS.md gotcha #2).
|
// go through the settings API path, AGENTS.md gotcha #2).
|
||||||
func (r *SettingsRegistry) PodcastPollInterval() time.Duration {
|
func (r *SettingsRegistry) PodcastPollInterval() time.Duration {
|
||||||
return time.Duration(r.getInt("podcast_poll_interval_seconds")) * time.Second
|
return time.Duration(r.getInt("podcast_poll_interval_seconds")) * time.Second
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// PodcastDownloadConcurrency: episode downloads in flight server-wide
|
||||||
|
// (the connection is the limit; fair-share spreads them across feeds).
|
||||||
|
func (r *SettingsRegistry) PodcastDownloadConcurrency() int {
|
||||||
|
return r.getInt("podcast_download_concurrency")
|
||||||
|
}
|
||||||
|
|
||||||
|
// PodcastDownloadSpeedMbps: shared bandwidth cap (0 = unlimited).
|
||||||
|
func (r *SettingsRegistry) PodcastDownloadSpeedMbps() float64 {
|
||||||
|
return r.getFloat("podcast_download_speed_mbps")
|
||||||
|
}
|
||||||
|
|
||||||
// AutoScanEnabled reports whether auto-scanning is on.
|
// AutoScanEnabled reports whether auto-scanning is on.
|
||||||
func (r *SettingsRegistry) AutoScanEnabled() bool { return r.getBool("auto_scan_enabled") }
|
func (r *SettingsRegistry) AutoScanEnabled() bool { return r.getBool("auto_scan_enabled") }
|
||||||
|
|
||||||
|
|||||||
@@ -178,6 +178,21 @@ func validateSettingValue(def database.SettingDefault, value string) error {
|
|||||||
if _, err := strconv.ParseBool(value); err != nil {
|
if _, err := strconv.ParseBool(value); err != nil {
|
||||||
return fmt.Errorf("value must be true or false")
|
return fmt.Errorf("value must be true or false")
|
||||||
}
|
}
|
||||||
|
case database.SettingTypeFloat:
|
||||||
|
f, err := strconv.ParseFloat(value, 64)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("value must be a number")
|
||||||
|
}
|
||||||
|
if def.Min != "" {
|
||||||
|
if mn, err := strconv.ParseFloat(def.Min, 64); err == nil && f < mn {
|
||||||
|
return fmt.Errorf("value must be >= %s", def.Min)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if def.Max != "" {
|
||||||
|
if mx, err := strconv.ParseFloat(def.Max, 64); err == nil && f > mx {
|
||||||
|
return fmt.Errorf("value must be <= %s", def.Max)
|
||||||
|
}
|
||||||
|
}
|
||||||
case database.SettingTypeString:
|
case database.SettingTypeString:
|
||||||
if value == "" {
|
if value == "" {
|
||||||
return fmt.Errorf("value must not be empty")
|
return fmt.Errorf("value must not be empty")
|
||||||
|
|||||||
@@ -1,11 +1,21 @@
|
|||||||
// Podcast download worker (plan §7 / D6 / D8 / D19): ONE sequential
|
// Podcast download worker (plan §7 / D6 / D8 / D19; pool since §0 item
|
||||||
// worker (day one) driving the episode state machine
|
// 4): a download-manager-style pool driving the episode state machine
|
||||||
//
|
//
|
||||||
// pending → queued → downloading → downloaded
|
// pending → queued → downloading → downloaded
|
||||||
// ↘ failed (+ download_error)
|
// ↘ failed (+ download_error)
|
||||||
//
|
//
|
||||||
// queued rows are the DB-backed worklist — the state is persisted so
|
// queued rows are the DB-backed worklist — the state is persisted so
|
||||||
// the episode list ALWAYS renders truth (no in-memory queue to lose).
|
// 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
|
// Idempotency: a download request for an episode whose file already
|
||||||
// exists and probes cleanly re-marks it downloaded without re-fetching
|
// exists and probes cleanly re-marks it downloaded without re-fetching
|
||||||
@@ -29,10 +39,10 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"log"
|
"log"
|
||||||
"os"
|
|
||||||
"path/filepath"
|
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/url"
|
"net/url"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
@@ -50,8 +60,18 @@ const maxAutoDownloadAttempts = 5
|
|||||||
type PodcastDownloadWorker struct {
|
type PodcastDownloadWorker struct {
|
||||||
db *database.Queries
|
db *database.Queries
|
||||||
svc *PodcastService
|
svc *PodcastService
|
||||||
|
settings *database.SettingsRegistry
|
||||||
wake chan struct{}
|
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).
|
// Live byte-progress slots for in-flight downloads (issue #1).
|
||||||
// Keyed by episode UUID string; guarded by progMu.
|
// Keyed by episode UUID string; guarded by progMu.
|
||||||
progMu sync.Mutex
|
progMu sync.Mutex
|
||||||
@@ -63,10 +83,39 @@ func NewPodcastDownloadWorker(db *database.Queries, svc *PodcastService) *Podcas
|
|||||||
db: db,
|
db: db,
|
||||||
svc: svc,
|
svc: svc,
|
||||||
wake: make(chan struct{}, 1),
|
wake: make(chan struct{}, 1),
|
||||||
|
lastServed: make(map[string]time.Time),
|
||||||
progress: make(map[string]*episodeProgress),
|
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).
|
// Wake nudges the loop (new work queued).
|
||||||
func (w *PodcastDownloadWorker) Wake() {
|
func (w *PodcastDownloadWorker) Wake() {
|
||||||
select {
|
select {
|
||||||
@@ -93,23 +142,131 @@ 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) {
|
func (w *PodcastDownloadWorker) drain(ctx context.Context) {
|
||||||
|
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 {
|
for {
|
||||||
if ctx.Err() != nil {
|
if ctx.Err() != nil {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
acted, err := w.processNext(ctx)
|
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 {
|
if err != nil {
|
||||||
log.Printf("[PODCAST-DL] worker step: %v", err)
|
return database.MediaItems{}, false
|
||||||
return
|
|
||||||
}
|
}
|
||||||
if !acted {
|
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 {
|
||||||
|
continue // claimed elsewhere / state moved on
|
||||||
|
}
|
||||||
|
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
|
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).
|
// QueueEpisode marks one episode queued (manual download / retry path).
|
||||||
// Idempotent by design: downloaded → no-op. The per-feed lock
|
// Idempotent by design: downloaded → no-op. The per-feed lock
|
||||||
// serializes against polls rewriting the same row.
|
// serializes against polls rewriting the same row.
|
||||||
@@ -228,33 +385,6 @@ func (w *PodcastDownloadWorker) ApplyAutoPolicies(ctx context.Context, feedID pg
|
|||||||
return queued, nil
|
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/
|
// runPipeline: fetch enclosure (SSRF/size guards) → temp file → MIME/
|
||||||
// size validation → ffprobe reconcile → move into place → SUCCESS-only
|
// size validation → ffprobe reconcile → move into place → SUCCESS-only
|
||||||
// row update.
|
// row update.
|
||||||
@@ -398,6 +528,42 @@ func (w *PodcastDownloadWorker) EpisodeProgress(episodeID pgtype.UUID) (received
|
|||||||
return p.Received, p.Total, true
|
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
|
// streamEnclosure performs the SSRF-guarded enclosure download straight
|
||||||
// to tmpPath while updating the progress slot. Returns bytes received
|
// to tmpPath while updating the progress slot. Returns bytes received
|
||||||
// and the authoritative total (0 when unknown). Scheme and size guards
|
// 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 {
|
if received > maxEnclosureCapBytes {
|
||||||
return received, total, fmt.Errorf("rejected: enclosure exceeds %d byte cap", 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))
|
w.addProgress(ep.ID, int64(n))
|
||||||
if _, werr := out.Write(buf[:n]); werr != nil {
|
if _, werr := out.Write(buf[:n]); werr != nil {
|
||||||
return received, total, fmt.Errorf("write: %w", werr)
|
return received, total, fmt.Errorf("write: %w", werr)
|
||||||
|
|||||||
@@ -0,0 +1,80 @@
|
|||||||
|
// Unit tests for the §0 item 4 download-pool mechanics: the shared
|
||||||
|
// bandwidth bucket (reservation math, unlimited passthrough) and the
|
||||||
|
// fair-share feed pick (never-served first, least-recently-served
|
||||||
|
// after). Pure logic — no DB.
|
||||||
|
package services
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/jackc/pgx/v5/pgtype"
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
)
|
||||||
|
|
||||||
|
func uuidOf(s string) pgtype.UUID {
|
||||||
|
var u pgtype.UUID
|
||||||
|
_ = u.Scan(s)
|
||||||
|
return u
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestBandwidthLimiter(t *testing.T) {
|
||||||
|
t.Run("zero rate is unlimited", func(t *testing.T) {
|
||||||
|
var l bandwidthLimiter
|
||||||
|
l.setRate(0)
|
||||||
|
assert.Zero(t, l.take(1<<20), "unlimited → no wait, ever")
|
||||||
|
assert.Zero(t, l.take(1<<20))
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("reservations share one bucket", func(t *testing.T) {
|
||||||
|
var l bandwidthLimiter
|
||||||
|
l.setRate(1_000_000) // 1 MB/s
|
||||||
|
start := time.Now()
|
||||||
|
|
||||||
|
w1 := l.take(1_000_000) // first caller: free flow
|
||||||
|
w2 := l.take(1_000_000) // second caller: waits one full second
|
||||||
|
w3 := l.take(1_000_000) // third caller: waits two seconds
|
||||||
|
|
||||||
|
assert.Less(t, int64(w1), int64(50*time.Millisecond), "first reservation starts immediately")
|
||||||
|
assert.GreaterOrEqual(t, int64(w2), int64(900*time.Millisecond), "second waits ~1s at 1MB/s")
|
||||||
|
assert.GreaterOrEqual(t, int64(w3), int64(1900*time.Millisecond), "third waits ~2s — SHARED bucket, not per-worker")
|
||||||
|
assert.Less(t, time.Since(start), 100*time.Millisecond, "take() only reserves — callers sleep outside the lock")
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("idle bucket recovers", func(t *testing.T) {
|
||||||
|
var l bandwidthLimiter
|
||||||
|
l.setRate(1_000_000)
|
||||||
|
_ = l.take(1_000_000)
|
||||||
|
l.next = time.Now().Add(-time.Second) // simulate a second of idle flow
|
||||||
|
assert.Less(t, int64(l.take(500_000)), int64(50*time.Millisecond),
|
||||||
|
"budget accrued while idle → next take flows immediately")
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPickFeedToServe(t *testing.T) {
|
||||||
|
a, b, c := uuidOf("00000000-0000-0000-0000-00000000000a"),
|
||||||
|
uuidOf("00000000-0000-0000-0000-00000000000b"),
|
||||||
|
uuidOf("00000000-0000-0000-0000-00000000000c")
|
||||||
|
all := map[pgtype.UUID][]pgtype.UUID{a: {}, b: {}, c: {}}
|
||||||
|
|
||||||
|
t.Run("never-served beats served", func(t *testing.T) {
|
||||||
|
served := map[string]time.Time{
|
||||||
|
a.String(): time.Now().Add(-time.Hour),
|
||||||
|
b.String(): time.Now().Add(-time.Minute),
|
||||||
|
}
|
||||||
|
assert.Equal(t, c, pickFeedToServe(all, served), "c was never served → c wins even though a starved longest")
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("least-recently-served wins", func(t *testing.T) {
|
||||||
|
served := map[string]time.Time{
|
||||||
|
a.String(): time.Now().Add(-time.Hour), // starved longest
|
||||||
|
b.String(): time.Now().Add(-time.Minute), // served recently
|
||||||
|
}
|
||||||
|
two := map[pgtype.UUID][]pgtype.UUID{a: {}, b: {}}
|
||||||
|
assert.Equal(t, a, pickFeedToServe(two, served))
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("empty queue picks nothing", func(t *testing.T) {
|
||||||
|
assert.False(t, pickFeedToServe(map[pgtype.UUID][]pgtype.UUID{}, nil).Valid)
|
||||||
|
})
|
||||||
|
}
|
||||||
@@ -244,6 +244,20 @@ templ TunableSettingRow(e SettingEntry) {
|
|||||||
}
|
}
|
||||||
class="input py-1.5 text-sm w-32"
|
class="input py-1.5 text-sm w-32"
|
||||||
/>
|
/>
|
||||||
|
} else if e.Type == "float" {
|
||||||
|
<input
|
||||||
|
type="number"
|
||||||
|
step="any"
|
||||||
|
name="value"
|
||||||
|
value={ e.Value }
|
||||||
|
if e.Min != "" {
|
||||||
|
min={ e.Min }
|
||||||
|
}
|
||||||
|
if e.Max != "" {
|
||||||
|
max={ e.Max }
|
||||||
|
}
|
||||||
|
class="input py-1.5 text-sm w-32"
|
||||||
|
/>
|
||||||
} else {
|
} else {
|
||||||
<input
|
<input
|
||||||
type="text"
|
type="text"
|
||||||
|
|||||||
@@ -813,26 +813,86 @@ func TunableSettingRow(e SettingEntry) templ.Component {
|
|||||||
if templ_7745c5c3_Err != nil {
|
if templ_7745c5c3_Err != nil {
|
||||||
return templ_7745c5c3_Err
|
return templ_7745c5c3_Err
|
||||||
}
|
}
|
||||||
} else {
|
} else if e.Type == "float" {
|
||||||
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 104, "<input type=\"text\" name=\"value\" value=\"")
|
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 104, "<input type=\"number\" step=\"any\" name=\"value\" value=\"")
|
||||||
if templ_7745c5c3_Err != nil {
|
if templ_7745c5c3_Err != nil {
|
||||||
return templ_7745c5c3_Err
|
return templ_7745c5c3_Err
|
||||||
}
|
}
|
||||||
var templ_7745c5c3_Var22 string
|
var templ_7745c5c3_Var22 string
|
||||||
templ_7745c5c3_Var22, templ_7745c5c3_Err = templ.ResolveAttributeValue(e.Value)
|
templ_7745c5c3_Var22, templ_7745c5c3_Err = templ.ResolveAttributeValue(e.Value)
|
||||||
if templ_7745c5c3_Err != nil {
|
if templ_7745c5c3_Err != nil {
|
||||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `templates/admin_settings.templ`, Line: 251, Col: 20}
|
return templ.Error{Err: templ_7745c5c3_Err, FileName: `templates/admin_settings.templ`, Line: 252, Col: 20}
|
||||||
}
|
}
|
||||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ_7745c5c3_Var22)
|
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ_7745c5c3_Var22)
|
||||||
if templ_7745c5c3_Err != nil {
|
if templ_7745c5c3_Err != nil {
|
||||||
return templ_7745c5c3_Err
|
return templ_7745c5c3_Err
|
||||||
}
|
}
|
||||||
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 105, "\" class=\"input py-1.5 text-sm w-40\"> ")
|
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 105, "\"")
|
||||||
|
if templ_7745c5c3_Err != nil {
|
||||||
|
return templ_7745c5c3_Err
|
||||||
|
}
|
||||||
|
if e.Min != "" {
|
||||||
|
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 106, " min=\"")
|
||||||
|
if templ_7745c5c3_Err != nil {
|
||||||
|
return templ_7745c5c3_Err
|
||||||
|
}
|
||||||
|
var templ_7745c5c3_Var23 string
|
||||||
|
templ_7745c5c3_Var23, templ_7745c5c3_Err = templ.ResolveAttributeValue(e.Min)
|
||||||
|
if templ_7745c5c3_Err != nil {
|
||||||
|
return templ.Error{Err: templ_7745c5c3_Err, FileName: `templates/admin_settings.templ`, Line: 254, Col: 17}
|
||||||
|
}
|
||||||
|
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ_7745c5c3_Var23)
|
||||||
|
if templ_7745c5c3_Err != nil {
|
||||||
|
return templ_7745c5c3_Err
|
||||||
|
}
|
||||||
|
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 107, "\"")
|
||||||
if templ_7745c5c3_Err != nil {
|
if templ_7745c5c3_Err != nil {
|
||||||
return templ_7745c5c3_Err
|
return templ_7745c5c3_Err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 106, "<button type=\"submit\" class=\"btn btn-secondary px-3 py-1.5 text-sm\" hx-disabled-elt=\"this\">")
|
if e.Max != "" {
|
||||||
|
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 108, " max=\"")
|
||||||
|
if templ_7745c5c3_Err != nil {
|
||||||
|
return templ_7745c5c3_Err
|
||||||
|
}
|
||||||
|
var templ_7745c5c3_Var24 string
|
||||||
|
templ_7745c5c3_Var24, templ_7745c5c3_Err = templ.ResolveAttributeValue(e.Max)
|
||||||
|
if templ_7745c5c3_Err != nil {
|
||||||
|
return templ.Error{Err: templ_7745c5c3_Err, FileName: `templates/admin_settings.templ`, Line: 257, Col: 17}
|
||||||
|
}
|
||||||
|
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ_7745c5c3_Var24)
|
||||||
|
if templ_7745c5c3_Err != nil {
|
||||||
|
return templ_7745c5c3_Err
|
||||||
|
}
|
||||||
|
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 109, "\"")
|
||||||
|
if templ_7745c5c3_Err != nil {
|
||||||
|
return templ_7745c5c3_Err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 110, " class=\"input py-1.5 text-sm w-32\"> ")
|
||||||
|
if templ_7745c5c3_Err != nil {
|
||||||
|
return templ_7745c5c3_Err
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 111, "<input type=\"text\" name=\"value\" value=\"")
|
||||||
|
if templ_7745c5c3_Err != nil {
|
||||||
|
return templ_7745c5c3_Err
|
||||||
|
}
|
||||||
|
var templ_7745c5c3_Var25 string
|
||||||
|
templ_7745c5c3_Var25, templ_7745c5c3_Err = templ.ResolveAttributeValue(e.Value)
|
||||||
|
if templ_7745c5c3_Err != nil {
|
||||||
|
return templ.Error{Err: templ_7745c5c3_Err, FileName: `templates/admin_settings.templ`, Line: 265, Col: 20}
|
||||||
|
}
|
||||||
|
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ_7745c5c3_Var25)
|
||||||
|
if templ_7745c5c3_Err != nil {
|
||||||
|
return templ_7745c5c3_Err
|
||||||
|
}
|
||||||
|
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 112, "\" class=\"input py-1.5 text-sm w-40\"> ")
|
||||||
|
if templ_7745c5c3_Err != nil {
|
||||||
|
return templ_7745c5c3_Err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 113, "<button type=\"submit\" class=\"btn btn-secondary px-3 py-1.5 text-sm\" hx-disabled-elt=\"this\">")
|
||||||
if templ_7745c5c3_Err != nil {
|
if templ_7745c5c3_Err != nil {
|
||||||
return templ_7745c5c3_Err
|
return templ_7745c5c3_Err
|
||||||
}
|
}
|
||||||
@@ -840,20 +900,20 @@ func TunableSettingRow(e SettingEntry) templ.Component {
|
|||||||
if templ_7745c5c3_Err != nil {
|
if templ_7745c5c3_Err != nil {
|
||||||
return templ_7745c5c3_Err
|
return templ_7745c5c3_Err
|
||||||
}
|
}
|
||||||
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 107, "Save</button></form><span id=\"")
|
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 114, "Save</button></form><span id=\"")
|
||||||
if templ_7745c5c3_Err != nil {
|
if templ_7745c5c3_Err != nil {
|
||||||
return templ_7745c5c3_Err
|
return templ_7745c5c3_Err
|
||||||
}
|
}
|
||||||
var templ_7745c5c3_Var23 string
|
var templ_7745c5c3_Var26 string
|
||||||
templ_7745c5c3_Var23, templ_7745c5c3_Err = templ.ResolveAttributeValue("status-" + e.Key)
|
templ_7745c5c3_Var26, templ_7745c5c3_Err = templ.ResolveAttributeValue("status-" + e.Key)
|
||||||
if templ_7745c5c3_Err != nil {
|
if templ_7745c5c3_Err != nil {
|
||||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `templates/admin_settings.templ`, Line: 260, Col: 30}
|
return templ.Error{Err: templ_7745c5c3_Err, FileName: `templates/admin_settings.templ`, Line: 274, Col: 30}
|
||||||
}
|
}
|
||||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ_7745c5c3_Var23)
|
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ_7745c5c3_Var26)
|
||||||
if templ_7745c5c3_Err != nil {
|
if templ_7745c5c3_Err != nil {
|
||||||
return templ_7745c5c3_Err
|
return templ_7745c5c3_Err
|
||||||
}
|
}
|
||||||
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 108, "\" class=\"text-xs w-24 text-right\" style=\"color: var(--text-secondary)\"></span></div>")
|
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 115, "\" class=\"text-xs w-24 text-right\" style=\"color: var(--text-secondary)\"></span></div>")
|
||||||
if templ_7745c5c3_Err != nil {
|
if templ_7745c5c3_Err != nil {
|
||||||
return templ_7745c5c3_Err
|
return templ_7745c5c3_Err
|
||||||
}
|
}
|
||||||
@@ -880,12 +940,12 @@ func AboutSection() templ.Component {
|
|||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
ctx = templ.InitializeContext(ctx)
|
ctx = templ.InitializeContext(ctx)
|
||||||
templ_7745c5c3_Var24 := templ.GetChildren(ctx)
|
templ_7745c5c3_Var27 := templ.GetChildren(ctx)
|
||||||
if templ_7745c5c3_Var24 == nil {
|
if templ_7745c5c3_Var27 == nil {
|
||||||
templ_7745c5c3_Var24 = templ.NopComponent
|
templ_7745c5c3_Var27 = templ.NopComponent
|
||||||
}
|
}
|
||||||
ctx = templ.ClearChildren(ctx)
|
ctx = templ.ClearChildren(ctx)
|
||||||
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 109, "<div class=\"mt-6 card p-6\"><div class=\"flex items-center gap-2 mb-4\">")
|
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 116, "<div class=\"mt-6 card p-6\"><div class=\"flex items-center gap-2 mb-4\">")
|
||||||
if templ_7745c5c3_Err != nil {
|
if templ_7745c5c3_Err != nil {
|
||||||
return templ_7745c5c3_Err
|
return templ_7745c5c3_Err
|
||||||
}
|
}
|
||||||
@@ -893,20 +953,20 @@ func AboutSection() templ.Component {
|
|||||||
if templ_7745c5c3_Err != nil {
|
if templ_7745c5c3_Err != nil {
|
||||||
return templ_7745c5c3_Err
|
return templ_7745c5c3_Err
|
||||||
}
|
}
|
||||||
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 110, "<h3 class=\"text-lg font-semibold\" style=\"color: var(--text-primary)\">About</h3></div><div class=\"space-y-2 text-sm\" style=\"color: var(--text-secondary);\"><p><span class=\"font-medium uppercase tracking-wide text-xs\" style=\"color: var(--text-secondary)\">Version:</span> ")
|
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 117, "<h3 class=\"text-lg font-semibold\" style=\"color: var(--text-primary)\">About</h3></div><div class=\"space-y-2 text-sm\" style=\"color: var(--text-secondary);\"><p><span class=\"font-medium uppercase tracking-wide text-xs\" style=\"color: var(--text-secondary)\">Version:</span> ")
|
||||||
if templ_7745c5c3_Err != nil {
|
if templ_7745c5c3_Err != nil {
|
||||||
return templ_7745c5c3_Err
|
return templ_7745c5c3_Err
|
||||||
}
|
}
|
||||||
var templ_7745c5c3_Var25 string
|
var templ_7745c5c3_Var28 string
|
||||||
templ_7745c5c3_Var25, templ_7745c5c3_Err = templ.JoinStringErrs(version.Version)
|
templ_7745c5c3_Var28, templ_7745c5c3_Err = templ.JoinStringErrs(version.Version)
|
||||||
if templ_7745c5c3_Err != nil {
|
if templ_7745c5c3_Err != nil {
|
||||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `templates/admin_settings.templ`, Line: 274, Col: 134}
|
return templ.Error{Err: templ_7745c5c3_Err, FileName: `templates/admin_settings.templ`, Line: 288, Col: 134}
|
||||||
}
|
}
|
||||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var25))
|
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var28))
|
||||||
if templ_7745c5c3_Err != nil {
|
if templ_7745c5c3_Err != nil {
|
||||||
return templ_7745c5c3_Err
|
return templ_7745c5c3_Err
|
||||||
}
|
}
|
||||||
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 111, "</p></div></div>")
|
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 118, "</p></div></div>")
|
||||||
if templ_7745c5c3_Err != nil {
|
if templ_7745c5c3_Err != nil {
|
||||||
return templ_7745c5c3_Err
|
return templ_7745c5c3_Err
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in new issue
Block a user