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

fix(ebooks): harden enrichment queue semantics

Browse Source
This commit is contained in:
rxwatcher
2026-07-22 13:43:57 +02:00
parent ef7eedf3fc
commit eb6b968ed1
5 changed files with 777 additions and 240 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/ebooks/enrichment.go
+133 -66
View File
@@ -33,6 +33,8 @@ const (
defaultEnrichBatchSize = 50
defaultEnrichWorkers = 4
defaultEnrichmentItemTimeout = 2 * time.Minute
)
// errEnrichmentSkipped preserves the direct helper contract for an item that
@@ -58,33 +60,35 @@ func ebookEnrichWorkers() int {
}
type enrichmentItemRow struct {
ContentID string
Title string
Year int
FolderID int
Language string
Author string
ProviderIDs map[string]string
Status string
Overview string
ReleaseDate string
Genres []string
Studios []string
PosterPath string
ContentID string
Title string
Year int
FolderID int
Language string
Author string
ProviderIDs map[string]string
Status string
Overview string
Tagline string
ContentRating string
Runtime int
ReleaseDate string
Genres []string
Studios []string
PosterPath string
BackdropPath string
LogoPath string
LockedFields []int
ProtectedFields []string
}
type enrichmentQueue interface {
MaterializeCandidates(ctx context.Context) error
ClaimBatch(ctx context.Context, limit int, leaseDuration time.Duration) ([]EnrichmentJob, error)
Complete(ctx context.Context, contentID string, outcome EnrichmentOutcome, refreshAfter time.Duration) error
Fail(ctx context.Context, contentID string, errorClass EnrichmentErrorClass, message string, retryAfter time.Duration) error
Release(ctx context.Context, contentID string) error
}
type claimAwareEnrichmentQueue interface {
CompleteClaim(ctx context.Context, job EnrichmentJob, outcome EnrichmentOutcome, refreshAfter time.Duration) error
FailClaim(ctx context.Context, job EnrichmentJob, errorClass EnrichmentErrorClass, message string, retryAfter time.Duration) error
ReleaseClaim(ctx context.Context, job EnrichmentJob) error
Complete(ctx context.Context, job EnrichmentJob, outcome EnrichmentOutcome, refreshAfter time.Duration) error
Fail(ctx context.Context, job EnrichmentJob, errorClass EnrichmentErrorClass, message string, retryAfter time.Duration) error
Release(ctx context.Context, job EnrichmentJob) error
Discard(ctx context.Context, job EnrichmentJob) error
}
// Enricher drives the ebook metadata enrichment sweep.
@@ -100,6 +104,7 @@ type Enricher struct {
workLinker literaryWorkLinker
batchSize int
workers int
itemTimeout time.Duration
queue enrichmentQueue
loadClaimedItemsFn func(context.Context, []EnrichmentJob) ([]enrichmentItemRow, error)
@@ -214,13 +219,16 @@ func (e *Enricher) runQueueBatch(
for _, job := range jobs {
claimedJobs[job.ContentID] = job
}
var transitionErrs []error
loaded := make(map[string]struct{}, len(items))
for i := range items {
loaded[items[i].ContentID] = struct{}{}
}
for _, job := range jobs {
if _, ok := loaded[job.ContentID]; !ok {
_ = e.releaseJob(queue, job)
if err := e.discardJob(queue, job); err != nil && !errors.Is(err, ErrEnrichmentLeaseLost) {
transitionErrs = append(transitionErrs, fmt.Errorf("%s: %w", job.ContentID, err))
}
}
}
@@ -232,15 +240,18 @@ func (e *Enricher) runQueueBatch(
workers = len(items)
}
if workers == 0 {
return 0, nil
return 0, errors.Join(transitionErrs...)
}
itemTimeout := e.itemTimeout
if itemTimeout <= 0 || itemTimeout >= defaultEnrichmentLease {
itemTimeout = defaultEnrichmentItemTimeout
}
ch := make(chan enrichmentItemRow, workers)
var (
wg sync.WaitGroup
enriched int64
transitionMu sync.Mutex
transitionErrs []error
wg sync.WaitGroup
enriched int64
transitionMu sync.Mutex
)
recordTransitionError := func(contentID string, err error) {
if err == nil || errors.Is(err, ErrEnrichmentLeaseLost) {
@@ -262,8 +273,12 @@ func (e *Enricher) runQueueBatch(
continue
}
outcome, enrichErr := enrichFn(ctx, item)
if ctx.Err() != nil || errors.Is(enrichErr, context.Canceled) || errors.Is(enrichErr, context.DeadlineExceeded) {
itemCtx, cancelItem := context.WithTimeout(ctx, itemTimeout)
outcome, enrichErr := enrichFn(itemCtx, item)
itemCtxErr := itemCtx.Err()
cancelItem()
if ctx.Err() != nil || itemCtxErr != nil ||
errors.Is(enrichErr, context.Canceled) || errors.Is(enrichErr, context.DeadlineExceeded) {
recordTransitionError(item.ContentID, e.releaseJob(queue, job))
continue
}
@@ -330,10 +345,7 @@ func (e *Enricher) completeJob(
job EnrichmentJob,
outcome EnrichmentOutcome,
) error {
if claimQueue, ok := queue.(claimAwareEnrichmentQueue); ok {
return claimQueue.CompleteClaim(ctx, job, outcome, enrichmentRefreshHorizon(outcome))
}
return queue.Complete(ctx, job.ContentID, outcome, enrichmentRefreshHorizon(outcome))
return queue.Complete(ctx, job, outcome, enrichmentRefreshHorizon(outcome))
}
func (e *Enricher) failJob(
@@ -344,19 +356,19 @@ func (e *Enricher) failJob(
message string,
retryAfter time.Duration,
) error {
if claimQueue, ok := queue.(claimAwareEnrichmentQueue); ok {
return claimQueue.FailClaim(ctx, job, errorClass, message, retryAfter)
}
return queue.Fail(ctx, job.ContentID, errorClass, message, retryAfter)
return queue.Fail(ctx, job, errorClass, message, retryAfter)
}
func (e *Enricher) releaseJob(queue enrichmentQueue, job EnrichmentJob) error {
releaseCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if claimQueue, ok := queue.(claimAwareEnrichmentQueue); ok {
return claimQueue.ReleaseClaim(releaseCtx, job)
}
return queue.Release(releaseCtx, job.ContentID)
return queue.Release(releaseCtx, job)
}
func (e *Enricher) discardJob(queue enrichmentQueue, job EnrichmentJob) error {
discardCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
return queue.Discard(discardCtx, job)
}
func (e *Enricher) runBatch(
@@ -441,10 +453,16 @@ var loadEnrichmentItemsQuery = `
) AS author,
COALESCE(mi.status, ''),
COALESCE(mi.overview, ''),
COALESCE(mi.release_date, ''),
COALESCE(mi.tagline, ''),
COALESCE(mi.content_rating, ''),
COALESCE(mi.runtime, 0),
COALESCE(mi.release_date::text, ''),
COALESCE(mi.genres, '{}'),
COALESCE(mi.studios, '{}'),
COALESCE(mi.poster_path, '')
COALESCE(mi.poster_path, ''),
COALESCE(mi.backdrop_path, ''),
COALESCE(mi.logo_path, ''),
COALESCE(mi.locked_fields, '{}'::integer[])
FROM unnest($1::text[]) WITH ORDINALITY AS claimed(content_id, position)
JOIN media_items mi ON mi.content_id = claimed.content_id
LEFT JOIN LATERAL (
@@ -462,8 +480,10 @@ var loadEnrichmentItemsQuery = `
func (e *Enricher) loadClaimedItems(ctx context.Context, jobs []EnrichmentJob) ([]enrichmentItemRow, error) {
contentIDs := make([]string, 0, len(jobs))
claimedJobs := make(map[string]EnrichmentJob, len(jobs))
for _, job := range jobs {
contentIDs = append(contentIDs, job.ContentID)
claimedJobs[job.ContentID] = job
}
rows, err := e.pool.Query(ctx, loadEnrichmentItemsQuery, contentIDs)
if err != nil {
@@ -483,13 +503,20 @@ func (e *Enricher) loadClaimedItems(ctx context.Context, jobs []EnrichmentJob) (
&item.Author,
&item.Status,
&item.Overview,
&item.Tagline,
&item.ContentRating,
&item.Runtime,
&item.ReleaseDate,
&item.Genres,
&item.Studios,
&item.PosterPath,
&item.BackdropPath,
&item.LogoPath,
&item.LockedFields,
); err != nil {
return nil, fmt.Errorf("scanning ebook enrichment row: %w", err)
}
item.ProtectedFields = append([]string(nil), claimedJobs[item.ContentID].ProtectedFields...)
items = append(items, item)
}
if err := rows.Err(); err != nil {
@@ -604,35 +631,75 @@ func preserveEbookLocalMetadata(item enrichmentItemRow, result *metadata.Metadat
if result == nil {
return
}
if item.PosterPath != "" && !ebookPosterOwnedByRemoteProvider(item.PosterPath) {
protected := make(map[string]struct{}, len(item.ProtectedFields))
for _, field := range item.ProtectedFields {
protected[strings.ToLower(strings.TrimSpace(field))] = struct{}{}
}
isProtected := func(field string) bool {
_, ok := protected[field]
return ok
}
isLocked := func(field metadata.MetadataField) bool {
for _, locked := range item.LockedFields {
if locked == int(field) {
return true
}
}
return false
}
if isProtected("title") || isLocked(metadata.FieldName) {
result.Title = ""
result.OriginalTitle = ""
result.SortTitle = ""
}
if isProtected("year") || isLocked(metadata.FieldReleaseDates) {
result.Year = 0
}
if isProtected("overview") || isLocked(metadata.FieldOverview) {
result.Overview = ""
}
if isProtected("tagline") {
result.Tagline = ""
}
if isProtected("content_rating") || isLocked(metadata.FieldContentRating) {
result.ContentRating = ""
}
if isProtected("runtime") || isLocked(metadata.FieldRuntime) {
result.Runtime = 0
}
if isProtected("release_date") || isLocked(metadata.FieldReleaseDates) {
result.ReleaseDate = ""
}
if isProtected("genres") || isLocked(metadata.FieldGenres) {
result.Genres = nil
}
if isProtected("studios") || isLocked(metadata.FieldStudios) {
result.Studios = nil
}
if isProtected("authors") || isLocked(metadata.FieldCrew) || isLocked(metadata.FieldCast) {
result.People = nil
}
imagesLocked := isLocked(metadata.FieldImages)
if imagesLocked || isProtected("poster_path") ||
(item.PosterPath != "" && !ebookArtworkOwnedByRemoteProvider(item.PosterPath)) {
result.PosterPath = ""
result.PosterThumbhash = ""
}
if !strings.EqualFold(strings.TrimSpace(item.Status), "pending") {
return
if imagesLocked || isProtected("backdrop_path") ||
(item.BackdropPath != "" && !ebookArtworkOwnedByRemoteProvider(item.BackdropPath)) {
result.BackdropPath = ""
result.BackdropThumbhash = ""
}
if item.Year > 0 {
result.Year = 0
}
if item.Overview != "" {
result.Overview = ""
}
if item.ReleaseDate != "" {
result.ReleaseDate = ""
}
if len(item.Genres) > 0 {
result.Genres = nil
}
if len(item.Studios) > 0 {
result.Studios = nil
}
if item.Author != "" {
result.People = nil
if imagesLocked || isProtected("logo_path") ||
(item.LogoPath != "" && !ebookArtworkOwnedByRemoteProvider(item.LogoPath)) {
result.LogoPath = ""
}
}
func ebookPosterOwnedByRemoteProvider(path string) bool {
path = strings.TrimSpace(path)
func ebookArtworkOwnedByRemoteProvider(path string) bool {
path = strings.ToLower(strings.TrimSpace(path))
return isRemoteHTTPImage(path) || strings.HasPrefix(path, ebookMetadataImageProviderID+"/ebooks/")
}
internal/ebooks/enrichment_queue.go
+187 -96
View File
@@ -5,7 +5,6 @@ import (
"errors"
"fmt"
"strings"
"sync"
"time"
"github.com/jackc/pgx/v5/pgxpool"
@@ -39,29 +38,92 @@ const (
var ErrEnrichmentLeaseLost = errors.New("ebook enrichment lease lost")
type EnrichmentJob struct {
ContentID string
Attempts int
LastAttemptAt time.Time
ContentID string
Token string
Attempts int
LastAttemptAt time.Time
ProtectedFields []string
}
type EnrichmentQueue struct {
pool *pgxpool.Pool
claimsMu sync.Mutex
claims map[string]EnrichmentJob
}
func NewEnrichmentQueue(pool *pgxpool.Pool) *EnrichmentQueue {
return &EnrichmentQueue{pool: pool}
}
const ebookProtectedFieldsSQL = `
ARRAY_REMOVE(ARRAY[
CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.title, '')) <> '' THEN 'title' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND COALESCE(mi.year, 0) > 0 THEN 'year' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.overview, '')) <> '' THEN 'overview' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.tagline, '')) <> '' THEN 'tagline' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.content_rating, '')) <> '' THEN 'content_rating' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND COALESCE(mi.runtime, 0) > 0 THEN 'runtime' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.release_date::text, '')) <> '' THEN 'release_date' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND cardinality(COALESCE(mi.genres, '{}'::text[])) > 0 THEN 'genres' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND cardinality(COALESCE(mi.studios, '{}'::text[])) > 0 THEN 'studios' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND EXISTS (
SELECT 1 FROM item_people ip WHERE ip.content_id = mi.content_id AND ip.kind = 7
) THEN 'authors' END,
CASE WHEN trim(COALESCE(mi.poster_path, '')) <> ''
AND lower(trim(mi.poster_path)) NOT LIKE 'http://%'
AND lower(trim(mi.poster_path)) NOT LIKE 'https://%'
AND lower(trim(mi.poster_path)) NOT LIKE 'ebook-metadata/ebooks/%'
THEN 'poster_path' END,
CASE WHEN trim(COALESCE(mi.backdrop_path, '')) <> ''
AND lower(trim(mi.backdrop_path)) NOT LIKE 'http://%'
AND lower(trim(mi.backdrop_path)) NOT LIKE 'https://%'
AND lower(trim(mi.backdrop_path)) NOT LIKE 'ebook-metadata/ebooks/%'
THEN 'backdrop_path' END,
CASE WHEN trim(COALESCE(mi.logo_path, '')) <> ''
AND lower(trim(mi.logo_path)) NOT LIKE 'http://%'
AND lower(trim(mi.logo_path)) NOT LIKE 'https://%'
AND lower(trim(mi.logo_path)) NOT LIKE 'ebook-metadata/ebooks/%'
THEN 'logo_path' END
]::text[], NULL)
`
const mergeEbookProtectedFieldsSQL = `
ARRAY(
SELECT DISTINCT field
FROM unnest(
ebook_enrichment_state.protected_fields ||
ARRAY(
SELECT candidate
FROM unnest(EXCLUDED.protected_fields) AS candidate
WHERE candidate IN ('poster_path', 'backdrop_path', 'logo_path')
)
) AS field
ORDER BY field
)
`
var enqueueEnrichmentJobQuery = `
INSERT INTO ebook_enrichment_state (
content_id, status, priority, next_attempt_at, updated_at
content_id, status, priority, next_attempt_at, protected_fields, updated_at
)
VALUES ($1, 'pending', $2, now(), now())
SELECT mi.content_id, 'pending', $2, now(), ` + ebookProtectedFieldsSQL + `, now()
FROM media_items mi
WHERE mi.content_id = $1
AND mi.type = 'ebook'
AND ` + catalog.MangaChapterExclusionWhere("mi") + `
ON CONFLICT (content_id) DO UPDATE SET
priority = GREATEST(ebook_enrichment_state.priority, EXCLUDED.priority),
status = CASE
WHEN ebook_enrichment_state.status = 'running' THEN ebook_enrichment_state.status
ELSE 'pending'
END,
priority = CASE
WHEN ebook_enrichment_state.status = 'running' THEN ebook_enrichment_state.priority
ELSE GREATEST(ebook_enrichment_state.priority, EXCLUDED.priority)
END,
next_attempt_at = CASE
WHEN ebook_enrichment_state.status = 'running' THEN ebook_enrichment_state.next_attempt_at
ELSE now()
END,
requeue_requested = ebook_enrichment_state.requeue_requested OR ebook_enrichment_state.status = 'running',
protected_fields = ` + mergeEbookProtectedFieldsSQL + `,
updated_at = now()
`
@@ -78,14 +140,54 @@ func (q *EnrichmentQueue) Enqueue(ctx context.Context, contentID string, priorit
var materializeEnrichmentJobsQuery = `
INSERT INTO ebook_enrichment_state (
content_id, status, priority, next_attempt_at, updated_at
content_id, status, priority, next_attempt_at, completed_at, protected_fields, updated_at
)
SELECT mi.content_id, 'pending', 100, now(), now()
SELECT
mi.content_id,
'pending',
CASE WHEN mi.last_refreshed IS NULL THEN 100 ELSE 0 END,
CASE
WHEN mi.last_refreshed IS NULL THEN now()
ELSE GREATEST(mi.last_refreshed + interval '90 days', now())
END,
mi.last_refreshed,
` + ebookProtectedFieldsSQL + `,
now()
FROM media_items mi
WHERE mi.type = 'ebook'
AND ` + catalog.MangaChapterExclusionWhere("mi") + `
AND mi.last_refreshed IS NULL
ON CONFLICT (content_id) DO NOTHING
ON CONFLICT (content_id) DO UPDATE SET
status = CASE
WHEN ebook_enrichment_state.status = 'running' THEN ebook_enrichment_state.status
WHEN ebook_enrichment_state.status = 'discarded' THEN 'pending'
ELSE ebook_enrichment_state.status
END,
priority = CASE
WHEN ebook_enrichment_state.status = 'running' THEN ebook_enrichment_state.priority
WHEN ebook_enrichment_state.status = 'discarded' THEN EXCLUDED.priority
WHEN EXCLUDED.completed_at IS NULL AND ebook_enrichment_state.completed_at IS NOT NULL THEN 100
ELSE ebook_enrichment_state.priority
END,
next_attempt_at = CASE
WHEN ebook_enrichment_state.status = 'running' THEN ebook_enrichment_state.next_attempt_at
WHEN ebook_enrichment_state.status = 'discarded' THEN EXCLUDED.next_attempt_at
WHEN EXCLUDED.completed_at IS NULL AND ebook_enrichment_state.completed_at IS NOT NULL THEN now()
ELSE ebook_enrichment_state.next_attempt_at
END,
completed_at = CASE
WHEN ebook_enrichment_state.status = 'running' THEN ebook_enrichment_state.completed_at
WHEN ebook_enrichment_state.status = 'discarded' THEN EXCLUDED.completed_at
WHEN EXCLUDED.completed_at IS NULL AND ebook_enrichment_state.completed_at IS NOT NULL THEN NULL
ELSE ebook_enrichment_state.completed_at
END,
outcome = CASE
WHEN ebook_enrichment_state.status = 'discarded'
OR (EXCLUDED.completed_at IS NULL AND ebook_enrichment_state.completed_at IS NOT NULL)
THEN NULL
ELSE ebook_enrichment_state.outcome
END,
protected_fields = ` + mergeEbookProtectedFieldsSQL + `,
updated_at = now()
`
func (q *EnrichmentQueue) MaterializeCandidates(ctx context.Context) error {
@@ -102,19 +204,24 @@ var claimEnrichmentJobsQuery = `
FROM ebook_enrichment_state
WHERE next_attempt_at <= now()
AND (status = 'pending' OR (status = 'running' AND lease_until < now()))
ORDER BY priority DESC, next_attempt_at, updated_at
ORDER BY
(priority + FLOOR(EXTRACT(EPOCH FROM (now() - next_attempt_at)) / 3600)::integer) DESC,
priority DESC,
next_attempt_at,
updated_at
FOR UPDATE SKIP LOCKED
LIMIT $1
)
UPDATE ebook_enrichment_state state
SET status = 'running',
lease_until = now() + $2::interval,
claim_token = gen_random_uuid()::text,
last_attempt_at = now(),
attempts = attempts + 1,
updated_at = now()
FROM candidates
WHERE state.content_id = candidates.content_id
RETURNING state.content_id, state.attempts, state.last_attempt_at
RETURNING state.content_id, state.claim_token, state.attempts, state.last_attempt_at, state.protected_fields
`
func (q *EnrichmentQueue) ClaimBatch(ctx context.Context, limit int, leaseDuration time.Duration) ([]EnrichmentJob, error) {
@@ -137,11 +244,10 @@ func (q *EnrichmentQueue) ClaimBatch(ctx context.Context, limit int, leaseDurati
jobs := make([]EnrichmentJob, 0, limit)
for rows.Next() {
var job EnrichmentJob
if err := rows.Scan(&job.ContentID, &job.Attempts, &job.LastAttemptAt); err != nil {
if err := rows.Scan(&job.ContentID, &job.Token, &job.Attempts, &job.LastAttemptAt, &job.ProtectedFields); err != nil {
return nil, err
}
jobs = append(jobs, job)
q.rememberClaim(job)
}
if err := rows.Err(); err != nil {
return nil, err
@@ -153,36 +259,25 @@ var completeEnrichmentJobQuery = `
UPDATE ebook_enrichment_state
SET status = 'pending',
lease_until = NULL,
claim_token = NULL,
completed_at = now(),
next_attempt_at = now() + $3::interval,
next_attempt_at = CASE
WHEN requeue_requested THEN now()
ELSE now() + $3::interval
END,
outcome = $2,
attempts = 0,
priority = GREATEST(priority, 0),
priority = CASE WHEN requeue_requested THEN 100 ELSE 0 END,
requeue_requested = false,
last_error_class = NULL,
last_error = NULL,
updated_at = now()
WHERE content_id = $1
AND status = 'running'
AND last_attempt_at = $4
AND claim_token = $4
`
func (q *EnrichmentQueue) Complete(
ctx context.Context,
contentID string,
outcome EnrichmentOutcome,
refreshAfter time.Duration,
) error {
if q == nil || q.pool == nil {
return errors.New("ebook enrichment queue is not configured")
}
job, ok := q.claimedJob(contentID)
if !ok {
return ErrEnrichmentLeaseLost
}
return q.CompleteClaim(ctx, job, outcome, refreshAfter)
}
func (q *EnrichmentQueue) CompleteClaim(
ctx context.Context,
job EnrichmentJob,
outcome EnrichmentOutcome,
@@ -191,6 +286,9 @@ func (q *EnrichmentQueue) CompleteClaim(
if q == nil || q.pool == nil {
return errors.New("ebook enrichment queue is not configured")
}
if job.ContentID == "" || job.Token == "" {
return ErrEnrichmentLeaseLost
}
if refreshAfter <= 0 {
refreshAfter = enrichmentRefreshHorizon(outcome)
}
@@ -204,16 +302,14 @@ func (q *EnrichmentQueue) CompleteClaim(
job.ContentID,
string(outcome),
postgresInterval(refreshAfter),
job.LastAttemptAt,
job.Token,
)
if err != nil {
return err
}
if tag.RowsAffected() == 0 {
q.forgetClaim(job)
return ErrEnrichmentLeaseLost
}
q.forgetClaim(job)
return nil
}
@@ -221,34 +317,23 @@ var failEnrichmentJobQuery = `
UPDATE ebook_enrichment_state
SET status = 'pending',
lease_until = NULL,
next_attempt_at = now() + $4::interval,
claim_token = NULL,
next_attempt_at = CASE
WHEN requeue_requested THEN now()
ELSE now() + $4::interval
END,
priority = CASE WHEN requeue_requested THEN 100 ELSE priority END,
requeue_requested = false,
outcome = 'failed',
last_error_class = $2,
last_error = $3,
updated_at = now()
WHERE content_id = $1
AND status = 'running'
AND last_attempt_at = $5
AND claim_token = $5
`
func (q *EnrichmentQueue) Fail(
ctx context.Context,
contentID string,
errorClass EnrichmentErrorClass,
message string,
retryAfter time.Duration,
) error {
if q == nil || q.pool == nil {
return errors.New("ebook enrichment queue is not configured")
}
job, ok := q.claimedJob(contentID)
if !ok {
return ErrEnrichmentLeaseLost
}
return q.FailClaim(ctx, job, errorClass, message, retryAfter)
}
func (q *EnrichmentQueue) FailClaim(
ctx context.Context,
job EnrichmentJob,
errorClass EnrichmentErrorClass,
@@ -258,6 +343,9 @@ func (q *EnrichmentQueue) FailClaim(
if q == nil || q.pool == nil {
return errors.New("ebook enrichment queue is not configured")
}
if job.ContentID == "" || job.Token == "" {
return ErrEnrichmentLeaseLost
}
delay := enrichmentRetryDelay(errorClass, job.Attempts, retryAfter)
tag, err := q.pool.Exec(
ctx,
@@ -266,16 +354,14 @@ func (q *EnrichmentQueue) FailClaim(
string(errorClass),
message,
postgresInterval(delay),
job.LastAttemptAt,
job.Token,
)
if err != nil {
return err
}
if tag.RowsAffected() == 0 {
q.forgetClaim(job)
return ErrEnrichmentLeaseLost
}
q.forgetClaim(job)
return nil
}
@@ -283,62 +369,67 @@ var releaseEnrichmentJobQuery = `
UPDATE ebook_enrichment_state
SET status = 'pending',
lease_until = NULL,
claim_token = NULL,
attempts = GREATEST(attempts - 1, 0),
next_attempt_at = CASE WHEN requeue_requested THEN now() ELSE next_attempt_at END,
priority = CASE WHEN requeue_requested THEN 100 ELSE priority END,
requeue_requested = false,
updated_at = now()
WHERE content_id = $1
AND status = 'running'
AND last_attempt_at = $2
AND claim_token = $2
`
func (q *EnrichmentQueue) Release(ctx context.Context, contentID string) error {
func (q *EnrichmentQueue) Release(ctx context.Context, job EnrichmentJob) error {
if q == nil || q.pool == nil {
return errors.New("ebook enrichment queue is not configured")
}
job, ok := q.claimedJob(contentID)
if !ok {
if job.ContentID == "" || job.Token == "" {
return ErrEnrichmentLeaseLost
}
return q.ReleaseClaim(ctx, job)
}
func (q *EnrichmentQueue) ReleaseClaim(ctx context.Context, job EnrichmentJob) error {
if q == nil || q.pool == nil {
return errors.New("ebook enrichment queue is not configured")
}
tag, err := q.pool.Exec(ctx, releaseEnrichmentJobQuery, job.ContentID, job.LastAttemptAt)
tag, err := q.pool.Exec(ctx, releaseEnrichmentJobQuery, job.ContentID, job.Token)
if err != nil {
return err
}
q.forgetClaim(job)
if tag.RowsAffected() == 0 {
return ErrEnrichmentLeaseLost
}
return nil
}
func (q *EnrichmentQueue) rememberClaim(job EnrichmentJob) {
q.claimsMu.Lock()
defer q.claimsMu.Unlock()
if q.claims == nil {
q.claims = make(map[string]EnrichmentJob)
}
q.claims[job.ContentID] = job
}
var discardEnrichmentJobQuery = `
UPDATE ebook_enrichment_state
SET status = 'discarded',
lease_until = NULL,
claim_token = NULL,
completed_at = now(),
outcome = 'discarded',
attempts = 0,
priority = 0,
requeue_requested = false,
last_error_class = NULL,
last_error = NULL,
updated_at = now()
WHERE content_id = $1
AND status = 'running'
AND claim_token = $2
`
func (q *EnrichmentQueue) claimedJob(contentID string) (EnrichmentJob, bool) {
q.claimsMu.Lock()
defer q.claimsMu.Unlock()
job, ok := q.claims[contentID]
return job, ok
}
func (q *EnrichmentQueue) forgetClaim(job EnrichmentJob) {
q.claimsMu.Lock()
defer q.claimsMu.Unlock()
current, ok := q.claims[job.ContentID]
if ok && current.LastAttemptAt.Equal(job.LastAttemptAt) {
delete(q.claims, job.ContentID)
func (q *EnrichmentQueue) Discard(ctx context.Context, job EnrichmentJob) error {
if q == nil || q.pool == nil {
return errors.New("ebook enrichment queue is not configured")
}
if job.ContentID == "" || job.Token == "" {
return ErrEnrichmentLeaseLost
}
tag, err := q.pool.Exec(ctx, discardEnrichmentJobQuery, job.ContentID, job.Token)
if err != nil {
return err
}
if tag.RowsAffected() == 0 {
return ErrEnrichmentLeaseLost
}
return nil
}
func enrichmentRefreshHorizon(outcome EnrichmentOutcome) time.Duration {
internal/ebooks/enrichment_queue_test.go
+132 -21
View File
@@ -22,11 +22,13 @@ func TestEnrichmentQueueClaimQueryUsesAtomicLeasedClaims(t *testing.T) {
"FOR UPDATE SKIP LOCKED",
"status = 'pending' OR (status = 'running' AND lease_until < now())",
"next_attempt_at <= now()",
"ORDER BY priority DESC, next_attempt_at, updated_at",
"priority + FLOOR(EXTRACT(EPOCH FROM (now() - next_attempt_at)) / 3600)::integer",
"SET status = 'running'",
"lease_until = now() + $2::interval",
"claim_token = gen_random_uuid()::text",
"attempts = attempts + 1",
"RETURNING state.content_id, state.attempts",
"RETURNING state.content_id, state.claim_token, state.attempts",
"state.protected_fields",
} {
if !strings.Contains(query, fragment) {
t.Fatalf("claim query missing %q:\n%s", fragment, claimEnrichmentJobsQuery)
@@ -34,21 +36,76 @@ func TestEnrichmentQueueClaimQueryUsesAtomicLeasedClaims(t *testing.T) {
}
}
func TestEnrichmentQueueMaterializesUnrefreshedEbooksRegardlessOfPoster(t *testing.T) {
func TestEnrichmentQueueMaterializesEveryStandaloneEbookAndReactivatesResets(t *testing.T) {
query := strings.Join(strings.Fields(materializeEnrichmentJobsQuery), " ")
if !strings.Contains(query, "SELECT mi.content_id, 'pending', 100, now(), now()") {
t.Fatalf("newly discovered ebooks must enter ahead of legacy backfill:\n%s", materializeEnrichmentJobsQuery)
for _, fragment := range []string{
"CASE WHEN mi.last_refreshed IS NULL THEN 100 ELSE 0 END",
"GREATEST(mi.last_refreshed + interval '90 days', now())",
"mi.type = 'ebook'",
"manga_chapters",
"ON CONFLICT (content_id) DO UPDATE SET",
"ebook_enrichment_state.status = 'discarded'",
"EXCLUDED.completed_at IS NULL AND ebook_enrichment_state.completed_at IS NOT NULL",
"protected_fields",
"lower(trim(mi.status)) = 'pending'",
"ip.kind = 7",
"poster_path",
"ebook_enrichment_state.protected_fields ||",
"WHERE candidate IN ('poster_path', 'backdrop_path', 'logo_path')",
} {
if !strings.Contains(query, fragment) {
t.Fatalf("materialization query missing %q:\n%s", fragment, materializeEnrichmentJobsQuery)
}
}
if !strings.Contains(query, "mi.last_refreshed IS NULL") {
t.Fatalf("materialization must use refresh state as its eligibility gate:\n%s", materializeEnrichmentJobsQuery)
}
if strings.Contains(query, "poster_path") {
t.Fatalf("local or embedded covers must not make ebooks ineligible for metadata enrichment:\n%s", materializeEnrichmentJobsQuery)
if strings.Contains(query, "AND mi.last_refreshed IS NULL") {
t.Fatalf("last_refreshed must schedule work, not exclude refreshed ebooks:\n%s", materializeEnrichmentJobsQuery)
}
}
func TestEnrichmentQueueMigrationSeedsLegacyBacklogAtLowPriority(t *testing.T) {
func TestEnrichmentQueueCapturesEveryProviderWritablePendingField(t *testing.T) {
query := strings.Join(strings.Fields(ebookProtectedFieldsSQL), " ")
for _, field := range []string{
"title",
"year",
"overview",
"tagline",
"content_rating",
"runtime",
"release_date",
"genres",
"studios",
"authors",
"poster_path",
"backdrop_path",
"logo_path",
} {
if !strings.Contains(query, "'"+field+"'") {
t.Fatalf("protected-field capture missing %q:\n%s", field, ebookProtectedFieldsSQL)
}
}
}
func TestEnrichmentQueueEnqueueMakesPendingRowsDueAndDefersRunningRequeue(t *testing.T) {
query := strings.Join(strings.Fields(enqueueEnrichmentJobQuery), " ")
for _, fragment := range []string{
"INSERT INTO ebook_enrichment_state",
"FROM media_items mi",
"ON CONFLICT (content_id) DO UPDATE SET",
"WHEN ebook_enrichment_state.status = 'running' THEN ebook_enrichment_state.next_attempt_at",
"ELSE now()",
"requeue_requested = ebook_enrichment_state.requeue_requested OR ebook_enrichment_state.status = 'running'",
} {
if !strings.Contains(query, fragment) {
t.Fatalf("enqueue query missing %q:\n%s", fragment, enqueueEnrichmentJobQuery)
}
}
if strings.Contains(query, "lease_until =") || strings.Contains(query, "claim_token =") {
t.Fatalf("enqueue must not mutate an active lease:\n%s", enqueueEnrichmentJobQuery)
}
}
func TestEnrichmentQueueMigrationIsCrashSafeAndSeedsAllLegacyEbooks(t *testing.T) {
body, err := os.ReadFile("../../migrations/sql/20260719090000_ebook_enrichment_jobs.sql")
if err != nil {
t.Fatalf("read enrichment queue migration: %v", err)
@@ -56,20 +113,53 @@ func TestEnrichmentQueueMigrationSeedsLegacyBacklogAtLowPriority(t *testing.T) {
migration := strings.Join(strings.Fields(string(body)), " ")
for _, fragment := range []string{
"-- +goose NO TRANSACTION",
"ADD COLUMN IF NOT EXISTS claim_token text",
"ADD COLUMN IF NOT EXISTS requeue_requested boolean NOT NULL DEFAULT false",
"ADD COLUMN IF NOT EXISTS protected_fields text[] NOT NULL DEFAULT '{}'::text[]",
"INSERT INTO ebook_enrichment_state",
"SELECT mi.content_id, 'pending', -100, 0, now(), now()",
"CASE WHEN mi.last_refreshed IS NULL THEN -100 ELSE 0 END",
"GREATEST(mi.last_refreshed + interval '90 days', now())",
"mi.type = 'ebook'",
"mi.last_refreshed IS NULL",
"NOT EXISTS",
"manga_chapters",
"ON CONFLICT (content_id) DO NOTHING",
"NOT i.indisvalid",
"DROP INDEX public.ebook_enrichment_state_claim_idx",
"CREATE INDEX CONCURRENTLY IF NOT EXISTS ebook_enrichment_state_claim_idx",
"DROP INDEX CONCURRENTLY IF EXISTS ebook_enrichment_state_claim_idx",
} {
if !strings.Contains(migration, fragment) {
t.Fatalf("legacy backlog migration missing %q:\n%s", fragment, body)
}
}
if strings.Contains(migration, "poster_path") {
t.Fatalf("legacy backlog must include ebooks with local or embedded covers:\n%s", body)
if strings.Contains(migration, "AND mi.last_refreshed IS NULL") {
t.Fatalf("migration must seed refreshed ebooks too:\n%s", body)
}
}
func TestEnrichmentQueueMigrationDownPreservesCurrentFailureHistory(t *testing.T) {
body, err := os.ReadFile("../../migrations/sql/20260719090000_ebook_enrichment_jobs.sql")
if err != nil {
t.Fatalf("read enrichment queue migration: %v", err)
}
parts := strings.SplitN(strings.Join(strings.Fields(string(body)), " "), "-- +goose Down", 2)
if len(parts) != 2 {
t.Fatal("migration missing Down section")
}
down := parts[1]
for _, fragment := range []string{
"SET failures = attempts",
"WHERE outcome = 'failed'",
"DELETE FROM ebook_enrichment_state WHERE outcome IN ('success', 'no_match')",
"IF EXISTS",
} {
if !strings.Contains(down, fragment) {
t.Fatalf("down migration missing %q:\n%s", fragment, body)
}
}
if strings.Contains(down, "WHERE failures = 0") {
t.Fatalf("down migration must not erase post-migration failures based on the stale legacy counter:\n%s", body)
}
}
@@ -80,11 +170,12 @@ func TestEnrichmentQueueTransitionsKeepDurableRowsAndReleaseLeases(t *testing.T)
"status = 'pending'",
"lease_until = NULL",
"completed_at = now()",
"next_attempt_at = now() + $3::interval",
"ELSE now() + $3::interval",
"outcome = $2",
"attempts = 0",
"priority = GREATEST(priority, 0)",
"AND last_attempt_at = $4",
"WHEN requeue_requested THEN 100 ELSE 0 END",
"requeue_requested = false",
"AND claim_token = $4",
} {
if !strings.Contains(complete, fragment) {
t.Fatalf("complete query missing %q:\n%s", fragment, completeEnrichmentJobQuery)
@@ -101,7 +192,7 @@ func TestEnrichmentQueueTransitionsKeepDurableRowsAndReleaseLeases(t *testing.T)
"attempts = GREATEST(attempts - 1, 0)",
"WHERE content_id = $1",
"AND status = 'running'",
"AND last_attempt_at = $2",
"AND claim_token = $2",
} {
if !strings.Contains(release, fragment) {
t.Fatalf("release query missing %q:\n%s", fragment, releaseEnrichmentJobQuery)
@@ -109,8 +200,28 @@ func TestEnrichmentQueueTransitionsKeepDurableRowsAndReleaseLeases(t *testing.T)
}
failure := strings.Join(strings.Fields(failEnrichmentJobQuery), " ")
if !strings.Contains(failure, "AND last_attempt_at = $5") {
t.Fatalf("failure transition must reject stale leases:\n%s", failEnrichmentJobQuery)
for _, fragment := range []string{
"WHEN requeue_requested THEN now()",
"WHEN requeue_requested THEN 100 ELSE priority END",
"AND claim_token = $5",
} {
if !strings.Contains(failure, fragment) {
t.Fatalf("failure transition missing %q:\n%s", fragment, failEnrichmentJobQuery)
}
}
discard := strings.Join(strings.Fields(discardEnrichmentJobQuery), " ")
for _, fragment := range []string{
"status = 'discarded'",
"lease_until = NULL",
"claim_token = NULL",
"outcome = 'discarded'",
"WHERE content_id = $1",
"AND claim_token = $2",
} {
if !strings.Contains(discard, fragment) {
t.Fatalf("discard transition missing %q:\n%s", fragment, discardEnrichmentJobQuery)
}
}
}
internal/ebooks/enrichment_test.go
+164 -29
View File
@@ -184,15 +184,20 @@ func TestEnrichmentQueriesKeepEbookAndAudiobookMetadataSeparate(t *testing.T) {
if !strings.Contains(loadEnrichmentItemsQuery, "ip.kind = 7") {
t.Fatalf("ebook load query must load author credits only:\n%s", loadEnrichmentItemsQuery)
}
for _, field := range []string{"locked_fields", "backdrop_path", "logo_path"} {
if !strings.Contains(loadEnrichmentItemsQuery, field) {
t.Fatalf("ebook load query must load %s for protection decisions:\n%s", field, loadEnrichmentItemsQuery)
}
}
}
func TestEnricherRunTransitionsClaimedJobsByOutcome(t *testing.T) {
queue := &fakeEnrichmentQueue{
jobs: []EnrichmentJob{
{ContentID: "success", Attempts: 1},
{ContentID: "no-match", Attempts: 1},
{ContentID: "skipped", Attempts: 1},
{ContentID: "failed", Attempts: 3},
{ContentID: "success", Token: "success-token", Attempts: 1},
{ContentID: "no-match", Token: "no-match-token", Attempts: 1},
{ContentID: "skipped", Token: "skipped-token", Attempts: 1},
{ContentID: "failed", Token: "failed-token", Attempts: 3},
},
}
items := []enrichmentItemRow{
@@ -247,14 +252,19 @@ func TestEnricherRunTransitionsClaimedJobsByOutcome(t *testing.T) {
if got := queue.failed["failed"]; got != EnrichmentErrorTransient {
t.Fatalf("failed class = %q, want transient", got)
}
for contentID, job := range queue.transitionedJobs {
if job.Token != contentID+"-token" {
t.Fatalf("transition for %q used token %q", contentID, job.Token)
}
}
}
func TestEnricherRunReleasesEveryLeaseOnCancellation(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
queue := &fakeEnrichmentQueue{
jobs: []EnrichmentJob{
{ContentID: "first", Attempts: 1},
{ContentID: "second", Attempts: 1},
{ContentID: "first", Token: "first-token", Attempts: 1},
{ContentID: "second", Token: "second-token", Attempts: 1},
},
}
items := []enrichmentItemRow{{ContentID: "first"}, {ContentID: "second"}}
@@ -290,7 +300,7 @@ func TestEnricherRunReleasesEveryLeaseOnCancellation(t *testing.T) {
func TestEnricherRunReleasesLeaseWhenCompletionLosesContext(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
queue := &fakeEnrichmentQueue{
jobs: []EnrichmentJob{{ContentID: "success", Attempts: 1}},
jobs: []EnrichmentJob{{ContentID: "success", Token: "success-token", Attempts: 1}},
completeCancel: cancel,
completeErr: context.Canceled,
}
@@ -318,6 +328,72 @@ func TestEnricherRunReleasesLeaseWhenCompletionLosesContext(t *testing.T) {
}
}
func TestEnricherRunDiscardsClaimedRowsThatAreNoLongerEligible(t *testing.T) {
queue := &fakeEnrichmentQueue{
jobs: []EnrichmentJob{
{ContentID: "became-manga", Token: "manga-token", Attempts: 1},
{ContentID: "folderless", Token: "folderless-token", Attempts: 1},
},
}
e := &Enricher{
queue: queue,
batchSize: 2,
workers: 1,
loadClaimedItemsFn: func(context.Context, []EnrichmentJob) ([]enrichmentItemRow, error) {
return []enrichmentItemRow{{ContentID: "folderless", FolderID: 0}}, nil
},
enrichClaimedItemFn: func(context.Context, enrichmentItemRow) (EnrichmentOutcome, error) {
return EnrichmentOutcomeSkipped, nil
},
}
if _, err := e.Run(context.Background()); err != nil {
t.Fatalf("Run() error = %v", err)
}
if got := strings.Join(queue.discarded, ","); got != "became-manga" {
t.Fatalf("discarded = %q, want became-manga", got)
}
if got := queue.completed["folderless"]; got != EnrichmentOutcomeSkipped {
t.Fatalf("folderless outcome = %q, want skipped", got)
}
if len(queue.released) != 0 {
t.Fatalf("ineligible rows were released back into immediate churn: %v", queue.released)
}
}
func TestEnricherRunBoundsEachItemBelowTheLease(t *testing.T) {
queue := &fakeEnrichmentQueue{
jobs: []EnrichmentJob{{ContentID: "slow", Token: "slow-token", Attempts: 1}},
}
e := &Enricher{
queue: queue,
batchSize: 1,
workers: 1,
itemTimeout: 20 * time.Millisecond,
loadClaimedItemsFn: func(context.Context, []EnrichmentJob) ([]enrichmentItemRow, error) {
return []enrichmentItemRow{{ContentID: "slow"}}, nil
},
enrichClaimedItemFn: func(ctx context.Context, _ enrichmentItemRow) (EnrichmentOutcome, error) {
<-ctx.Done()
return "", ctx.Err()
},
}
started := time.Now()
if _, err := e.Run(context.Background()); err != nil {
t.Fatalf("Run() error = %v", err)
}
if elapsed := time.Since(started); elapsed > time.Second {
t.Fatalf("per-item timeout took %s", elapsed)
}
if got := strings.Join(queue.released, ","); got != "slow" {
t.Fatalf("released = %q, want slow", got)
}
if defaultEnrichmentItemTimeout >= defaultEnrichmentLease/2 {
t.Fatalf("default item timeout %s is not materially shorter than lease %s", defaultEnrichmentItemTimeout, defaultEnrichmentLease)
}
}
func TestEnrichItemSkipsItemWithoutLibraryFolder(t *testing.T) {
// Membership rows are inserted after the item upsert, so a scan-window
// race can claim an item before its folder link exists. The item must be
@@ -718,16 +794,12 @@ func TestBuildEbookMetadataRequestCarriesAccumulatedISBN(t *testing.T) {
}
}
func TestPreservePendingEbookLocalMetadata(t *testing.T) {
func TestPreserveDurableEbookLocalMetadataAcrossRefreshes(t *testing.T) {
item := enrichmentItemRow{
Status: "pending",
Year: 2021,
Overview: "Embedded description",
ReleaseDate: "2021-03-04",
Genres: []string{"Local genre"},
Studios: []string{"Local publisher"},
PosterPath: "local/ebooks/book/poster/original.webp",
Author: "Embedded Author",
Status: "matched",
ProtectedFields: []string{
"year", "overview", "release_date", "genres", "studios", "poster_path", "authors",
},
}
result := &metadata.MetadataResult{
HasMetadata: true,
@@ -759,7 +831,7 @@ func TestPreservePendingEbookLocalMetadata(t *testing.T) {
}
}
func TestPreservePendingEbookLocalMetadataAllowsRefreshReplacement(t *testing.T) {
func TestPreserveEbookMetadataAllowsProviderOwnedRefreshReplacement(t *testing.T) {
result := &metadata.MetadataResult{
HasMetadata: true,
Year: 2022,
@@ -767,9 +839,7 @@ func TestPreservePendingEbookLocalMetadataAllowsRefreshReplacement(t *testing.T)
}
preserveEbookLocalMetadata(enrichmentItemRow{
Status: "matched",
Year: 2021,
Overview: "Old remote description",
Status: "matched",
}, result)
if result.Year != 2022 || result.Overview != "Corrected remote description" {
@@ -782,15 +852,20 @@ func TestPreserveEbookLocalPosterDuringControlledRefresh(t *testing.T) {
HasMetadata: true,
PosterPath: "https://example.test/replacement.jpg",
PosterThumbhash: "remote-thumb",
BackdropPath: "https://example.test/backdrop.jpg",
LogoPath: "https://example.test/logo.png",
Overview: "Corrected remote description",
}
preserveEbookLocalMetadata(enrichmentItemRow{
Status: "matched",
PosterPath: "local/ebooks/book/poster/original.webp",
Status: "matched",
PosterPath: "local/ebooks/book/poster/original.webp",
BackdropPath: "/books/book/backdrop.jpg",
LogoPath: "embedded/book/logo.png",
}, result)
if result.PosterPath != "" || result.PosterThumbhash != "" {
if result.PosterPath != "" || result.PosterThumbhash != "" ||
result.BackdropPath != "" || result.LogoPath != "" {
t.Fatalf("controlled refresh would replace local poster: %+v", result)
}
if result.Overview != "Corrected remote description" {
@@ -798,6 +873,46 @@ func TestPreserveEbookLocalPosterDuringControlledRefresh(t *testing.T) {
}
}
func TestPreserveEbookMetadataHonorsGenericLockedFields(t *testing.T) {
result := &metadata.MetadataResult{
HasMetadata: true,
Title: "Remote title",
Overview: "Remote overview",
Year: 2024,
ReleaseDate: "2024-01-02",
Runtime: 500,
Genres: []string{"Remote genre"},
Studios: []string{"Remote publisher"},
ContentRating: "Teen",
PosterPath: "https://example.test/poster.jpg",
BackdropPath: "https://example.test/backdrop.jpg",
LogoPath: "https://example.test/logo.png",
People: []models.ItemPerson{
{Person: models.Person{Name: "Remote Author"}, Kind: models.PersonKindAuthor},
},
}
item := enrichmentItemRow{LockedFields: []int{
int(metadata.FieldName),
int(metadata.FieldOverview),
int(metadata.FieldGenres),
int(metadata.FieldStudios),
int(metadata.FieldCrew),
int(metadata.FieldRuntime),
int(metadata.FieldContentRating),
int(metadata.FieldImages),
int(metadata.FieldReleaseDates),
}}
preserveEbookLocalMetadata(item, result)
if result.Title != "" || result.Overview != "" || result.Year != 0 || result.ReleaseDate != "" ||
result.Runtime != 0 || len(result.Genres) != 0 || len(result.Studios) != 0 ||
result.ContentRating != "" || result.PosterPath != "" || result.BackdropPath != "" ||
result.LogoPath != "" || len(result.People) != 0 {
t.Fatalf("locked metadata was not protected: %+v", result)
}
}
func TestPreserveEbookLocalPosterAllowsProviderOwnedReplacement(t *testing.T) {
for _, current := range []string{
"",
@@ -872,6 +987,8 @@ type fakeEnrichmentQueue struct {
completed map[string]EnrichmentOutcome
failed map[string]EnrichmentErrorClass
released []string
discarded []string
transitionedJobs map[string]EnrichmentJob
releaseSawCanceledContext bool
completeCancel context.CancelFunc
completeErr error
@@ -893,37 +1010,55 @@ func (f *fakeEnrichmentQueue) ClaimBatch(_ context.Context, limit int, leaseDura
return append([]EnrichmentJob(nil), f.jobs...), nil
}
func (f *fakeEnrichmentQueue) Complete(_ context.Context, contentID string, outcome EnrichmentOutcome, _ time.Duration) error {
func (f *fakeEnrichmentQueue) Complete(_ context.Context, job EnrichmentJob, outcome EnrichmentOutcome, _ time.Duration) error {
f.mu.Lock()
defer f.mu.Unlock()
if f.completed == nil {
f.completed = make(map[string]EnrichmentOutcome)
}
f.completed[contentID] = outcome
f.completed[job.ContentID] = outcome
f.recordTransition(job)
if f.completeCancel != nil {
f.completeCancel()
}
return f.completeErr
}
func (f *fakeEnrichmentQueue) Fail(_ context.Context, contentID string, errorClass EnrichmentErrorClass, _ string, _ time.Duration) error {
func (f *fakeEnrichmentQueue) Fail(_ context.Context, job EnrichmentJob, errorClass EnrichmentErrorClass, _ string, _ time.Duration) error {
f.mu.Lock()
defer f.mu.Unlock()
if f.failed == nil {
f.failed = make(map[string]EnrichmentErrorClass)
}
f.failed[contentID] = errorClass
f.failed[job.ContentID] = errorClass
f.recordTransition(job)
return nil
}
func (f *fakeEnrichmentQueue) Release(ctx context.Context, contentID string) error {
func (f *fakeEnrichmentQueue) Release(ctx context.Context, job EnrichmentJob) error {
f.mu.Lock()
defer f.mu.Unlock()
f.releaseSawCanceledContext = f.releaseSawCanceledContext || ctx.Err() != nil
f.released = append(f.released, contentID)
f.released = append(f.released, job.ContentID)
f.recordTransition(job)
return nil
}
func (f *fakeEnrichmentQueue) Discard(_ context.Context, job EnrichmentJob) error {
f.mu.Lock()
defer f.mu.Unlock()
f.discarded = append(f.discarded, job.ContentID)
f.recordTransition(job)
return nil
}
func (f *fakeEnrichmentQueue) recordTransition(job EnrichmentJob) {
if f.transitionedJobs == nil {
f.transitionedJobs = make(map[string]EnrichmentJob)
}
f.transitionedJobs[job.ContentID] = job
}
func TestCleanEbookSearchTitle(t *testing.T) {
cases := []struct {
title, author, want string
migrations/sql/20260719090000_ebook_enrichment_jobs.sql
+161 -28
View File
@@ -1,38 +1,130 @@
-- +goose NO TRANSACTION
-- +goose Up
-- This ALTER is intentionally its own autocommitted statement so its table
-- lock is released before the potentially large backfill runs separately.
ALTER TABLE ebook_enrichment_state
ADD COLUMN IF NOT EXISTS status text NOT NULL DEFAULT 'pending',
ADD COLUMN IF NOT EXISTS priority integer NOT NULL DEFAULT 0,
ADD COLUMN IF NOT EXISTS attempts integer NOT NULL DEFAULT 0,
ADD COLUMN IF NOT EXISTS next_attempt_at timestamptz NOT NULL DEFAULT now(),
ADD COLUMN IF NOT EXISTS lease_until timestamptz,
ADD COLUMN IF NOT EXISTS claim_token text,
ADD COLUMN IF NOT EXISTS requeue_requested boolean NOT NULL DEFAULT false,
ADD COLUMN IF NOT EXISTS protected_fields text[] NOT NULL DEFAULT '{}'::text[],
ADD COLUMN IF NOT EXISTS last_attempt_at timestamptz,
ADD COLUMN IF NOT EXISTS completed_at timestamptz,
ADD COLUMN IF NOT EXISTS outcome text,
ADD COLUMN IF NOT EXISTS last_error_class text,
ADD COLUMN IF NOT EXISTS last_error text;
ALTER TABLE ebook_enrichment_state
DROP CONSTRAINT IF EXISTS ebook_enrichment_state_status_check,
DROP CONSTRAINT IF EXISTS ebook_enrichment_state_attempts_check;
ALTER TABLE ebook_enrichment_state
ADD COLUMN status text NOT NULL DEFAULT 'pending',
ADD COLUMN priority integer NOT NULL DEFAULT 0,
ADD COLUMN attempts integer NOT NULL DEFAULT 0,
ADD COLUMN next_attempt_at timestamptz NOT NULL DEFAULT now(),
ADD COLUMN lease_until timestamptz,
ADD COLUMN last_attempt_at timestamptz,
ADD COLUMN completed_at timestamptz,
ADD COLUMN outcome text,
ADD COLUMN last_error_class text,
ADD COLUMN last_error text,
ADD CONSTRAINT ebook_enrichment_state_status_check
CHECK (status IN ('pending', 'running')),
CHECK (status IN ('pending', 'running', 'discarded')),
ADD CONSTRAINT ebook_enrichment_state_attempts_check
CHECK (attempts >= 0);
-- Rows already tracked by the former failure counter belong to the legacy
-- backlog. Preserve their history but keep them behind incremental work.
UPDATE ebook_enrichment_state
SET attempts = failures,
priority = -100;
-- Existing failure rows are part of the legacy backlog. Refreshed rows retain
-- their refresh date as the start of the 90-day schedule; unrefreshed rows keep
-- their old failure count and enter at legacy priority -100.
UPDATE ebook_enrichment_state state
SET status = 'pending',
priority = CASE WHEN mi.last_refreshed IS NULL THEN -100 ELSE 0 END,
attempts = state.failures,
next_attempt_at = CASE
WHEN mi.last_refreshed IS NULL THEN now()
ELSE GREATEST(mi.last_refreshed + interval '90 days', now())
END,
completed_at = mi.last_refreshed,
protected_fields = ARRAY_REMOVE(ARRAY[
CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.title, '')) <> '' THEN 'title' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND COALESCE(mi.year, 0) > 0 THEN 'year' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.overview, '')) <> '' THEN 'overview' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.tagline, '')) <> '' THEN 'tagline' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.content_rating, '')) <> '' THEN 'content_rating' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND COALESCE(mi.runtime, 0) > 0 THEN 'runtime' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.release_date::text, '')) <> '' THEN 'release_date' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND cardinality(COALESCE(mi.genres, '{}'::text[])) > 0 THEN 'genres' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND cardinality(COALESCE(mi.studios, '{}'::text[])) > 0 THEN 'studios' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND EXISTS (
SELECT 1 FROM item_people ip WHERE ip.content_id = mi.content_id AND ip.kind = 7
) THEN 'authors' END,
CASE WHEN trim(COALESCE(mi.poster_path, '')) <> ''
AND lower(trim(mi.poster_path)) NOT LIKE 'http://%'
AND lower(trim(mi.poster_path)) NOT LIKE 'https://%'
AND lower(trim(mi.poster_path)) NOT LIKE 'ebook-metadata/ebooks/%'
THEN 'poster_path' END,
CASE WHEN trim(COALESCE(mi.backdrop_path, '')) <> ''
AND lower(trim(mi.backdrop_path)) NOT LIKE 'http://%'
AND lower(trim(mi.backdrop_path)) NOT LIKE 'https://%'
AND lower(trim(mi.backdrop_path)) NOT LIKE 'ebook-metadata/ebooks/%'
THEN 'backdrop_path' END,
CASE WHEN trim(COALESCE(mi.logo_path, '')) <> ''
AND lower(trim(mi.logo_path)) NOT LIKE 'http://%'
AND lower(trim(mi.logo_path)) NOT LIKE 'https://%'
AND lower(trim(mi.logo_path)) NOT LIKE 'ebook-metadata/ebooks/%'
THEN 'logo_path' END
]::text[], NULL),
updated_at = now()
FROM media_items mi
WHERE mi.content_id = state.content_id;
-- Snapshot the pre-migration library as durable low-priority backfill. Runtime
-- materialization assigns new discoveries priority 100, so this backlog cannot
-- occupy claim slots ahead of incremental or due refresh work.
-- Snapshot every pre-migration standalone ebook. Refreshed rows become normal
-- 90-day refresh work; only the unrefreshed legacy backlog starts at -100.
INSERT INTO ebook_enrichment_state (
content_id,
status,
priority,
attempts,
next_attempt_at,
completed_at,
protected_fields,
updated_at
)
SELECT mi.content_id, 'pending', -100, 0, now(), now()
SELECT
mi.content_id,
'pending',
CASE WHEN mi.last_refreshed IS NULL THEN -100 ELSE 0 END,
0,
CASE
WHEN mi.last_refreshed IS NULL THEN now()
ELSE GREATEST(mi.last_refreshed + interval '90 days', now())
END,
mi.last_refreshed,
ARRAY_REMOVE(ARRAY[
CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.title, '')) <> '' THEN 'title' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND COALESCE(mi.year, 0) > 0 THEN 'year' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.overview, '')) <> '' THEN 'overview' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.tagline, '')) <> '' THEN 'tagline' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.content_rating, '')) <> '' THEN 'content_rating' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND COALESCE(mi.runtime, 0) > 0 THEN 'runtime' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND trim(COALESCE(mi.release_date::text, '')) <> '' THEN 'release_date' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND cardinality(COALESCE(mi.genres, '{}'::text[])) > 0 THEN 'genres' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND cardinality(COALESCE(mi.studios, '{}'::text[])) > 0 THEN 'studios' END,
CASE WHEN lower(trim(mi.status)) = 'pending' AND EXISTS (
SELECT 1 FROM item_people ip WHERE ip.content_id = mi.content_id AND ip.kind = 7
) THEN 'authors' END,
CASE WHEN trim(COALESCE(mi.poster_path, '')) <> ''
AND lower(trim(mi.poster_path)) NOT LIKE 'http://%'
AND lower(trim(mi.poster_path)) NOT LIKE 'https://%'
AND lower(trim(mi.poster_path)) NOT LIKE 'ebook-metadata/ebooks/%'
THEN 'poster_path' END,
CASE WHEN trim(COALESCE(mi.backdrop_path, '')) <> ''
AND lower(trim(mi.backdrop_path)) NOT LIKE 'http://%'
AND lower(trim(mi.backdrop_path)) NOT LIKE 'https://%'
AND lower(trim(mi.backdrop_path)) NOT LIKE 'ebook-metadata/ebooks/%'
THEN 'backdrop_path' END,
CASE WHEN trim(COALESCE(mi.logo_path, '')) <> ''
AND lower(trim(mi.logo_path)) NOT LIKE 'http://%'
AND lower(trim(mi.logo_path)) NOT LIKE 'https://%'
AND lower(trim(mi.logo_path)) NOT LIKE 'ebook-metadata/ebooks/%'
THEN 'logo_path' END
]::text[], NULL),
now()
FROM media_items mi
WHERE mi.type = 'ebook'
AND NOT EXISTS (
@@ -40,20 +132,58 @@ WHERE mi.type = 'ebook'
FROM manga_chapters mc
WHERE mc.chapter_content_id = mi.content_id
)
AND mi.last_refreshed IS NULL
ON CONFLICT (content_id) DO NOTHING;
CREATE INDEX ebook_enrichment_state_claim_idx
ON ebook_enrichment_state (priority DESC, next_attempt_at, updated_at)
-- A crashed CREATE INDEX CONCURRENTLY can leave an INVALID index which blocks
-- an IF NOT EXISTS retry. Match the repository's crash-safe index pattern.
-- +goose StatementBegin
DO $$
BEGIN
IF EXISTS (
SELECT 1
FROM pg_class c
JOIN pg_namespace n ON n.oid = c.relnamespace
JOIN pg_index i ON i.indexrelid = c.oid
WHERE n.nspname = 'public'
AND c.relname = 'ebook_enrichment_state_claim_idx'
AND NOT i.indisvalid
) THEN
DROP INDEX public.ebook_enrichment_state_claim_idx;
END IF;
END;
$$;
-- +goose StatementEnd
CREATE INDEX CONCURRENTLY IF NOT EXISTS ebook_enrichment_state_claim_idx
ON public.ebook_enrichment_state (next_attempt_at, priority DESC, updated_at)
WHERE status IN ('pending', 'running');
-- +goose Down
DROP INDEX IF EXISTS ebook_enrichment_state_claim_idx;
DROP INDEX CONCURRENTLY IF EXISTS ebook_enrichment_state_claim_idx;
-- The former state table only retained positive failure rows. Remove queue rows
-- created by this migration/runtime before restoring that representation.
DELETE FROM ebook_enrichment_state
WHERE failures = 0;
-- +goose StatementBegin
DO $$
BEGIN
IF EXISTS (
SELECT 1
FROM information_schema.columns
WHERE table_schema = 'public'
AND table_name = 'ebook_enrichment_state'
AND column_name = 'attempts'
) THEN
UPDATE ebook_enrichment_state
SET failures = attempts
WHERE outcome = 'failed';
-- Success and no-match rows have no representation in the old
-- failure-only schema. Keep pending, skipped, discarded, and failed
-- rows; old code can faithfully interpret their copied failure count.
DELETE FROM ebook_enrichment_state
WHERE outcome IN ('success', 'no_match');
END IF;
END;
$$;
-- +goose StatementEnd
ALTER TABLE ebook_enrichment_state
DROP CONSTRAINT IF EXISTS ebook_enrichment_state_attempts_check,
@@ -63,6 +193,9 @@ ALTER TABLE ebook_enrichment_state
DROP COLUMN IF EXISTS outcome,
DROP COLUMN IF EXISTS completed_at,
DROP COLUMN IF EXISTS last_attempt_at,
DROP COLUMN IF EXISTS protected_fields,
DROP COLUMN IF EXISTS requeue_requested,
DROP COLUMN IF EXISTS claim_token,
DROP COLUMN IF EXISTS lease_until,
DROP COLUMN IF EXISTS next_attempt_at,
DROP COLUMN IF EXISTS attempts,