refactor(cross-seed): replace blocking rate limiter with async job-based scheduler

internal/qbittorrent/sync_manager.go

@@ -36,7 +36,7 @@ import (
 type FilesManager interface {
 	GetCachedFiles(ctx context.Context, instanceID int, hash string) (qbt.TorrentFiles, error)
 	// GetCachedFilesBatch returns cached files for a set of torrents and the hashes that were missing/stale.
-	// Callers must pass hashes already trimmed/normalized (e.g. lowercase hex)
+	// Callers must pass hashes already trimmed/normalized (e.g. uppercase hex)
 	// because implementations treat the provided keys as-is when populating lookups and cache metadata.
 	GetCachedFilesBatch(ctx context.Context, instanceID int, hashes []string) (map[string]qbt.TorrentFiles, []string, error)
 	CacheFiles(ctx context.Context, instanceID int, hash string, files qbt.TorrentFiles) error
@@ -1272,13 +1272,13 @@ func (sm *SyncManager) GetTorrentFilesBatch(ctx context.Context, instanceID int,
 				return nil
 			}
 
-			// Clone the slice so later fetches cannot mutate the stored entry if the
-			// client reuses backing arrays across calls.
-			copied := make(qbt.TorrentFiles, len(*files))
-			copy(copied, *files)
+			// Clone the API response once. This clone is shared between the caller's
+			// result map and the cache. Callers must treat returned slices as read-only.
+			callerCopy := make(qbt.TorrentFiles, len(*files))
+			copy(callerCopy, *files)
 
 			mu.Lock()
-			filesByHash[ch] = copied
+			filesByHash[ch] = callerCopy
 			mu.Unlock()
 			return nil
 		})
@@ -1288,15 +1288,15 @@ func (sm *SyncManager) GetTorrentFilesBatch(ctx context.Context, instanceID int,
 		return filesByHash, err
 	}
 
-	// Cache all newly fetched files in batch
+	// Cache all newly fetched files in batch.
+	// Fresh fetches share the cloned slice between caller and cache (one clone total).
+	// Cache hits (handled earlier) return isolated clones.
+	// IMPORTANT: Callers must treat qbt.TorrentFiles as read-only to avoid cache corruption.
 	if fm := sm.getFilesManager(); fm != nil && len(hashesToFetch) > 0 {
 		fetchedFiles := make(map[string]qbt.TorrentFiles)
 		for _, canonicalHash := range hashesToFetch {
 			if files, ok := filesByHash[canonicalHash]; ok {
-				// Cache a clone to keep cache entries isolated.
-				cloned := make(qbt.TorrentFiles, len(files))
-				copy(cloned, files)
-				fetchedFiles[canonicalHash] = cloned
+				fetchedFiles[canonicalHash] = files
 			}
 		}
 		if len(fetchedFiles) > 0 {

internal/services/crossseed/service.go

@@ -31,6 +31,7 @@ import (
 
 	"github.com/autobrr/autobrr/pkg/ttlcache"
 	qbt "github.com/autobrr/go-qbittorrent"
+	"github.com/cespare/xxhash/v2"
 	"github.com/moistari/rls"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
@@ -69,6 +70,12 @@ type qbittorrentSync interface {
 	SetTags(ctx context.Context, instanceID int, hashes []string, tags string) error
 }
 
+// dedupCacheEntry stores cached deduplication results to avoid recomputation.
+type dedupCacheEntry struct {
+	deduplicated    []qbt.Torrent
+	duplicateHashes map[string][]string
+}
+
 // ServiceMetrics contains Prometheus metrics for the cross-seed service
 type ServiceMetrics struct {
 	FindCandidatesDuration    prometheus.Histogram
@@ -247,6 +254,11 @@ type Service struct {
 	// Cached torrent file metadata for repeated analyze/search calls.
 	torrentFilesCache *ttlcache.Cache[string, qbt.TorrentFiles]
 
+	// Cached deduplication results to avoid recomputing on every search run.
+	// Key: "dedup:{instanceID}:{torrentCount}:{hashSignature}" where hashSignature
+	// is derived from a sample of torrent hashes to detect list changes.
+	dedupCache *ttlcache.Cache[string, *dedupCacheEntry]
+
 	searchMu     sync.RWMutex
 	searchCancel context.CancelFunc
 	searchState  *searchRunState
@@ -280,6 +292,8 @@ func NewService(
 		SetDefaultTTL(indexerDomainCacheTTL))
 	contentFilesCache := ttlcache.New(ttlcache.Options[string, qbt.TorrentFiles]{}.
 		SetDefaultTTL(5 * time.Minute))
+	dedupCache := ttlcache.New(ttlcache.Options[string, *dedupCacheEntry]{}.
+		SetDefaultTTL(5 * time.Minute))
 
 	return &Service{
 		instanceStore:        instanceStore,
@@ -296,6 +310,7 @@ func NewService(
 		automationWake:       make(chan struct{}, 1),
 		domainMappings:       initializeDomainMappings(),
 		torrentFilesCache:    contentFilesCache,
+		dedupCache:           dedupCache,
 		metrics:              NewServiceMetrics(),
 	}
 }
@@ -3931,6 +3946,25 @@ func (s *Service) finalizeSearchRun(state *searchRunState, canceled bool) {
 	s.searchMu.Unlock()
 }
 
+// dedupCacheKey generates a cache key for deduplication results based on instance ID
+// and a signature derived from torrent hashes. Uses XOR of xxhash values for order-independent
+// hashing - the same set of torrents produces the same key regardless of order.
+func dedupCacheKey(instanceID int, torrents []qbt.Torrent) string {
+	n := len(torrents)
+	if n == 0 {
+		return fmt.Sprintf("dedup:%d:0:0", instanceID)
+	}
+
+	// XOR all individual hash digests for order-independent signature.
+	// XOR is commutative and associative, so order doesn't matter.
+	var sig uint64
+	for i := range torrents {
+		sig ^= xxhash.Sum64String(torrents[i].Hash)
+	}
+
+	return fmt.Sprintf("dedup:%d:%d:%x", instanceID, n, sig)
+}
+
 // deduplicateSourceTorrents removes duplicate torrents from the search queue by keeping only
 // the best representative of each unique content group. It prefers torrents that already
 // have a top-level folder (so subsequent cross-seeds inherit cleaner layouts) and falls back to
@@ -3951,6 +3985,19 @@ func (s *Service) deduplicateSourceTorrents(ctx context.Context, instanceID int,
 		return torrents, map[string][]string{}
 	}
 
+	// Generate cache key from instance ID and an order-independent signature of torrent hashes.
+	cacheKey := dedupCacheKey(instanceID, torrents)
+	if s.dedupCache != nil {
+		if entry, ok := s.dedupCache.Get(cacheKey); ok && entry != nil {
+			log.Trace().
+				Int("instanceID", instanceID).
+				Int("cachedCount", len(entry.deduplicated)).
+				Msg("[CROSSSEED-DEDUP] Using cached deduplication result")
+			// IMPORTANT: Returned slices are cache-backed. Do not modify.
+			return entry.deduplicated, entry.duplicateHashes
+		}
+	}
+
 	// Parse all torrents and track their releases
 	type torrentWithRelease struct {
 		torrent         qbt.Torrent
@@ -4141,6 +4188,14 @@ func (s *Service) deduplicateSourceTorrents(ctx context.Context, instanceID int,
 		duplicateMap[rep.torrent.Hash] = append([]string(nil), group.duplicates...)
 	}
 
+	// Cache the result for future runs
+	if s.dedupCache != nil {
+		s.dedupCache.Set(cacheKey, &dedupCacheEntry{
+			deduplicated:    deduplicated,
+			duplicateHashes: duplicateMap,
+		}, ttlcache.DefaultTTL)
+	}
+
 	return deduplicated, duplicateMap
 }
 
@@ -4398,6 +4453,22 @@ func (s *Service) processSearchCandidate(ctx context.Context, state *searchRunSt
 		return nil
 	}
 
+	// Wait for RSS automation to complete before searching to avoid rate limiter contention.
+	// RSS uses higher priority (shorter intervals) so we yield to it.
+	// Cap the wait to 5 minutes to prevent indefinite blocking if RSS gets stuck.
+	rssWaitDeadline := time.After(5 * time.Minute)
+rssWaitLoop:
+	for s.runActive.Load() {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-rssWaitDeadline:
+			log.Warn().Msg("[CROSSSEED-SEARCH] RSS wait timeout exceeded, proceeding with search")
+			break rssWaitLoop
+		case <-time.After(500 * time.Millisecond):
+		}
+	}
+
 	searchCtx := ctx
 	var searchCancel context.CancelFunc
 	searchTimeout := computeAutomationSearchTimeout(len(allowedIndexerIDs))
@@ -4407,7 +4478,7 @@ func (s *Service) processSearchCandidate(ctx context.Context, state *searchRunSt
 	if searchCancel != nil {
 		defer searchCancel()
 	}
-	searchCtx = jackett.WithSearchPriority(searchCtx, jackett.RateLimitPriorityRSS)
+	searchCtx = jackett.WithSearchPriority(searchCtx, jackett.RateLimitPriorityBackground)
 
 	searchResp, err := s.SearchTorrentMatches(searchCtx, state.opts.InstanceID, torrent.Hash, TorrentSearchOptions{
 		IndexerIDs:             allowedIndexerIDs,

internal/services/jackett/ratelimiter.go

@@ -109,13 +109,15 @@ func (r *RateLimiter) BeforeRequest(ctx context.Context, indexer *models.Torznab
 		}
 
 		if cfg.MaxWait > 0 && wait > cfg.MaxWait {
-			// Treat large waits as a short-lived cooldown so schedulers can skip instead of retrying.
-			r.setCooldownLocked(indexer.ID, wait)
+			// Cap cooldown to maxWait to prevent cascading blocks across priorities.
+			// Setting the full wait would cause lower-priority tasks to inherit cooldowns
+			// far exceeding their own budgets, starving them unnecessarily.
+			r.setCooldownLocked(indexer.ID, cfg.MaxWait)
 			r.mu.Unlock()
 			return &RateLimitWaitError{
 				IndexerID:   indexer.ID,
 				IndexerName: indexer.Name,
-				Wait:        wait,
+				Wait:        cfg.MaxWait, // Report capped wait, not theoretical wait
 				MaxWait:     cfg.MaxWait,
 				Priority:    cfg.Priority,
 			}

internal/services/jackett/ratelimiter_test.go

@@ -89,7 +89,8 @@ func TestRateLimiterMaxWaitBudget(t *testing.T) {
 		t.Fatalf("expected RateLimitWaitError, got %v", err)
 	}
 
-	if waitErr.Wait <= waitErr.MaxWait {
-		t.Fatalf("expected wait to exceed max wait, got wait %v max %v", waitErr.Wait, waitErr.MaxWait)
+	// Wait is now capped to MaxWait to provide consistent semantics for callers
+	if waitErr.Wait != waitErr.MaxWait {
+		t.Fatalf("expected wait to be capped to max wait, got wait %v max %v", waitErr.Wait, waitErr.MaxWait)
 	}
 }