diff --git a/internal/downloads/artifact_repo.go b/internal/downloads/artifact_repo.go index 9e91ee26..674ebbe4 100644 --- a/internal/downloads/artifact_repo.go +++ b/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 { diff --git a/internal/downloads/artifact_repo_test.go b/internal/downloads/artifact_repo_test.go index 7bb049a3..3aa39f9e 100644 --- a/internal/downloads/artifact_repo_test.go +++ b/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) + } +} diff --git a/internal/downloads/artifacts.go b/internal/downloads/artifacts.go index 19d2d10c..c5a0ae4d 100644 --- a/internal/downloads/artifacts.go +++ b/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 { diff --git a/internal/downloads/model.go b/internal/downloads/model.go index c58d04e8..3eb53b52 100644 --- a/internal/downloads/model.go +++ b/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") diff --git a/internal/downloads/repo.go b/internal/downloads/repo.go index ae0e1188..692ccd83 100644 --- a/internal/downloads/repo.go +++ b/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) diff --git a/internal/downloads/service.go b/internal/downloads/service.go index 61c21c1c..469d0940 100644 --- a/internal/downloads/service.go +++ b/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