fix(requests): retire stalled targets when presence confirms the media (#470)
reconcileRequest completes a presence-confirmed request only when it has no live targets, so a quality-agnostic TMDB hit cannot orphan an in-flight download. That gate assumes the router eventually moves every target to a terminal state; when it does not, the request is pinned open forever even though the media is in the library. Retire targets stuck in queued past a 24h horizon with no status transition when presence confirms the media, and let the existing target aggregate drive request status. Targets actively downloading are never retired. Each retirement logs at WARN, since reaching this path means a router is misbehaving. Part of #469 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Quick
@@ -1557,11 +1557,11 @@ func (s *Service) reconcileRequest(ctx context.Context, req Request, fc *fulfill
|
||||
// orphan in-progress downloads. Only take the shortcut for legacy/no-live
|
||||
// -target requests; otherwise let per-target reconcile + aggregate drive
|
||||
// completion.
|
||||
hasLiveTargets, err := s.hasLiveTargets(ctx, req.ID)
|
||||
live, err := s.liveTargets(ctx, req.ID)
|
||||
if err != nil {
|
||||
return reconcileUnchanged, err
|
||||
}
|
||||
if !hasLiveTargets {
|
||||
if len(live) == 0 {
|
||||
if req.Status == StatusCompleted {
|
||||
return reconcileUnchanged, nil
|
||||
}
|
||||
@@ -1570,6 +1570,13 @@ func (s *Service) reconcileRequest(ctx context.Context, req Request, fc *fulfill
|
||||
}
|
||||
return reconcileCompleted, nil
|
||||
}
|
||||
updated, retired, err := s.retireStalledTargets(ctx, req, live)
|
||||
if err != nil {
|
||||
return reconcileUnchanged, err
|
||||
}
|
||||
if retired && updated != nil && updated.Status == StatusCompleted {
|
||||
return reconcileCompleted, nil
|
||||
}
|
||||
}
|
||||
|
||||
if req.Status == StatusApproved {
|
||||
@@ -1654,19 +1661,72 @@ func (s *Service) reconcileRequest(ctx context.Context, req Request, fc *fulfill
|
||||
return change, nil
|
||||
}
|
||||
|
||||
// hasLiveTargets reports whether the request has any non-terminal (queued or
|
||||
// downloading) fulfillment target.
|
||||
func (s *Service) hasLiveTargets(ctx context.Context, requestID string) (bool, error) {
|
||||
// liveTargets returns the request's non-terminal (queued or downloading)
|
||||
// fulfillment targets.
|
||||
func (s *Service) liveTargets(ctx context.Context, requestID string) ([]Target, error) {
|
||||
targets, err := s.store.ListTargets(ctx, requestID)
|
||||
if err != nil {
|
||||
return false, err
|
||||
return nil, err
|
||||
}
|
||||
var live []Target
|
||||
for _, t := range targets {
|
||||
if t.Status == StatusQueued || t.Status == StatusDownloading {
|
||||
return true, nil
|
||||
live = append(live, t)
|
||||
}
|
||||
}
|
||||
return false, nil
|
||||
return live, nil
|
||||
}
|
||||
|
||||
// stalledTargetHorizon is how long a queued target may go without a single
|
||||
// status transition before presence-confirmed media is allowed to retire it.
|
||||
// Reconciliation runs on a schedule measured in minutes and only writes on a
|
||||
// real status change, so a target older than this has had many chances to move
|
||||
// and has not.
|
||||
const stalledTargetHorizon = 24 * time.Hour
|
||||
|
||||
// ExternalStatusPresenceConfirmed marks a target closed out by the presence
|
||||
// backstop rather than by a router-reported completion. Exported so clients can
|
||||
// distinguish "the arr said it finished" from "we found the media ourselves".
|
||||
const ExternalStatusPresenceConfirmed = "presence_confirmed"
|
||||
|
||||
// retireStalledTargets is the backstop for a router that never reports
|
||||
// completion. The presence shortcut in reconcileRequest stays disabled for as
|
||||
// long as any target looks live, so a router that reports "queued" forever — a
|
||||
// buggy plugin, a connection pointing at an instance that no longer tracks the
|
||||
// item — pins the request open permanently even though the media is sitting in
|
||||
// the library.
|
||||
//
|
||||
// Targets that are actively downloading are never retired: those are exactly
|
||||
// the in-flight downloads the quality-agnostic presence check must not orphan.
|
||||
// Only targets stuck in queued past the horizon are closed out, and the request
|
||||
// status follows from the usual target aggregate. It returns the request as of
|
||||
// the last retirement, if any.
|
||||
func (s *Service) retireStalledTargets(ctx context.Context, req Request, live []Target) (*Request, bool, error) {
|
||||
cutoff := s.now().Add(-stalledTargetHorizon)
|
||||
var updated *Request
|
||||
retired := false
|
||||
for _, t := range live {
|
||||
if t.Status != StatusQueued || t.UpdatedAt.After(cutoff) {
|
||||
continue
|
||||
}
|
||||
next, err := s.store.UpdateTargetStatus(ctx, t.ID, StatusCompleted, "", ExternalStatusPresenceConfirmed, "", Viewer{})
|
||||
if err != nil {
|
||||
return updated, retired, err
|
||||
}
|
||||
updated, retired = next, true
|
||||
slog.WarnContext(ctx, "requests: retired stalled target on presence", "component", "requests",
|
||||
"request_id", req.ID,
|
||||
"target_id", t.ID,
|
||||
"media_type", req.MediaType,
|
||||
"tmdb_id", req.TMDBID,
|
||||
"quality", t.Quality,
|
||||
"integration_kind", t.IntegrationKind,
|
||||
"external_id", t.ExternalID,
|
||||
"external_status", t.ExternalStatus,
|
||||
"target_updated_at", t.UpdatedAt,
|
||||
)
|
||||
}
|
||||
return updated, retired, nil
|
||||
}
|
||||
|
||||
func (s *Service) requestAvailable(ctx context.Context, req Request) (bool, error) {
|
||||
|
||||
@@ -829,6 +829,126 @@ func TestReconcileRequestsCompletesByStoredTVDBID(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// seedStalledPresenceRequest builds a presence-available request carrying one
|
||||
// live target whose last status transition was `age` ago.
|
||||
func seedStalledPresenceRequest(t *testing.T, status Status, age time.Duration) (*fakeStore, *Service, time.Time) {
|
||||
t.Helper()
|
||||
now := time.Date(2026, 7, 25, 12, 0, 0, 0, time.UTC)
|
||||
store := newFakeStore()
|
||||
req := &Request{ID: "req-1", MediaType: MediaTypeMovie, TMDBID: 550, Status: StatusQueued, Outcome: OutcomeActive}
|
||||
store.candidates = []*Request{req}
|
||||
store.requests["req-1"] = req
|
||||
if _, err := store.CreateTarget(context.Background(), Target{
|
||||
RequestID: "req-1", IntegrationID: "router-1", IntegrationKind: "radarr",
|
||||
Quality: Quality1080p, Status: status, ExternalID: "123", ExternalStatus: "queued",
|
||||
UpdatedAt: now.Add(-age),
|
||||
}); err != nil {
|
||||
t.Fatalf("seed target: %v", err)
|
||||
}
|
||||
service := NewService(store, &fakeTMDBClient{}, &fakePresence{available: map[MediaType]map[int]bool{
|
||||
MediaTypeMovie: {550: true},
|
||||
}})
|
||||
service.Now = func() time.Time { return now }
|
||||
return store, service, now
|
||||
}
|
||||
|
||||
// A router that never reports completion would otherwise pin the request open
|
||||
// forever, because the presence shortcut stays disabled while a target is live.
|
||||
func TestReconcileRequestsRetiresStalledQueuedTargetOnPresence(t *testing.T) {
|
||||
store, service, _ := seedStalledPresenceRequest(t, StatusQueued, 48*time.Hour)
|
||||
|
||||
result, err := service.ReconcileRequests(context.Background(), 100)
|
||||
if err != nil {
|
||||
t.Fatalf("ReconcileRequests returned error: %v", err)
|
||||
}
|
||||
if result.Completed != 1 {
|
||||
t.Fatalf("result = %+v, want one completed", result)
|
||||
}
|
||||
targets, _ := store.ListTargets(context.Background(), "req-1")
|
||||
if len(targets) != 1 || targets[0].Status != StatusCompleted {
|
||||
t.Fatalf("targets = %+v, want the stalled target completed", targets)
|
||||
}
|
||||
if targets[0].ExternalStatus != ExternalStatusPresenceConfirmed {
|
||||
t.Fatalf("external status = %q, want %s", targets[0].ExternalStatus, ExternalStatusPresenceConfirmed)
|
||||
}
|
||||
if store.requests["req-1"].Status != StatusCompleted {
|
||||
t.Fatalf("request status = %q, want completed", store.requests["req-1"].Status)
|
||||
}
|
||||
}
|
||||
|
||||
// Inside the horizon the router still owns the target — presence must not
|
||||
// short-circuit a submission that may simply be young.
|
||||
func TestReconcileRequestsKeepsRecentQueuedTargetOnPresence(t *testing.T) {
|
||||
store, service, _ := seedStalledPresenceRequest(t, StatusQueued, time.Hour)
|
||||
|
||||
result, err := service.ReconcileRequests(context.Background(), 100)
|
||||
if err != nil {
|
||||
t.Fatalf("ReconcileRequests returned error: %v", err)
|
||||
}
|
||||
if result.Completed != 0 {
|
||||
t.Fatalf("result = %+v, want nothing completed", result)
|
||||
}
|
||||
targets, _ := store.ListTargets(context.Background(), "req-1")
|
||||
if targets[0].Status != StatusQueued {
|
||||
t.Fatalf("target status = %q, want queued", targets[0].Status)
|
||||
}
|
||||
}
|
||||
|
||||
// The presence check is quality-agnostic, so a 2160p target still downloading
|
||||
// against an already-present 1080p copy must never be retired out from under
|
||||
// the in-flight download.
|
||||
func TestReconcileRequestsNeverRetiresDownloadingTargetOnPresence(t *testing.T) {
|
||||
store, service, _ := seedStalledPresenceRequest(t, StatusDownloading, 30*24*time.Hour)
|
||||
|
||||
result, err := service.ReconcileRequests(context.Background(), 100)
|
||||
if err != nil {
|
||||
t.Fatalf("ReconcileRequests returned error: %v", err)
|
||||
}
|
||||
if result.Completed != 0 {
|
||||
t.Fatalf("result = %+v, want nothing completed", result)
|
||||
}
|
||||
targets, _ := store.ListTargets(context.Background(), "req-1")
|
||||
if targets[0].Status != StatusDownloading {
|
||||
t.Fatalf("target status = %q, want downloading left alone", targets[0].Status)
|
||||
}
|
||||
}
|
||||
|
||||
// Partial retirement: the stalled quality is closed out while a sibling target
|
||||
// that is genuinely downloading keeps the request open.
|
||||
func TestReconcileRequestsRetiresStalledQueuedWhileDownloadingStaysOpen(t *testing.T) {
|
||||
store, service, now := seedStalledPresenceRequest(t, StatusQueued, 48*time.Hour)
|
||||
if _, err := store.CreateTarget(context.Background(), Target{
|
||||
RequestID: "req-1", IntegrationID: "router-1", IntegrationKind: "radarr",
|
||||
Quality: Quality2160p, Status: StatusDownloading, ExternalID: "456", ExternalStatus: "downloading",
|
||||
UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("seed target: %v", err)
|
||||
}
|
||||
|
||||
result, err := service.ReconcileRequests(context.Background(), 100)
|
||||
if err != nil {
|
||||
t.Fatalf("ReconcileRequests returned error: %v", err)
|
||||
}
|
||||
if result.Completed != 0 {
|
||||
t.Fatalf("result = %+v, want nothing completed while a download is in flight", result)
|
||||
}
|
||||
|
||||
targets, _ := store.ListTargets(context.Background(), "req-1")
|
||||
byQuality := map[Quality]Target{}
|
||||
for _, t := range targets {
|
||||
byQuality[t.Quality] = t
|
||||
}
|
||||
if got := byQuality[Quality1080p]; got.Status != StatusCompleted || got.ExternalStatus != ExternalStatusPresenceConfirmed {
|
||||
t.Fatalf("1080p target = %+v, want completed via presence", got)
|
||||
}
|
||||
if got := byQuality[Quality2160p]; got.Status != StatusDownloading {
|
||||
t.Fatalf("2160p target = %+v, want left downloading", got)
|
||||
}
|
||||
if store.requests["req-1"].Status != StatusDownloading {
|
||||
t.Fatalf("request status = %q, want downloading (the live target keeps it open)", store.requests["req-1"].Status)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReconcileRequestsMarksDownloadingFromProvider(t *testing.T) {
|
||||
store := newFakeStore()
|
||||
store.integrations = []Integration{routerInst("router-1")}
|
||||
|
||||
Reference in New Issue
Block a user