fix(downloads): fence LRU eviction against concurrent artifact links

Cleanup deleted ready artifacts after a separate HasActiveLink check, so a
cancel-then-redownload during remote byte deletion could persist a ready
row whose artifact id was then removed. ReconcileLinkedDownloads cannot
heal a missing artifact, so the download stayed advertised and 404ed.

Delete the ready row only when no active link exists, refuse Create and
ReplaceManagedEntry against a vanished id, and retry Ensure once so the
new download attaches to a fresh artifact instead.

Co-authored-by: Quick <Quick104@users.noreply.github.com>

internal/downloads/artifact_repo.go

@@ -16,6 +16,13 @@ const artifactColumns = `id, media_file_id, format, params_hash, container, code
 	attempts, max_attempts, lease_owner, lease_expires_at, next_retry_at,
 	created_at, completed_at, last_used_at`
 
+// activeArtifactLinkSQL is the EXISTS predicate that protects an artifact from
+// eviction. Completed/ready/preparing/queued/downloading rows remain servable
+// handles; cancelled/failed/revoked do not.
+const activeArtifactLinkSQL = `EXISTS(SELECT 1 FROM downloads
+		 WHERE artifact_id = $1
+		   AND status NOT IN ('cancelled','failed','revoked'))`
+
 // ArtifactRepository provides CRUD + durable-queue operations for
 // download_artifacts.
 type ArtifactRepository struct {
@@ -408,9 +415,7 @@ func (r *ArtifactRepository) TotalReadyBytes(ctx context.Context) (int64, error)
 func (r *ArtifactRepository) HasActiveLink(ctx context.Context, artifactID string) (bool, error) {
 	var exists bool
 	if err := r.pool.QueryRow(ctx,
-		`SELECT EXISTS(SELECT 1 FROM downloads
-		 WHERE artifact_id = $1
-		   AND status NOT IN ('cancelled','failed','revoked'))`,
+		`SELECT `+activeArtifactLinkSQL,
 		artifactID,
 	).Scan(&exists); err != nil {
 		return false, fmt.Errorf("checking artifact links: %w", err)
@@ -456,6 +461,39 @@ func (r *ArtifactRepository) DeleteArtifact(ctx context.Context, id string) erro
 	return nil
 }
 
+// DeleteReadyIfEvictable atomically deletes a ready artifact only when no
+// active download row references it. Check-then-delete eviction races a new
+// download linking the same id; this fence makes that outcome impossible.
+func (r *ArtifactRepository) DeleteReadyIfEvictable(ctx context.Context, id string) (bool, error) {
+	tag, err := r.pool.Exec(ctx,
+		`DELETE FROM download_artifacts
+		 WHERE id = $1 AND status = 'ready'
+		   AND NOT `+activeArtifactLinkSQL,
+		id,
+	)
+	if err != nil {
+		return false, fmt.Errorf("evicting unlinked artifact: %w", err)
+	}
+	return tag.RowsAffected() == 1, nil
+}
+
+// OutputPathInUse reports whether any remaining artifact row still claims path.
+// Eviction deletes the DB row before bytes; a concurrent EnsureQueued may have
+// already reused the deterministic local path for a replacement job.
+func (r *ArtifactRepository) OutputPathInUse(ctx context.Context, path string) (bool, error) {
+	if path == "" {
+		return false, nil
+	}
+	var exists bool
+	if err := r.pool.QueryRow(ctx,
+		`SELECT EXISTS(SELECT 1 FROM download_artifacts WHERE output_path = $1)`,
+		path,
+	).Scan(&exists); err != nil {
+		return false, fmt.Errorf("checking artifact output path: %w", err)
+	}
+	return exists, nil
+}
+
 // EnqueueRemoteOrphan durably records an abandoned attempt before its owning
 // artifact lease or locator is discarded.
 func (r *ArtifactRepository) EnqueueRemoteOrphan(ctx context.Context, downloadArtifactID string, nodeID int, nodeURL, artifactID string) error {

internal/downloads/artifact_repo_test.go

@@ -773,3 +773,112 @@ func TestHasActiveLinkCoversEphemeralRows(t *testing.T) {
 		t.Fatal("terminal-only links must not protect an artifact")
 	}
 }
+
+func markArtifactReady(t *testing.T, pool *pgxpool.Pool, id string) {
+	t.Helper()
+	if _, err := pool.Exec(context.Background(),
+		`UPDATE download_artifacts SET status = 'ready', file_size = 1024, completed_at = now() WHERE id = $1`,
+		id,
+	); err != nil {
+		t.Fatalf("mark artifact ready: %v", err)
+	}
+}
+
+func seedDownloadUser(t *testing.T, pool *pgxpool.Pool) int {
+	t.Helper()
+	var userID int
+	if err := pool.QueryRow(context.Background(),
+		`INSERT INTO users (username, role, download_allowed) VALUES ($1, 'user', true) RETURNING id`,
+		fmt.Sprintf("evictuser-%d", time.Now().UnixNano()),
+	).Scan(&userID); err != nil {
+		t.Fatalf("seed user: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), `DELETE FROM downloads WHERE user_id = $1`, userID)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id = $1`, userID)
+	})
+	return userID
+}
+
+// TestDeleteReadyIfEvictableFencesConcurrentLink pins the LRU eviction
+// invariant: a ready download that lands after the unlinked check must keep
+// the artifact row, and a download insert against a deleted id must fail
+// rather than advertise a 404-forever handle.
+func TestDeleteReadyIfEvictableFencesConcurrentLink(t *testing.T) {
+	repo, pool, fileID := newArtifactTestRepo(t)
+	ctx := context.Background()
+
+	art := newArtifact(t, fileID, fmt.Sprintf("hash-evict-%d", time.Now().UnixNano()))
+	if _, _, err := repo.EnsureQueued(ctx, art); err != nil {
+		t.Fatalf("ensure artifact: %v", err)
+	}
+	markArtifactReady(t, pool, art.ID)
+	userID := seedDownloadUser(t, pool)
+	dlRepo := NewRepository(pool)
+	now := time.Now()
+	contentID := fmt.Sprintf("evict-content-%d", now.UnixNano())
+
+	deleted, err := repo.DeleteReadyIfEvictable(ctx, art.ID)
+	if err != nil {
+		t.Fatalf("DeleteReadyIfEvictable unlinked: %v", err)
+	}
+	if !deleted {
+		t.Fatal("unlinked ready artifact must be evictable")
+	}
+	if err := dlRepo.Create(ctx, &Download{
+		ID: fmt.Sprintf("dl-evict-miss-%d", now.UnixNano()), UserID: userID, MediaFileID: fileID, ContentID: contentID,
+		Kind: KindQueued, Status: StatusReady, Format: FormatTranscode,
+		ArtifactID: art.ID, FileSize: 1024, CreatedAt: now, UpdatedAt: now,
+	}); !errors.Is(err, ErrArtifactEvicted) {
+		t.Fatalf("Create after eviction = %v, want ErrArtifactEvicted", err)
+	}
+
+	art2 := newArtifact(t, fileID, fmt.Sprintf("hash-evict-live-%d", time.Now().UnixNano()))
+	if _, _, err := repo.EnsureQueued(ctx, art2); err != nil {
+		t.Fatalf("ensure second artifact: %v", err)
+	}
+	markArtifactReady(t, pool, art2.ID)
+	if err := dlRepo.Create(ctx, &Download{
+		ID: fmt.Sprintf("dl-evict-live-%d", now.UnixNano()), UserID: userID, MediaFileID: fileID, ContentID: contentID + "-live",
+		Kind: KindQueued, Status: StatusReady, Format: FormatTranscode,
+		ArtifactID: art2.ID, FileSize: 1024, CreatedAt: now, UpdatedAt: now,
+	}); err != nil {
+		t.Fatalf("Create live download: %v", err)
+	}
+	deleted, err = repo.DeleteReadyIfEvictable(ctx, art2.ID)
+	if err != nil {
+		t.Fatalf("DeleteReadyIfEvictable linked: %v", err)
+	}
+	if deleted {
+		t.Fatal("ready download must fence LRU eviction")
+	}
+	if _, err := repo.GetByID(ctx, art2.ID); err != nil {
+		t.Fatalf("linked artifact missing after eviction attempt: %v", err)
+	}
+
+	if _, err := pool.Exec(ctx, `UPDATE downloads SET status = 'cancelled' WHERE artifact_id = $1`, art2.ID); err != nil {
+		t.Fatalf("cancel download: %v", err)
+	}
+	deleted, err = repo.DeleteReadyIfEvictable(ctx, art2.ID)
+	if err != nil {
+		t.Fatalf("DeleteReadyIfEvictable after cancel: %v", err)
+	}
+	if !deleted {
+		t.Fatal("terminal-only links must not fence eviction")
+	}
+}
+
+func TestCreateOriginalDownloadDoesNotRequireArtifact(t *testing.T) {
+	_, pool, fileID := newArtifactTestRepo(t)
+	ctx := context.Background()
+	userID := seedDownloadUser(t, pool)
+	now := time.Now()
+	if err := NewRepository(pool).Create(ctx, &Download{
+		ID: fmt.Sprintf("dl-original-%d", now.UnixNano()), UserID: userID, MediaFileID: fileID,
+		ContentID: fmt.Sprintf("original-content-%d", now.UnixNano()),
+		Kind:      KindQueued, Status: StatusQueued, Format: FormatOriginal,
+		FileSize: 1024, CreatedAt: now, UpdatedAt: now,
+	}); err != nil {
+		t.Fatalf("Create original download: %v", err)
+	}
+}

internal/downloads/artifacts.go

@@ -813,19 +813,21 @@ func (m *ArtifactManager) Cleanup(ctx context.Context) error {
 		if total <= budget {
 			break
 		}
-		active, err := m.repo.HasActiveLink(ctx, a.ID)
+		// Delete the row first, gated on the same active-link predicate
+		// HasActiveLink uses. Checking then deleting bytes then deleting the
+		// row lets a concurrent Ensure+Create land a ready download on an id
+		// that this pass then removes — the download stays advertised and
+		// never heals. The atomic delete closes that window; Create refuses
+		// to link a row that no longer exists.
+		deleted, err := m.repo.DeleteReadyIfEvictable(ctx, a.ID)
 		if err != nil {
-			slog.WarnContext(ctx, "artifact link check failed", "component", "downloads", "artifact_id", a.ID, "error", err)
+			slog.WarnContext(ctx, "artifact eviction fence failed", "component", "downloads", "artifact_id", a.ID, "error", err)
 			continue
 		}
-		if active {
+		if !deleted {
 			continue
 		}
-		if !m.deleteArtifactBytes(ctx, a) {
-			continue
-		}
-		if err := m.repo.DeleteArtifact(ctx, a.ID); err != nil {
-			slog.WarnContext(ctx, "deleting evicted artifact row failed", "component", "downloads", "artifact_id", a.ID, "error", err)
+		if !m.deleteEvictedArtifactBytes(ctx, a) {
 			continue
 		}
 		slog.InfoContext(ctx, "evicted download artifact (LRU)", "component", "downloads", "artifact_id", a.ID, "bytes", a.FileSize)
@@ -931,12 +933,24 @@ func (m *ArtifactManager) sweepStale(ctx context.Context) {
 // removeArtifact deletes an artifact's output file, its .part leftover, and
 // its row. Used by the hygiene sweep for rows nothing can serve again.
 func (m *ArtifactManager) removeArtifact(ctx context.Context, a *Artifact, reason string) {
-	if !m.deleteArtifactBytes(ctx, a) {
-		return
-	}
-	if err := m.repo.DeleteArtifact(ctx, a.ID); err != nil {
-		slog.WarnContext(ctx, "deleting swept artifact row failed", "component", "downloads", "artifact_id", a.ID, "error", err)
-		return
+	if a.Status == ArtifactReady {
+		deleted, err := m.repo.DeleteReadyIfEvictable(ctx, a.ID)
+		if err != nil {
+			slog.WarnContext(ctx, "deleting swept artifact row failed", "component", "downloads", "artifact_id", a.ID, "error", err)
+			return
+		}
+		if !deleted {
+			return
+		}
+		_ = m.deleteEvictedArtifactBytes(ctx, a)
+	} else {
+		if !m.deleteArtifactBytes(ctx, a) {
+			return
+		}
+		if err := m.repo.DeleteArtifact(ctx, a.ID); err != nil {
+			slog.WarnContext(ctx, "deleting swept artifact row failed", "component", "downloads", "artifact_id", a.ID, "error", err)
+			return
+		}
 	}
 	slog.InfoContext(ctx, "swept stale download artifact", "component", "downloads", "artifact_id", a.ID, "reason", reason, "bytes", a.FileSize)
 }
@@ -968,6 +982,33 @@ func (m *ArtifactManager) deleteArtifactBytes(ctx context.Context, a *Artifact)
 	return true
 }
 
+// deleteEvictedArtifactBytes removes bytes after the artifact row is already
+// gone. Remote locators are unique per attempt, so they are always deleted
+// (and re-queued for orphan cleanup on failure). Local output paths are
+// deterministic: skip the unlink if a replacement job already claimed the path.
+func (m *ArtifactManager) deleteEvictedArtifactBytes(ctx context.Context, a *Artifact) bool {
+	if a.OriginArtifactID != "" {
+		if m.deleteArtifactBytes(ctx, a) {
+			return true
+		}
+		m.enqueueRemoteCleanup(ctx, a.ID, PreparedArtifact{
+			OriginNodeID: a.OriginNodeID, OriginNodeURL: a.OriginNodeURL, OriginArtifactID: a.OriginArtifactID,
+		}, false)
+		return false
+	}
+	if a.OutputPath != "" {
+		inUse, err := m.repo.OutputPathInUse(ctx, a.OutputPath)
+		if err != nil {
+			slog.WarnContext(ctx, "artifact output path reuse check failed", "component", "downloads", "artifact_id", a.ID, "error", err)
+			return false
+		}
+		if inUse {
+			return true
+		}
+	}
+	return m.deleteArtifactBytes(ctx, a)
+}
+
 // backoffFor returns the retry delay for the next attempt after a failure.
 func backoffFor(attempts int) time.Duration {
 	if attempts < 1 {

internal/downloads/model.go

@@ -86,6 +86,9 @@ var (
 	// ErrArtifactOriginRemoved means a durable remote locator points at a node
 	// no longer present in the enabled transcode pool.
 	ErrArtifactOriginRemoved = errors.New("download artifact origin node was removed")
+	// ErrArtifactEvicted means a download tried to link to an artifact that LRU
+	// or hygiene eviction removed between Ensure and the row write.
+	ErrArtifactEvicted = errors.New("download artifact was evicted")
 	// ErrQualityUnavailable means the requested quality is valid and permitted in
 	// principle but cannot be fulfilled by the current server wiring.
 	ErrQualityUnavailable = errors.New("requested download quality is not available")

internal/downloads/repo.go

@@ -21,6 +21,14 @@ const insertDownloadSQL = `INSERT INTO downloads (id, user_id, profile_id, devic
 		created_at, updated_at, completed_at)
 	VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22)`
 
+// insertDownloadRequiringArtifactSQL refuses to create a row that would point
+// at an artifact LRU/hygiene already deleted. $16 is artifact_id.
+const insertDownloadRequiringArtifactSQL = `INSERT INTO downloads (id, user_id, profile_id, device_id, media_file_id, content_id,
+		episode_id, batch_id, kind, status, format, quality, effective_quality, target_bitrate_kbps, revision, artifact_id, file_size, bytes_sent, error_message,
+		created_at, updated_at, completed_at)
+	SELECT $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22
+	WHERE EXISTS (SELECT 1 FROM download_artifacts WHERE id = $16)`
+
 // Repository provides CRUD operations for the downloads table.
 type Repository struct {
 	pool *pgxpool.Pool
@@ -122,11 +130,21 @@ func (r *Repository) insertArgs(d *Download) []any {
 	}
 }
 
-// Create inserts a new download record.
+// Create inserts a new download record. Artifact-backed rows are inserted only
+// while the artifact row still exists, so LRU eviction cannot leave a ready
+// download pointing at a deleted id.
 func (r *Repository) Create(ctx context.Context, d *Download) error {
-	if _, err := r.pool.Exec(ctx, insertDownloadSQL, r.insertArgs(d)...); err != nil {
+	query := insertDownloadSQL
+	if d.ArtifactID != "" {
+		query = insertDownloadRequiringArtifactSQL
+	}
+	tag, err := r.pool.Exec(ctx, query, r.insertArgs(d)...)
+	if err != nil {
 		return fmt.Errorf("inserting download: %w", err)
 	}
+	if d.ArtifactID != "" && tag.RowsAffected() == 0 {
+		return ErrArtifactEvicted
+	}
 	return nil
 }
 
@@ -139,9 +157,17 @@ func (r *Repository) CreateBatch(ctx context.Context, downloads []*Download) err
 	defer func() { _ = tx.Rollback(ctx) }()
 
 	for _, d := range downloads {
-		if _, err := tx.Exec(ctx, insertDownloadSQL, r.insertArgs(d)...); err != nil {
+		query := insertDownloadSQL
+		if d.ArtifactID != "" {
+			query = insertDownloadRequiringArtifactSQL
+		}
+		tag, err := tx.Exec(ctx, query, r.insertArgs(d)...)
+		if err != nil {
 			return fmt.Errorf("inserting batch download: %w", err)
 		}
+		if d.ArtifactID != "" && tag.RowsAffected() == 0 {
+			return ErrArtifactEvicted
+		}
 	}
 
 	if err := tx.Commit(ctx); err != nil {
@@ -435,6 +461,7 @@ func (r *Repository) ReplaceManagedEntry(ctx context.Context, existing *Download
 			revision = revision + 1,
 			updated_at = now()
 		WHERE id = $1 AND user_id = $2 AND profile_id = $3 AND device_id = $4 AND revision = $5
+		  AND ($14::text IS NULL OR EXISTS (SELECT 1 FROM download_artifacts WHERE id = $14))
 		RETURNING ` + downloadColumns
 	row := r.pool.QueryRow(ctx, query,
 		existing.ID, existing.UserID, existing.ProfileID, existing.DeviceID, existing.Revision,
@@ -445,6 +472,18 @@ func (r *Repository) ReplaceManagedEntry(ctx context.Context, existing *Download
 	d, err := scanDownload(row)
 	if err != nil {
 		if errors.Is(err, ErrNotFound) {
+			if replacement.ArtifactID != "" {
+				var artifactExists bool
+				if aerr := r.pool.QueryRow(ctx,
+					`SELECT EXISTS(SELECT 1 FROM download_artifacts WHERE id = $1)`,
+					replacement.ArtifactID,
+				).Scan(&artifactExists); aerr != nil {
+					return nil, fmt.Errorf("replacing managed download: %w", aerr)
+				}
+				if !artifactExists {
+					return nil, ErrArtifactEvicted
+				}
+			}
 			return r.GetManagedEntry(ctx, existing.UserID, existing.ProfileID, existing.DeviceID, existing.ContentID, existing.EpisodeID)
 		}
 		return nil, fmt.Errorf("replacing managed download: %w", err)

internal/downloads/service.go

@@ -427,13 +427,7 @@ func (s *Service) createArtifactDownload(ctx context.Context, userID int, req Cr
 		}
 	}
 	if existing != nil {
-		artifact, err := s.artifacts.Ensure(ctx, file, decision.DeliveryFormat, decision.PrepareTarget)
-		if err != nil {
-			return nil, err
-		}
-		status, size := artifactRowStatus(artifact, file)
-		replacement := buildManagedDownload(userID, req.ProfileID, req.DeviceID, managedItem{file: file, contentID: file.ContentID, episodeID: file.EpisodeID}, decision, "", status, size, artifact.ID)
-		return s.reuseOrReplaceManaged(ctx, existing, replacement)
+		return s.replaceManagedArtifactDownload(ctx, existing, userID, req, file, decision)
 	}
 
 	// New row: the quota lock serializes check + insert across concurrent
@@ -446,24 +440,61 @@ func (s *Service) createArtifactDownload(ctx context.Context, userID int, req Cr
 		if err := s.limiter.Check(ctx, userID, 1); err != nil {
 			return err
 		}
-		artifact, err := s.artifacts.Ensure(ctx, file, decision.DeliveryFormat, decision.PrepareTarget)
+		row, err := s.insertArtifactDownload(ctx, userID, req, file, decision, managed)
 		if err != nil {
 			return err
 		}
+		d = row
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return d, nil
+}
+
+const artifactLinkAttempts = 2
+
+func (s *Service) replaceManagedArtifactDownload(ctx context.Context, existing *Download, userID int, req CreateRequest, file *models.MediaFile, decision QualityDecision) (*Download, error) {
+	var last error
+	for attempt := 0; attempt < artifactLinkAttempts; attempt++ {
+		artifact, err := s.artifacts.Ensure(ctx, file, decision.DeliveryFormat, decision.PrepareTarget)
+		if err != nil {
+			return nil, err
+		}
+		status, size := artifactRowStatus(artifact, file)
+		replacement := buildManagedDownload(userID, req.ProfileID, req.DeviceID, managedItem{file: file, contentID: file.ContentID, episodeID: file.EpisodeID}, decision, "", status, size, artifact.ID)
+		row, err := s.reuseOrReplaceManaged(ctx, existing, replacement)
+		if errors.Is(err, ErrArtifactEvicted) {
+			last = err
+			continue
+		}
+		return row, err
+	}
+	return nil, last
+}
+
+func (s *Service) insertArtifactDownload(ctx context.Context, userID int, req CreateRequest, file *models.MediaFile, decision QualityDecision, managed bool) (*Download, error) {
+	var last error
+	for attempt := 0; attempt < artifactLinkAttempts; attempt++ {
+		artifact, err := s.artifacts.Ensure(ctx, file, decision.DeliveryFormat, decision.PrepareTarget)
+		if err != nil {
+			return nil, err
+		}
 		status, size := artifactRowStatus(artifact, file)
 
 		if managed {
 			if err := s.repo.EnsureDevice(ctx, userID, req.ProfileID, req.DeviceID, req.DeviceName, req.DevicePlatform); err != nil {
-				return err
+				return nil, err
 			}
 		}
 
 		id, err := idgen.NextID()
 		if err != nil {
-			return fmt.Errorf("generating download ID: %w", err)
+			return nil, fmt.Errorf("generating download ID: %w", err)
 		}
 		now := time.Now()
-		d = &Download{
+		d := &Download{
 			ID:                id,
 			UserID:            userID,
 			MediaFileID:       file.ID,
@@ -486,25 +517,29 @@ func (s *Service) createArtifactDownload(ctx context.Context, userID int, req Cr
 			d.DeviceID = req.DeviceID
 		}
 		if err := s.repo.Create(ctx, d); err != nil {
+			if errors.Is(err, ErrArtifactEvicted) {
+				last = err
+				continue
+			}
 			if managed {
 				if existing, gerr := s.repo.GetManagedEntry(ctx, userID, req.ProfileID, req.DeviceID, file.ContentID, file.EpisodeID); gerr == nil {
 					replacement := buildManagedDownload(userID, req.ProfileID, req.DeviceID, managedItem{file: file, contentID: file.ContentID, episodeID: file.EpisodeID}, decision, "", status, size, artifact.ID)
 					row, rerr := s.reuseOrReplaceManaged(ctx, existing, replacement)
-					if rerr != nil {
-						return rerr
+					if errors.Is(rerr, ErrArtifactEvicted) {
+						last = rerr
+						continue
 					}
-					d = row
-					return nil
+					if rerr != nil {
+						return nil, rerr
+					}
+					return row, nil
 				}
 			}
-			return err
+			return nil, err
 		}
-		return nil
-	})
-	if err != nil {
-		return nil, err
+		return d, nil
 	}
-	return d, nil
+	return nil, last
 }
 
 // artifactRowStatus maps an ensured artifact to the download row status and