feat(usenet): auto-repair files whose bodies are missing on playback (NNTP error 430) using DFS as mount.

internal/config/config.go

@@ -200,7 +200,13 @@ type RepairConfig struct {
 	RecheckInterval       string       `json:"recheck_interval,omitempty"`
 	Arrs                  []string     `json:"arrs,omitempty"`
 	AutoRepair            bool         `json:"auto_repair,omitempty"`
-	SkipNZBRepair         bool         `json:"skip_nzb_repair,omitempty"`
+
+	// RepairOnPlaybackFailure, when true, escalates a streaming read that fails with a
+	// permanent NNTP article-not-found (430) into an immediate delete + re-search for the
+	// played file. Requires Enabled and AutoRepair to also be set. Only fires for reads
+	// through the built-in DFS mount — rclone/WebDAV playback does not trigger it.
+	RepairOnPlaybackFailure bool         `json:"repair_on_playback_failure,omitempty"`
+	SkipNZBRepair           bool         `json:"skip_nzb_repair,omitempty"`
 
 	// StopSchedule, when set, stops an in-progress repair sweep at this time/interval
 	// (same formats as Schedule: clock time, cron expression, or duration).
@@ -215,7 +221,8 @@ type RepairConfig struct {
 func (r RepairConfig) IsZero() bool {
 	return !r.Enabled && r.Source == "" && r.Schedule == "" && r.Workers == 0 &&
 		r.NNTPConnectionPercent == 0 && r.Strategy == "" && r.RecheckInterval == "" && len(r.Arrs) == 0 &&
-		!r.AutoRepair && !r.SkipNZBRepair && r.StopSchedule == ""
+		!r.AutoRepair && !r.SkipNZBRepair && r.StopSchedule == "" &&
+		!r.RepairOnPlaybackFailure
 }
 
 type Config struct {

pkg/arr/content.go

@@ -309,20 +309,58 @@ func (a *Arr) batchDeleteFiles(ctx context.Context, files []ContentFile) error {
 		payload = struct {
 			EpisodeFileIds []int `json:"episodeFileIds"`
 		}{EpisodeFileIds: ids}
-		_, err := a.RequestCtx(ctx, http.MethodDelete, "api/v3/episodefile/bulk", payload, nil)
-		if err != nil {
-			return err
+		if _, err := a.RequestCtx(ctx, http.MethodDelete, "api/v3/episodefile/bulk", payload, nil); err != nil {
+			// The bulk endpoint fails the WHOLE batch if any single id no longer
+			// exists (Sonarr's strict row-count check 500s). A stale id is
+			// common after a prior repair cycle replaced the file. Fall back to
+			// deleting each id individually, treating not-found as success — the
+			// goal (row gone) is already met for those.
+			return a.deleteFilesIndividually(ctx, "api/v3/episodefile", ids)
 		}
 	case Radarr:
 		payload = struct {
 			MovieFileIds []int `json:"movieFileIds"`
 		}{MovieFileIds: ids}
-		_, err := a.RequestCtx(ctx, http.MethodDelete, "api/v3/moviefile/bulk", payload, nil)
-		if err != nil {
-			return err
+		if _, err := a.RequestCtx(ctx, http.MethodDelete, "api/v3/moviefile/bulk", payload, nil); err != nil {
+			return a.deleteFilesIndividually(ctx, "api/v3/moviefile", ids)
 		}
 	default:
 		return fmt.Errorf("unknown arr type: %s", a.Type)
 	}
 	return nil
 }
+
+// deleteFilesIndividually deletes each file id with a single DELETE call,
+// treating a 404 (already gone) as success. This is the fallback when the bulk
+// delete fails because one or more ids are stale: per-id deletes don't trip the
+// bulk endpoint's all-or-nothing row-count check, so the ids that DO still
+// exist get cleared instead of the whole batch failing. Returns an error only
+// if a delete fails for a reason other than not-found.
+func (a *Arr) deleteFilesIndividually(ctx context.Context, basePath string, ids []int) error {
+	var firstErr error
+	for _, id := range ids {
+		if ctx != nil && ctx.Err() != nil {
+			return ctx.Err()
+		}
+		resp, err := a.RequestCtx(ctx, http.MethodDelete, fmt.Sprintf("%s/%d", basePath, id), nil, nil)
+		if err != nil {
+			// Transport/retry give-up. If the resource is already gone that's
+			// success; otherwise remember the first real error but keep going so
+			// the remaining (valid) ids still get deleted.
+			if resp != nil && resp.StatusCode == http.StatusNotFound {
+				continue
+			}
+			if firstErr == nil {
+				firstErr = err
+			}
+			continue
+		}
+		if resp != nil && resp.StatusCode != http.StatusNotFound &&
+			(resp.StatusCode < 200 || resp.StatusCode >= 300) {
+			if firstErr == nil {
+				firstErr = fmt.Errorf("delete %s/%d: status %d", basePath, id, resp.StatusCode)
+			}
+		}
+	}
+	return firstErr
+}

pkg/manager/manager.go

@@ -313,6 +313,13 @@ func (m *Manager) processJob(ctx context.Context, job *Job) {
 		if job.Entry != nil {
 			job.Entry.MarkAsError(err)
 			_ = m.queue.Update(job.Entry)
+			// A failed import (commonly a body-dead re-grab rejected by the
+			// parser) means this release is done and blocklisted. Release any
+			// playback-repair cooldown for it so the next playback can trigger
+			// the next candidate immediately instead of waiting out the timer.
+			if m.repair != nil {
+				m.repair.ClearPlaybackRepairCooldown(job.Entry.Name)
+			}
 		}
 		return
 	}

pkg/manager/repair.go

@@ -61,6 +61,21 @@ const (
 	// repairStopFinalRepairTimeout bounds the Arr delete + re-search pass run
 	// when StopSchedule fires and auto-repair is enabled.
 	repairStopFinalRepairTimeout = 5 * time.Minute
+	// playbackRepairCooldown is the minimum gap between playback-failure
+	// repairs for the SAME entry, enforced manager-side so it survives the
+	// CacheItem/Downloaders being recreated by a repair. It exists to collapse
+	// a burst of concurrent reads (and the immediate re-read after a re-grab
+	// recreates the item) into a single repair, NOT to rate-limit progress.
+	//
+	// Kept short on purpose: when a re-grabbed replacement is ALSO dead (common
+	// for purged releases with several bad copies on the indexers), the user
+	// re-pressing play should be able to kick the next attempt — which
+	// blocklists this bad copy too and pulls the next candidate — without
+	// waiting minutes. It only needs to outlast one re-grab + import-attempt
+	// cycle so concurrent reads during that window don't stampede; ~90s covers
+	// a typical import attempt while still letting a genuinely-still-broken file
+	// advance to the next release quickly.
+	playbackRepairCooldown = 2 * time.Minute
 )
 
 // Repair is the health-check / auto-repair service. One instance per Manager.
@@ -77,6 +92,16 @@ type Repair struct {
 	stopScheduled  bool
 	activeStopFunc func() // called by the stop job for the active run
 	runWG          sync.WaitGroup
+
+	// lastPlaybackRepair tracks, per entry name, when a playback-failure repair
+	// was last kicked off. This lives on the manager (not the per-file
+	// Downloaders) so the cooldown SURVIVES the file being deleted and
+	// re-imported: a repair recreates the CacheItem — and with it a fresh
+	// Downloaders whose own cooldown resets to zero — so without a
+	// manager-level guard a still-dead re-grab would re-escalate instantly and
+	// churn. Guarded by playbackRepairMu.
+	playbackRepairMu   sync.Mutex
+	lastPlaybackRepair map[string]time.Time
 }
 
 // NewRepair builds the repair service for the given manager. Call

pkg/manager/repair_sweep.go

@@ -666,6 +666,17 @@ func (r *Repair) healBrokenEntry(ctx context.Context, run *storage.RepairRun, st
 		}
 	}
 
+	// Repaired counts entries, not files, to match Broken/Probed/Healthy's
+	// granularity - a season pack with three broken episodes that all get
+	// blocklisted + re-searched is one repaired entry, not three, exactly
+	// like it's one broken entry above, not three.
+	if len(succeeded) > 0 {
+		statsMu.Lock()
+		run.Stats.Repaired++
+		r.saveRun(run)
+		statsMu.Unlock()
+	}
+
 	r.finalizeEntryRepair(name, h, succeeded)
 }
 
@@ -709,14 +720,17 @@ func (r *Repair) repairArrFiles(ctx context.Context, run *storage.RepairRun, sta
 	}
 
 	// Clear the EpisodeFile/MovieFile rows first so the upcoming re-search isn't
-	// rejected by upgrade-only quality logic.
+	// rejected by upgrade-only quality logic. A delete failure is NOT fatal to
+	// the repair: the captured FileId can be stale (a prior repair cycle for
+	// this same entry already replaced the file, so this ID no longer exists
+	// and Sonarr 500s on the bulk delete). When that happens the row we wanted
+	// gone is effectively gone anyway, and — more importantly — we must still
+	// blocklist + re-search so the Arr fetches a fresh copy. Aborting here was
+	// the cause of the playback-repair churn loop: delete fails → return →
+	// nothing re-searched → file stays broken → next playback 430 repeats.
 	if err := a.DeleteFiles(ctx, files); err != nil {
-		r.logger.Warn().Err(err).Str("arr", a.Name).Msg("Repair: DeleteFiles failed")
-		statsMu.Lock()
-		run.Stats.RepairFailed += len(files)
-		r.saveRun(run)
-		statsMu.Unlock()
-		return false
+		r.logger.Warn().Err(err).Str("arr", a.Name).
+			Msg("Repair: DeleteFiles failed (continuing to blocklist + re-search anyway)")
 	}
 
 	// Blocklist each unique grab. Errors here are non-fatal: a missing blocklist
@@ -740,8 +754,11 @@ func (r *Repair) repairArrFiles(ctx context.Context, run *storage.RepairRun, sta
 		}
 	}
 
+	// Repaired itself is incremented by the caller (healBrokenEntry), once per
+	// entry rather than once per file here - this save just keeps live
+	// progress (Probed/Broken/etc., already mutated elsewhere under statsMu)
+	// visible to a concurrent poller mid-run.
 	statsMu.Lock()
-	run.Stats.Repaired += len(files)
 	r.saveRun(run)
 	statsMu.Unlock()
 	return true
@@ -777,6 +794,19 @@ func (r *Repair) finalizeEntryRepair(name string, h *storage.EntryHealth, succee
 	if !shouldDelete {
 		h.LastRepairAt = now
 		r.saveHealth(h)
+		// A partial repair (some but not all of the entry's files were broken,
+		// or a full delete wasn't safe - e.g. a missing Arr file ID) still
+		// blocklisted + re-searched whatever succeeded above, and that heal
+		// action deserves a log line just as much as a full deletion does -
+		// otherwise it's a Repaired count with no corresponding evidence of
+		// what actually happened.
+		if len(succeeded) > 0 {
+			r.logger.Info().
+				Str("entry", name).
+				Int("broken_files", h.BrokenCount).
+				Int("total_files", h.FileCount).
+				Msg("Repair: partially repaired entry - blocklisted + re-searched broken files, entry kept")
+		}
 		return
 	}
 
@@ -1314,6 +1344,151 @@ func (r *Repair) clearBroken(ctx context.Context, run *storage.RepairRun, health
 	})
 }
 
+// RepairPlaybackFileNow repairs a file that just failed playback WITHOUT
+// re-probing it. The triggering read already hit a hard article-not-found
+// (BODY 430) — that is definitive proof the body is missing, so a confirming
+// BODY re-probe is redundant and, worse, unreliable: a re-probe sample may not
+// hit the exact dead segments the sequential read did, rolling the file up as
+// "healthy" and suppressing the repair. We trust the read: this resolves the
+// file's Arr mapping (no probe) and goes straight to delete + blocklist +
+// re-search for the single played entry.
+func (r *Repair) RepairPlaybackFileNow(ctx context.Context, entryName, fileName string) error {
+	if entryName == "" {
+		return errors.New("entry name is empty")
+	}
+
+	// Manager-level per-entry cooldown. This must be checked here (not only in
+	// the per-file Downloaders) because a repair recreates the CacheItem and
+	// its Downloaders, resetting that object's own cooldown — so a still-dead
+	// re-grab would otherwise re-escalate immediately. Claim the slot before
+	// doing any work so concurrent callers for the same entry collapse to one.
+	//
+	// The key is normalized (see normalizeCooldownKey) so it matches across the
+	// casing/punctuation variants the same episode's releases use, which lets
+	// the import-failure path (ClearPlaybackRepairCooldown) release it the
+	// instant a re-grabbed replacement is rejected as body-dead — so the next
+	// playback can immediately try the next candidate instead of waiting out
+	// the timer.
+	cooldownKey := normalizeCooldownKey(entryName)
+	r.playbackRepairMu.Lock()
+	if r.lastPlaybackRepair == nil {
+		r.lastPlaybackRepair = make(map[string]time.Time)
+	}
+	if last, ok := r.lastPlaybackRepair[cooldownKey]; ok && time.Since(last) < playbackRepairCooldown {
+		r.playbackRepairMu.Unlock()
+		r.logger.Debug().
+			Str("entry", entryName).
+			Dur("since_last", time.Since(last)).
+			Msg("playback repair: skipped, entry within cooldown")
+		return nil
+	}
+	r.lastPlaybackRepair[cooldownKey] = time.Now()
+	r.playbackRepairMu.Unlock()
+
+	item, err := r.manager.GetEntryItem(entryName)
+	if err != nil || item == nil {
+		return fmt.Errorf("entry %q not found", entryName)
+	}
+
+	// Detach from the caller's (short-lived) context: the escalation cancels
+	// its context as soon as this returns, and the Arr delete/search calls must
+	// outlive that.
+	runCtx := r.parentCtx
+	if runCtx == nil {
+		runCtx = context.Background()
+	}
+
+	// Resolve which Arr owns this entry and the per-file Arr identifiers needed
+	// to delete + re-search. This is the same lookup the probe uses, minus any
+	// verification.
+	c := &candidate{name: entryName, item: item}
+	r.attachArrContext(runCtx, c)
+	if len(c.contentMap) == 0 || c.arrName == "" {
+		return fmt.Errorf("no Arr owns entry %q; cannot re-acquire", entryName)
+	}
+
+	// Build the broken-file set. Scope to the single file that failed when we
+	// can match it; otherwise fall back to every Arr-known file in the entry
+	// (a single-file movie entry, or a filename we couldn't line up).
+	h := &storage.EntryHealth{EntryName: entryName, Status: storage.HealthBroken}
+	matched := false
+	for name, cf := range c.contentMap {
+		if fileName != "" && name != fileName && filepath.Base(name) != filepath.Base(fileName) {
+			continue
+		}
+		matched = true
+		bf := storage.BrokenFile{
+			EntryName:  entryName,
+			FileName:   name,
+			Protocol:   config.ProtocolNZB,
+			Reason:     "playback body missing (430)",
+			ArrName:    c.arrName,
+			ArrKind:    c.arrKind,
+			MediaID:    cf.Id,
+			EpisodeID:  cf.EpisodeId,
+			ArrFileID:  cf.FileId,
+			TargetPath: cf.TargetPath,
+			SourcePath: cf.Path,
+			Size:       cf.Size,
+		}
+		h.BrokenFiles = append(h.BrokenFiles, bf)
+	}
+	if !matched {
+		return fmt.Errorf("file %q not found among Arr-known files for entry %q", fileName, entryName)
+	}
+	h.BrokenCount = len(h.BrokenFiles)
+
+	r.logger.Info().
+		Str("entry", entryName).
+		Str("file", fileName).
+		Int("files_to_repair", h.BrokenCount).
+		Msg("Repair: playback failure — deleting + re-searching without re-probe")
+
+	pseudo := &storage.RepairRun{ID: "playback-" + entryName, Stats: storage.RepairRunStats{}}
+	var statsMu sync.Mutex
+	r.healBrokenEntry(runCtx, pseudo, &statsMu, entryName, h)
+	return nil
+}
+
+// normalizeCooldownKey reduces an entry/release name to lowercase alphanumerics
+// so the same episode's differently-formatted releases collapse to one key
+// (e.g. "Bosch.S05E07.The.Wisdom...REAL.REPACK...1-NTb" and
+// "bosch.s05e07.the.wisdom...real.repack...ntb" map identically). This lets the
+// playback-repair cooldown set in RepairPlaybackFileNow be matched and released
+// by the import-failure path even though the re-grabbed release name differs in
+// casing/punctuation from the originally-broken entry name.
+func normalizeCooldownKey(name string) string {
+	var b strings.Builder
+	b.Grow(len(name))
+	for _, r := range strings.ToLower(name) {
+		if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') {
+			b.WriteRune(r)
+		}
+	}
+	return b.String()
+}
+
+// ClearPlaybackRepairCooldown releases the playback-repair cooldown for an entry
+// so the next playback failure can immediately trigger the next repair attempt.
+// It is called when a re-grabbed replacement is rejected (e.g. body-dead at
+// import): that release is already blocklisted, so there is no reason to make
+// the user wait out the cooldown before the next candidate is tried. Matching is
+// by normalized key, so it works despite the re-grab's release name differing
+// from the original broken entry's name. Safe to call with names that were never
+// in cooldown (no-op).
+func (r *Repair) ClearPlaybackRepairCooldown(name string) {
+	if name == "" {
+		return
+	}
+	key := normalizeCooldownKey(name)
+	r.playbackRepairMu.Lock()
+	defer r.playbackRepairMu.Unlock()
+	if _, ok := r.lastPlaybackRepair[key]; ok {
+		delete(r.lastPlaybackRepair, key)
+		r.logger.Debug().Str("entry", name).Msg("playback repair: cooldown cleared after failed re-grab")
+	}
+}
+
 // RecheckEntry kicks off a recheck for a single entry and returns
 // immediately with an in-progress EntryHealth ack. The actual probe and
 // optional fix run in the background. With fix=true, broken Arr-known files

pkg/manager/stream.go

@@ -281,6 +281,18 @@ func (m *Manager) streamHTTP(ctx context.Context, torrent *storage.Entry, filena
 	})
 }
 
+// StreamFailureCause returns a recorded permanent stream failure for an NZB
+// file (e.g. article-not-found from a prior read/prefetch), or nil. Used to
+// surface the real cause when a stream produces no data because the file's
+// bodies are missing, so the circuit breaker and playback-repair escalation
+// classify it correctly instead of seeing a generic error.
+func (m *Manager) StreamFailureCause(entry *storage.Entry, filename string) error {
+	if m.usenet == nil || entry == nil || !entry.IsNZB() {
+		return nil
+	}
+	return m.usenet.FailedFileCause(entry.InfoHash, filename)
+}
+
 // streamUsenet handles streaming for NZB files via usenet
 func (m *Manager) streamUsenet(ctx context.Context, entry *storage.Entry, filename string, start, end int64, writer io.Writer, onReady StreamReadyFunc) error {
 	if m.usenet == nil {