fix: YouTube timeout bump to 90s + Reddit 429 fail-fast
YouTube: Add youtube_future timeout key (60/90/120s for quick/default/deep) separate from the shared future timeout. YouTube needs more time because it does search + parallel transcript fetching. Previously, 20 videos + 5 transcripts exceeded the 60s budget and all results were discarded. Reddit 429: Propagate rate-limit errors instead of swallowing them. Enrichment now uses 10s timeout / 1 retry (was 30s / 3 retries). On first 429, cancel remaining enrichment and skip Phase 2 Reddit. Total time wasted on 429 drops from ~75s to ~12s. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
+24
-7
@@ -38,9 +38,9 @@ _child_pids: set = set()
|
|||||||
_child_pids_lock = threading.Lock()
|
_child_pids_lock = threading.Lock()
|
||||||
|
|
||||||
TIMEOUT_PROFILES = {
|
TIMEOUT_PROFILES = {
|
||||||
"quick": {"global": 90, "future": 30, "http": 15, "enrich_per": 8, "enrich_total": 30, "enrich_max_items": 10},
|
"quick": {"global": 90, "future": 30, "youtube_future": 60, "http": 15, "enrich_per": 8, "enrich_total": 30, "enrich_max_items": 10},
|
||||||
"default": {"global": 180, "future": 60, "http": 30, "enrich_per": 15, "enrich_total": 45, "enrich_max_items": 15},
|
"default": {"global": 180, "future": 60, "youtube_future": 90, "http": 30, "enrich_per": 15, "enrich_total": 45, "enrich_max_items": 15},
|
||||||
"deep": {"global": 300, "future": 90, "http": 30, "enrich_per": 15, "enrich_total": 60, "enrich_max_items": 25},
|
"deep": {"global": 300, "future": 90, "youtube_future": 120, "http": 30, "enrich_per": 15, "enrich_total": 60, "enrich_max_items": 25},
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -364,6 +364,7 @@ def _run_supplemental(
|
|||||||
depth: str,
|
depth: str,
|
||||||
x_source: str,
|
x_source: str,
|
||||||
progress: ui.ProgressDisplay = None,
|
progress: ui.ProgressDisplay = None,
|
||||||
|
skip_reddit: bool = False,
|
||||||
) -> tuple:
|
) -> tuple:
|
||||||
"""Run Phase 2 supplemental searches based on entities from Phase 1.
|
"""Run Phase 2 supplemental searches based on entities from Phase 1.
|
||||||
|
|
||||||
@@ -379,6 +380,7 @@ def _run_supplemental(
|
|||||||
depth: Research depth
|
depth: Research depth
|
||||||
x_source: 'bird' or 'xai'
|
x_source: 'bird' or 'xai'
|
||||||
progress: Optional progress display
|
progress: Optional progress display
|
||||||
|
skip_reddit: If True, skip Reddit supplemental (e.g. rate-limited)
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Tuple of (supplemental_reddit, supplemental_x)
|
Tuple of (supplemental_reddit, supplemental_x)
|
||||||
@@ -401,7 +403,7 @@ def _run_supplemental(
|
|||||||
)
|
)
|
||||||
|
|
||||||
has_handles = entities["x_handles"] and x_source == "bird"
|
has_handles = entities["x_handles"] and x_source == "bird"
|
||||||
has_subs = entities["reddit_subreddits"]
|
has_subs = entities["reddit_subreddits"] and not skip_reddit
|
||||||
|
|
||||||
if not has_handles and not has_subs:
|
if not has_handles and not has_subs:
|
||||||
return [], []
|
return [], []
|
||||||
@@ -642,12 +644,13 @@ def run_research(
|
|||||||
progress.end_x(len(x_items))
|
progress.end_x(len(x_items))
|
||||||
|
|
||||||
if youtube_future:
|
if youtube_future:
|
||||||
|
yt_timeout = timeouts.get("youtube_future", future_timeout)
|
||||||
try:
|
try:
|
||||||
youtube_items, youtube_error = youtube_future.result(timeout=future_timeout)
|
youtube_items, youtube_error = youtube_future.result(timeout=yt_timeout)
|
||||||
if youtube_error and progress:
|
if youtube_error and progress:
|
||||||
progress.show_error(f"YouTube error: {youtube_error}")
|
progress.show_error(f"YouTube error: {youtube_error}")
|
||||||
except TimeoutError:
|
except TimeoutError:
|
||||||
youtube_error = f"YouTube search timed out after {future_timeout}s"
|
youtube_error = f"YouTube search timed out after {yt_timeout}s"
|
||||||
if progress:
|
if progress:
|
||||||
progress.show_error(youtube_error)
|
progress.show_error(youtube_error)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
@@ -677,6 +680,7 @@ def run_research(
|
|||||||
enrich_max = timeouts["enrich_max_items"]
|
enrich_max = timeouts["enrich_max_items"]
|
||||||
enrich_total_timeout = timeouts["enrich_total"]
|
enrich_total_timeout = timeouts["enrich_total"]
|
||||||
items_to_enrich = reddit_items[:enrich_max]
|
items_to_enrich = reddit_items[:enrich_max]
|
||||||
|
rate_limited = False # Set True if Reddit returns 429 during enrichment
|
||||||
|
|
||||||
if items_to_enrich:
|
if items_to_enrich:
|
||||||
if progress:
|
if progress:
|
||||||
@@ -696,7 +700,9 @@ def run_research(
|
|||||||
raw_reddit_enriched.append(reddit_items[i])
|
raw_reddit_enriched.append(reddit_items[i])
|
||||||
else:
|
else:
|
||||||
# Parallel enrichment with bounded concurrency and total timeout
|
# Parallel enrichment with bounded concurrency and total timeout
|
||||||
|
# Uses short HTTP timeout (10s) and 1 retry to fail fast on 429
|
||||||
completed_count = 0
|
completed_count = 0
|
||||||
|
rate_limited = False
|
||||||
with ThreadPoolExecutor(max_workers=5) as enrich_pool:
|
with ThreadPoolExecutor(max_workers=5) as enrich_pool:
|
||||||
futures = {
|
futures = {
|
||||||
enrich_pool.submit(reddit_enrich.enrich_reddit_item, item): i
|
enrich_pool.submit(reddit_enrich.enrich_reddit_item, item): i
|
||||||
@@ -710,6 +716,16 @@ def run_research(
|
|||||||
progress.update_reddit_enrich(completed_count, len(items_to_enrich))
|
progress.update_reddit_enrich(completed_count, len(items_to_enrich))
|
||||||
try:
|
try:
|
||||||
reddit_items[idx] = future.result(timeout=timeouts["enrich_per"])
|
reddit_items[idx] = future.result(timeout=timeouts["enrich_per"])
|
||||||
|
except reddit_enrich.RedditRateLimitError:
|
||||||
|
rate_limited = True
|
||||||
|
if progress:
|
||||||
|
progress.show_error(
|
||||||
|
"Reddit rate-limited (429) — skipping remaining enrichment"
|
||||||
|
)
|
||||||
|
# Cancel remaining futures and bail
|
||||||
|
for f in futures:
|
||||||
|
f.cancel()
|
||||||
|
break
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
if progress:
|
if progress:
|
||||||
progress.show_error(
|
progress.show_error(
|
||||||
@@ -731,11 +747,12 @@ def run_research(
|
|||||||
progress.end_reddit_enrich()
|
progress.end_reddit_enrich()
|
||||||
|
|
||||||
# Phase 2: Supplemental search based on entities from Phase 1
|
# Phase 2: Supplemental search based on entities from Phase 1
|
||||||
# Skip on --quick (speed matters) and mock mode
|
# Skip on --quick (speed matters), mock mode, or if Reddit is rate-limiting
|
||||||
if depth != "quick" and not mock and (reddit_items or x_items):
|
if depth != "quick" and not mock and (reddit_items or x_items):
|
||||||
sup_reddit, sup_x = _run_supplemental(
|
sup_reddit, sup_x = _run_supplemental(
|
||||||
topic, reddit_items, x_items,
|
topic, reddit_items, x_items,
|
||||||
from_date, to_date, depth, x_source, progress,
|
from_date, to_date, depth, x_source, progress,
|
||||||
|
skip_reddit=rate_limited,
|
||||||
)
|
)
|
||||||
if sup_reddit:
|
if sup_reddit:
|
||||||
reddit_items.extend(sup_reddit)
|
reddit_items.extend(sup_reddit)
|
||||||
|
|||||||
+4
-2
@@ -124,11 +124,13 @@ def post(url: str, json_data: Dict[str, Any], headers: Optional[Dict[str, str]]
|
|||||||
return request("POST", url, headers=headers, json_data=json_data, **kwargs)
|
return request("POST", url, headers=headers, json_data=json_data, **kwargs)
|
||||||
|
|
||||||
|
|
||||||
def get_reddit_json(path: str) -> Dict[str, Any]:
|
def get_reddit_json(path: str, timeout: int = DEFAULT_TIMEOUT, retries: int = MAX_RETRIES) -> Dict[str, Any]:
|
||||||
"""Fetch Reddit thread JSON.
|
"""Fetch Reddit thread JSON.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
path: Reddit path (e.g., /r/subreddit/comments/id/title)
|
path: Reddit path (e.g., /r/subreddit/comments/id/title)
|
||||||
|
timeout: HTTP timeout per attempt in seconds
|
||||||
|
retries: Number of retries on failure
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Parsed JSON response
|
Parsed JSON response
|
||||||
@@ -149,4 +151,4 @@ def get_reddit_json(path: str) -> Dict[str, Any]:
|
|||||||
"Accept": "application/json",
|
"Accept": "application/json",
|
||||||
}
|
}
|
||||||
|
|
||||||
return get(url, headers=headers)
|
return get(url, headers=headers, timeout=timeout, retries=retries)
|
||||||
|
|||||||
@@ -235,7 +235,7 @@ def search_subreddits(
|
|||||||
"Accept": "application/json",
|
"Accept": "application/json",
|
||||||
}
|
}
|
||||||
|
|
||||||
data = http.get(full_url, headers=headers, timeout=15)
|
data = http.get(full_url, headers=headers, timeout=15, retries=1)
|
||||||
|
|
||||||
# Reddit search returns {"data": {"children": [...]}}
|
# Reddit search returns {"data": {"children": [...]}}
|
||||||
children = data.get("data", {}).get("children", [])
|
children = data.get("data", {}).get("children", [])
|
||||||
@@ -267,6 +267,9 @@ def search_subreddits(
|
|||||||
|
|
||||||
except http.HTTPError as e:
|
except http.HTTPError as e:
|
||||||
_log_info(f"Subreddit search failed for r/{sub}: {e}")
|
_log_info(f"Subreddit search failed for r/{sub}: {e}")
|
||||||
|
if e.status_code == 429:
|
||||||
|
_log_info("Reddit rate-limited (429) — skipping remaining subreddits")
|
||||||
|
break
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
_log_info(f"Subreddit search error for r/{sub}: {e}")
|
_log_info(f"Subreddit search error for r/{sub}: {e}")
|
||||||
|
|
||||||
|
|||||||
@@ -25,15 +25,30 @@ def extract_reddit_path(url: str) -> Optional[str]:
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
def fetch_thread_data(url: str, mock_data: Optional[Dict] = None) -> Optional[Dict[str, Any]]:
|
class RedditRateLimitError(Exception):
|
||||||
|
"""Raised when Reddit returns HTTP 429 (rate limited)."""
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def fetch_thread_data(
|
||||||
|
url: str,
|
||||||
|
mock_data: Optional[Dict] = None,
|
||||||
|
timeout: int = 30,
|
||||||
|
retries: int = 3,
|
||||||
|
) -> Optional[Dict[str, Any]]:
|
||||||
"""Fetch Reddit thread JSON data.
|
"""Fetch Reddit thread JSON data.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
url: Reddit thread URL
|
url: Reddit thread URL
|
||||||
mock_data: Mock data for testing
|
mock_data: Mock data for testing
|
||||||
|
timeout: HTTP timeout per attempt in seconds
|
||||||
|
retries: Number of retries on failure
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Thread data dict or None on failure
|
Thread data dict or None on failure
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
RedditRateLimitError: When Reddit returns 429 (caller should bail)
|
||||||
"""
|
"""
|
||||||
if mock_data is not None:
|
if mock_data is not None:
|
||||||
return mock_data
|
return mock_data
|
||||||
@@ -43,9 +58,11 @@ def fetch_thread_data(url: str, mock_data: Optional[Dict] = None) -> Optional[Di
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
try:
|
try:
|
||||||
data = http.get_reddit_json(path)
|
data = http.get_reddit_json(path, timeout=timeout, retries=retries)
|
||||||
return data
|
return data
|
||||||
except http.HTTPError:
|
except http.HTTPError as e:
|
||||||
|
if e.status_code == 429:
|
||||||
|
raise RedditRateLimitError(f"Reddit rate limited (429) fetching {url}") from e
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
@@ -178,20 +195,27 @@ def extract_comment_insights(comments: List[Dict], limit: int = 7) -> List[str]:
|
|||||||
def enrich_reddit_item(
|
def enrich_reddit_item(
|
||||||
item: Dict[str, Any],
|
item: Dict[str, Any],
|
||||||
mock_thread_data: Optional[Dict] = None,
|
mock_thread_data: Optional[Dict] = None,
|
||||||
|
timeout: int = 10,
|
||||||
|
retries: int = 1,
|
||||||
) -> Dict[str, Any]:
|
) -> Dict[str, Any]:
|
||||||
"""Enrich a Reddit item with real engagement data.
|
"""Enrich a Reddit item with real engagement data.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
item: Reddit item dict
|
item: Reddit item dict
|
||||||
mock_thread_data: Mock data for testing
|
mock_thread_data: Mock data for testing
|
||||||
|
timeout: HTTP timeout per attempt (default 10s for enrichment)
|
||||||
|
retries: Number of retries (default 1 — fail fast for enrichment)
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Enriched item dict
|
Enriched item dict
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
RedditRateLimitError: Propagated so caller can bail on remaining items
|
||||||
"""
|
"""
|
||||||
url = item.get("url", "")
|
url = item.get("url", "")
|
||||||
|
|
||||||
# Fetch thread data
|
# Fetch thread data (RedditRateLimitError propagates to caller)
|
||||||
thread_data = fetch_thread_data(url, mock_thread_data)
|
thread_data = fetch_thread_data(url, mock_thread_data, timeout=timeout, retries=retries)
|
||||||
if not thread_data:
|
if not thread_data:
|
||||||
return item
|
return item
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user