From 04401f64cd7e8538dde943625d487f2a7880ef5e Mon Sep 17 00:00:00 2001 From: "renovate[bot]" <29139614+renovate[bot]@users.noreply.github.com> Date: Tue, 1 Sep 2026 13:28:39 -0400 Subject: [PATCH 01/12] Update dependency youtube-transcript-api to >=1.2.4,<1.3 (#3861) * Update dependency youtube-transcript-api to >=1.2.4,<1.3 * Migrate to youtube-transcript-api instance API v1.2.0 removed the static list_transcripts/get_transcript methods that were deprecated in v1.0.0; the instance method list() is the replacement. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_018S52vCvVmYyY7Q3ji519Nn --------- Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com> Co-authored-by: Tobias Macey Co-authored-by: Claude Fable 5 --- learning_resources/etl/youtube.py | 4 +--- learning_resources/etl/youtube_test.py | 16 ++++++++-------- pyproject.toml | 2 +- uv.lock | 8 ++++---- 4 files changed, 14 insertions(+), 16 deletions(-) diff --git a/learning_resources/etl/youtube.py b/learning_resources/etl/youtube.py index 07e60362da..b949970008 100644 --- a/learning_resources/etl/youtube.py +++ b/learning_resources/etl/youtube.py @@ -336,9 +336,7 @@ def get_captions_for_video(video_resource: LearningResource) -> str: """ try: - transcript_list = YouTubeTranscriptApi.list_transcripts( - video_resource.readable_id - ) + transcript_list = YouTubeTranscriptApi().list(video_resource.readable_id) transcript = transcript_list.find_manually_created_transcript(["en"]) if not transcript: transcript = transcript_list.find_generated_transcript(["en"]) diff --git a/learning_resources/etl/youtube_test.py b/learning_resources/etl/youtube_test.py index 2f2bb4562f..bff68851c3 100644 --- a/learning_resources/etl/youtube_test.py +++ b/learning_resources/etl/youtube_test.py @@ -255,8 +255,8 @@ def test_get_captions_for_video(mocker): """Test fetching caption data for a video when non auto-generated english caption is available""" caption_text = "English: Not Auto-generated" mock_captions_api = mocker.patch( - "learning_resources.etl.youtube.YouTubeTranscriptApi.list_transcripts" - ) + "learning_resources.etl.youtube.YouTubeTranscriptApi" + ).return_value.list mock_captions_api.return_value.find_manually_created_transcript.return_value = Mock( fetch=Mock( return_value=[ @@ -273,8 +273,8 @@ def test_get_captions_for_video_autogenerated_only(mocker): """Test fetching caption data for a video when only auto-generated english caption is available""" auto_text = "English: Auto-generated" mock_captions_api = mocker.patch( - "learning_resources.etl.youtube.YouTubeTranscriptApi.list_transcripts" - ) + "learning_resources.etl.youtube.YouTubeTranscriptApi" + ).return_value.list mock_captions_api.return_value.find_manually_created_transcript.return_value = None mock_captions_api.return_value.find_generated_transcript.return_value = Mock( fetch=Mock( @@ -290,8 +290,8 @@ def test_get_captions_for_video_autogenerated_only(mocker): def test_get_captions_for_video_no_english_caption(mocker): """Test fetching caption data for a video with no english caption available""" mock_captions_api = mocker.patch( - "learning_resources.etl.youtube.YouTubeTranscriptApi.list_transcripts" - ) + "learning_resources.etl.youtube.YouTubeTranscriptApi" + ).return_value.list mock_captions_api.return_value.find_manually_created_transcript.return_value = None mock_captions_api.return_value.find_generated_transcript.return_value = None @@ -302,8 +302,8 @@ def test_get_captions_for_video_no_english_caption(mocker): def test_get_captions_for_video_no_captions(mocker): """Test fetching caption data for a video with no captions available""" mock_captions_api = mocker.patch( - "learning_resources.etl.youtube.YouTubeTranscriptApi.list_transcripts" - ) + "learning_resources.etl.youtube.YouTubeTranscriptApi" + ).return_value.list mock_captions_api.side_effect = NoTranscriptFound("abc", "en", {}) video = Mock() diff --git a/pyproject.toml b/pyproject.toml index a39c0ad33d..fcc4d82de6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -113,7 +113,7 @@ dependencies = [ "ulid-py>=1.0.0,<2", "urllib3>=2.0.0,<3", "wrapt>=1.14.1,<2", - "youtube-transcript-api>=1.0.0,<1.1", + "youtube-transcript-api>=1.2.4,<1.3", "pypdf>=6.9.1,<7", "pypdfium2>=5.0.0,<6", "pyarrow>=24,<25", diff --git a/uv.lock b/uv.lock index ecd01fd729..be407c3cbd 100644 --- a/uv.lock +++ b/uv.lock @@ -2747,7 +2747,7 @@ requires-dist = [ { name = "ulid-py", specifier = ">=1.0.0,<2" }, { name = "urllib3", specifier = ">=2.0.0,<3" }, { name = "wrapt", specifier = ">=1.14.1,<2" }, - { name = "youtube-transcript-api", specifier = ">=1.0.0,<1.1" }, + { name = "youtube-transcript-api", specifier = ">=1.2.4,<1.3" }, ] [package.metadata.requires-dev] @@ -5417,15 +5417,15 @@ wheels = [ [[package]] name = "youtube-transcript-api" -version = "1.0.3" +version = "1.2.4" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "defusedxml" }, { name = "requests" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/b0/32/f60d87a99c05a53604c58f20f670c7ea6262b55e0bbeb836ffe4550b248b/youtube_transcript_api-1.0.3.tar.gz", hash = "sha256:902baf90e7840a42e1e148335e09fe5575dbff64c81414957aea7038e8a4db46", size = 2153252, upload-time = "2025-03-25T18:14:21.119Z" } +sdist = { url = "https://files.pythonhosted.org/packages/60/43/4104185a2eaa839daa693b30e15c37e7e58795e8e09ec414f22b3db54bec/youtube_transcript_api-1.2.4.tar.gz", hash = "sha256:b72d0e96a335df599d67cee51d49e143cff4f45b84bcafc202ff51291603ddcd", size = 469839, upload-time = "2026-01-29T09:09:17.088Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/f0/44/40c03bb0f8bddfb9d2beff2ed31641f52d96c287ba881d20e0c074784ac2/youtube_transcript_api-1.0.3-py3-none-any.whl", hash = "sha256:d1874e57de65cf14c9d7d09b2b37c814d6287fa0e770d4922c4cd32a5b3f6c47", size = 2169911, upload-time = "2025-03-25T18:14:19.416Z" }, + { url = "https://files.pythonhosted.org/packages/be/95/129ea37efd6cd6ed00f62baae6543345c677810b8a3bf0026756e1d3cf3c/youtube_transcript_api-1.2.4-py3-none-any.whl", hash = "sha256:03878759356da5caf5edac77431780b91448fb3d8c21d4496015bdc8a7bc43ff", size = 485227, upload-time = "2026-01-29T09:09:15.427Z" }, ] [[package]] From e39c8e09888ca57c18a769a879fffabdbc6850f8 Mon Sep 17 00:00:00 2001 From: Shankar Ambady Date: Tue, 1 Sep 2026 13:32:47 -0400 Subject: [PATCH 02/12] vector search: embedding call request hedging (#3853) * hedge requests * tests * switch default hedging delay to 0 * change default delay * change default delay * Isolate and bound hedged embedding request capacity Address review feedback that abandoned losing requests could occupy the shared pool and make later primary requests queue behind them. - split the primary and speculative requests into separate thread pools - bound each pool with a semaphore so submissions are rejected rather than queued: a full primary pool runs the request inline, a full hedge pool skips the backups - give every hedged provider call an explicit timeout so losers release their worker - cover pool saturation, inline fallback, and concurrent saturation in tests Co-Authored-By: Claude Opus 5 * fix ssl_cert_reqs * Address PR feedback on embedding request hedging - fail loudly with RuntimeError instead of returning None when every hedged request is abandoned - drop the duplicated hedge_count/hedge_delay defaults, embed_query always passes them explicitly - log when backups are fired and whether the primary or a backup won, so the hedge delay can be tuned from prod - cover EMBEDDING_HEDGE_COUNT=1 (short circuits to unhedged) and =3 Co-Authored-By: Claude Opus 5 --------- Co-authored-by: Claude Opus 5 --- main/settings.py | 24 ++ vector_search/encoders/litellm.py | 202 ++++++++++++- vector_search/encoders/litellm_test.py | 375 ++++++++++++++++++++++++- 3 files changed, 597 insertions(+), 4 deletions(-) diff --git a/main/settings.py b/main/settings.py index cdc5ef75ee..2c6bc7ed00 100644 --- a/main/settings.py +++ b/main/settings.py @@ -925,6 +925,30 @@ def get_all_config_keys(): default=None, ) +# Hedged embedding requests: if the first request for a search query has not +# come back within EMBEDDING_HEDGE_DELAY_SECONDS, send backup requests and use +# whichever finishes first. The delay keeps the extra backend load proportional +# to the tail rather than doubling it on every query - set it to 0 to always +# send the backups immediately. +EMBEDDING_REQUEST_HEDGING_ENABLED = get_bool( + name="EMBEDDING_REQUEST_HEDGING_ENABLED", default=True +) +EMBEDDING_HEDGE_COUNT = get_int(name="EMBEDDING_HEDGE_COUNT", default=2) +EMBEDDING_HEDGE_DELAY_SECONDS = get_float( + name="EMBEDDING_HEDGE_DELAY_SECONDS", default=0.45 +) +# Primary and speculative query embedding requests run in separate, bounded +# thread pools so abandoned hedges cannot starve later primary requests. Both +# pools reject work when full: a saturated primary pool falls back to running +# the request inline, a saturated hedge pool skips the backups. +EMBEDDING_QUERY_MAX_WORKERS = get_int(name="EMBEDDING_QUERY_MAX_WORKERS", default=16) +EMBEDDING_HEDGE_MAX_WORKERS = get_int(name="EMBEDDING_HEDGE_MAX_WORKERS", default=16) +# Explicit per-request timeout for hedged query embeddings, so a losing request +# releases its worker instead of occupying it for the life of the connection. +EMBEDDING_HEDGE_REQUEST_TIMEOUT_SECONDS = get_float( + name="EMBEDDING_HEDGE_REQUEST_TIMEOUT_SECONDS", default=10.0 +) + CONTENT_FILE_EMBEDDING_CHUNK_SIZE_OVERRIDE = get_int( name="CONTENT_FILE_EMBEDDING_CHUNK_SIZE", default=512 ) diff --git a/vector_search/encoders/litellm.py b/vector_search/encoders/litellm.py index ec67ad4359..5382848e73 100644 --- a/vector_search/encoders/litellm.py +++ b/vector_search/encoders/litellm.py @@ -1,5 +1,9 @@ +"""LiteLLM encoder module for vector search.""" + +import concurrent.futures import logging import os +import threading from urllib.parse import urlparse import litellm @@ -25,12 +29,103 @@ port=redis_url.port, password=redis_url.password, ssl=redis_ssl, - ssl_cert_reqs="required" if redis_ssl else None, supported_call_types=["embedding", "aembedding"], ttl=settings.QDRANT_QUERY_EMBEDDING_CACHE_TTL, + **({"ssl_cert_reqs": "required"} if redis_ssl else {}), ) +class CapacityLimitedExecutor: + """ + Thread pool that rejects work instead of queueing it. + + ``ThreadPoolExecutor`` has an unbounded work queue, so a burst of slow + requests would silently pile up behind the busy workers. Here every + in-flight task holds a slot for its whole lifetime, so at most + ``max_workers`` tasks exist at once and ``submit`` returns ``None`` when + the pool is saturated, letting the caller degrade instead of queueing. + """ + + def __init__(self, max_workers, thread_name_prefix): + """Create the pool and the semaphore bounding its in-flight work.""" + self._executor = concurrent.futures.ThreadPoolExecutor( + max_workers=max_workers, + thread_name_prefix=thread_name_prefix, + ) + self._slots = threading.BoundedSemaphore(max_workers) + + def submit(self, fn, *args, **kwargs): + """Submit ``fn``, returning ``None`` if every worker is busy.""" + if not self._slots.acquire(blocking=False): + return None + try: + future = self._executor.submit(fn, *args, **kwargs) + except RuntimeError: + # pool already shut down + self._slots.release() + return None + # the slot is held until the call finishes, including for losing + # requests the caller has already abandoned + future.add_done_callback(lambda _future: self._slots.release()) + return future + + def shutdown(self): + """Shut down the underlying pool without waiting on in-flight calls.""" + self._executor.shutdown(wait=False) + + +# Speculative requests get their own pool so a pile-up of abandoned hedges can +# never take capacity away from the primary request of a later query. +_executors = {} +_executor_lock = threading.Lock() + + +def _get_executor(key, max_workers, thread_name_prefix): + """ + Return (creating if needed) one of the process-wide embedding thread pools. + + Created lazily so that settings overrides (in tests) are picked up. + """ + executor = _executors.get(key) + if executor is None: + with _executor_lock: + executor = _executors.get(key) + if executor is None: + executor = CapacityLimitedExecutor( + max_workers=max_workers, + thread_name_prefix=thread_name_prefix, + ) + _executors[key] = executor + return executor + + +def get_primary_executor(): + """Return the pool running the first (non-speculative) embedding request.""" + return _get_executor( + "primary", + settings.EMBEDDING_QUERY_MAX_WORKERS, + "embed-query", + ) + + +def get_hedge_executor(): + """Return the pool running speculative (backup) embedding requests.""" + return _get_executor( + "hedge", + settings.EMBEDDING_HEDGE_MAX_WORKERS, + "embed-hedge", + ) + + +def reset_embedding_executors(): + """Drop the shared embedding executors. Intended for tests.""" + with _executor_lock: + executors = list(_executors.values()) + _executors.clear() + for executor in executors: + executor.shutdown() + + class LiteLLMEncoder(BaseEncoder): """ LiteLLM encoder @@ -39,6 +134,7 @@ class LiteLLMEncoder(BaseEncoder): token_encoding_name = settings.LITELLM_TOKEN_ENCODING_NAME def __init__(self, model_name): + """Initialize LiteLLM encoder with model name.""" self.model_name = model_name try: self.token_encoding_name = tiktoken.encoding_name_for_model(model_name) @@ -47,9 +143,108 @@ def __init__(self, model_name): log.warning(msg) def embed_documents(self, documents): + """Embed a list of documents without hedging.""" return [result["embedding"] for result in self.get_embedding(documents)["data"]] - def get_embedding(self, texts): + def _hedged_get_embedding(self, texts, hedge_count, hedge_delay): + """ + Run concurrent embedding requests and return the first successful result. + + The first request is issued immediately. When ``hedge_delay`` is + positive the backup requests are only issued if the first request has + not completed within that many seconds, so the extra backend load is + paid on the tail rather than on every query. Losing requests are + abandoned rather than waited on - the caller returns as soon as any + request succeeds - but each one carries an explicit request timeout so + it releases its worker instead of occupying it indefinitely. + + Both pools reject work when saturated: if the primary pool is full the + request runs inline on the calling thread, and if the hedge pool is + full the backups are skipped. Under load the behavior degrades to + un-hedged requests rather than queueing behind abandoned losers. + """ + timeout_seconds = settings.EMBEDDING_HEDGE_REQUEST_TIMEOUT_SECONDS + primary = get_primary_executor().submit( + self.get_embedding, texts, request_timeout=timeout_seconds + ) + if primary is None: + log.warning( + "Embedding pool saturated, running embedding request without hedging" + ) + return self.get_embedding(texts, request_timeout=timeout_seconds) + + pending = {primary} + hedge_futures = set() + hedges_remaining = max(hedge_count - 1, 0) + # only bound the wait while there are backups left to send + timeout = hedge_delay if hedges_remaining else None + first_exception = None + + while pending: + done, pending = concurrent.futures.wait( + pending, + timeout=timeout, + return_when=concurrent.futures.FIRST_COMPLETED, + ) + for fut in done: + try: + result = fut.result() + except Exception as exc: # noqa: BLE001 + log.warning("Hedged embedding request failed with error: %s", exc) + if first_exception is None: + first_exception = exc + else: + log.info( + "Embedding request won by %s request", + "backup" if fut in hedge_futures else "primary", + ) + return result + + if hedges_remaining: + # the first request is slow (or failed) - send the backups + hedge_executor = get_hedge_executor() + hedges = { + hedge_executor.submit( + self.get_embedding, texts, request_timeout=timeout_seconds + ) + for _ in range(hedges_remaining) + } + hedges.discard(None) + if hedges: + log.info( + "Sent %d backup embedding request(s) after %ss", + len(hedges), + hedge_delay, + ) + else: + log.warning("Hedge pool saturated, skipping backup requests") + hedge_futures |= hedges + pending |= hedges + hedges_remaining = 0 + timeout = None + + if first_exception is not None: + raise first_exception + # unreachable: pending only empties once a request returned or raised + msg = "All hedged embedding requests were abandoned" + raise RuntimeError(msg) + + def embed_query(self, query): + """ + Embed a single search query, hedging the request to mitigate tail latency. + """ + hedge_count = settings.EMBEDDING_HEDGE_COUNT + if settings.EMBEDDING_REQUEST_HEDGING_ENABLED and hedge_count > 1: + res = self._hedged_get_embedding( + [query], + hedge_count=hedge_count, + hedge_delay=settings.EMBEDDING_HEDGE_DELAY_SECONDS, + ) + return res["data"][0]["embedding"] + return self.embed(query) + + def get_embedding(self, texts, request_timeout=None): + """Generate embeddings using LiteLLM/OpenAI API.""" if self.cache: cache_params = { "caching": True, @@ -66,4 +261,7 @@ def get_embedding(self, texts): config["custom_llm_provider"] = settings.LITELLM_CUSTOM_PROVIDER if settings.LITELLM_API_BASE: config["api_base"] = settings.LITELLM_API_BASE + if request_timeout: + # bounds abandoned losing requests so they release their worker + config["timeout"] = request_timeout return embedding(**config).to_dict() diff --git a/vector_search/encoders/litellm_test.py b/vector_search/encoders/litellm_test.py index 94aec7b072..77f4c096b1 100644 --- a/vector_search/encoders/litellm_test.py +++ b/vector_search/encoders/litellm_test.py @@ -1,13 +1,57 @@ -from unittest.mock import patch +"""Tests for LiteLLM encoder.""" + +import concurrent.futures +import threading +import time +from collections import Counter +from unittest.mock import MagicMock, patch import pytest from django.conf import settings -from vector_search.encoders.litellm import LiteLLMEncoder +from vector_search.encoders.litellm import ( + LiteLLMEncoder, + get_hedge_executor, + get_primary_executor, + reset_embedding_executors, +) pytestmark = pytest.mark.django_db +@pytest.fixture(autouse=True) +def _fresh_embedding_executors(): + """Isolate the shared embedding thread pools between tests""" + reset_embedding_executors() + yield + reset_embedding_executors() + + +def _embedding_response(vector): + """Build a mock litellm embedding response""" + mock_resp = MagicMock() + mock_resp.to_dict.return_value = {"data": [{"embedding": vector}]} + return mock_resp + + +def _timed_responses(*delays): + """ + Return a mock side effect that sleeps a different amount per call. + + Each successive call sleeps the next value in ``delays``. + """ + counter = iter(range(len(delays))) + lock = threading.Lock() + + def side_effect(**_kwargs): + with lock: + idx = next(counter) + time.sleep(delays[idx]) + return _embedding_response([float(idx), 0.2, 0.3]) + + return side_effect + + @patch("vector_search.encoders.litellm.embedding") def test_litellm_encoder_cache_enabled(mock_embedding): """ @@ -56,3 +100,330 @@ def test_litellm_encoder_cache_disabled(mock_embedding): expected_kwargs["api_base"] = settings.LITELLM_API_BASE mock_embedding.assert_called_once_with(**expected_kwargs) + + +@patch("vector_search.encoders.litellm.embedding") +def test_litellm_encoder_embed_query_hedging_enabled(mock_embedding, settings): + """ + Test that embed_query issues hedged requests when the first one is slow + """ + settings.EMBEDDING_REQUEST_HEDGING_ENABLED = True + settings.EMBEDDING_HEDGE_COUNT = 2 + settings.EMBEDDING_HEDGE_DELAY_SECONDS = 0.05 + + # first request never finishes in time, the hedge is fast + mock_embedding.side_effect = _timed_responses(5.0, 0.0) + encoder = LiteLLMEncoder("test_model") + res = encoder.embed_query("search query") + + assert res == [1.0, 0.2, 0.3] + assert mock_embedding.call_count == 2 + + +@patch("vector_search.encoders.litellm.embedding") +def test_litellm_encoder_embed_query_does_not_wait_for_losers(mock_embedding, settings): + """ + Test that embed_query returns on the first success without waiting + for slower in-flight hedges to finish + """ + settings.EMBEDDING_REQUEST_HEDGING_ENABLED = True + settings.EMBEDDING_HEDGE_COUNT = 2 + settings.EMBEDDING_HEDGE_DELAY_SECONDS = 0 + + slow = 5.0 + mock_embedding.side_effect = _timed_responses(slow, 0.0) + encoder = LiteLLMEncoder("test_model") + + start = time.monotonic() + res = encoder.embed_query("search query") + elapsed = time.monotonic() - start + + assert res == [1.0, 0.2, 0.3] + # must return at the speed of the winner, not the loser + assert elapsed < slow / 2 + + +@patch("vector_search.encoders.litellm.embedding") +def test_litellm_encoder_embed_query_hedge_skipped_when_fast(mock_embedding, settings): + """ + Test that no backup request is sent when the first one beats the hedge delay + """ + settings.EMBEDDING_REQUEST_HEDGING_ENABLED = True + settings.EMBEDDING_HEDGE_COUNT = 2 + settings.EMBEDDING_HEDGE_DELAY_SECONDS = 5.0 + + mock_embedding.side_effect = _timed_responses(0.0, 0.0) + encoder = LiteLLMEncoder("test_model") + res = encoder.embed_query("search query") + + assert res == [0.0, 0.2, 0.3] + assert mock_embedding.call_count == 1 + + +@patch("vector_search.encoders.litellm.embedding") +def test_litellm_encoder_embed_query_hedge_covers_failure(mock_embedding, settings): + """ + Test that a failing request falls back to a hedged request + """ + settings.EMBEDDING_REQUEST_HEDGING_ENABLED = True + settings.EMBEDDING_HEDGE_COUNT = 2 + settings.EMBEDDING_HEDGE_DELAY_SECONDS = 5.0 + + mock_embedding.side_effect = [ + ConnectionError("boom"), + _embedding_response([0.1, 0.2, 0.3]), + ] + encoder = LiteLLMEncoder("test_model") + res = encoder.embed_query("search query") + + assert res == [0.1, 0.2, 0.3] + assert mock_embedding.call_count == 2 + + +@patch("vector_search.encoders.litellm.embedding") +def test_litellm_encoder_embed_query_raises_when_all_fail(mock_embedding, settings): + """ + Test that the first error is raised when every hedged request fails + """ + settings.EMBEDDING_REQUEST_HEDGING_ENABLED = True + settings.EMBEDDING_HEDGE_COUNT = 2 + settings.EMBEDDING_HEDGE_DELAY_SECONDS = 0 + + mock_embedding.side_effect = ConnectionError("boom") + encoder = LiteLLMEncoder("test_model") + + with pytest.raises(ConnectionError, match="boom"): + encoder.embed_query("search query") + + assert mock_embedding.call_count == 2 + + +@patch("vector_search.encoders.litellm.embedding") +def test_litellm_encoder_embed_query_hedging_disabled(mock_embedding, settings): + """ + Test that embed_query issues only a single request when hedging is disabled + """ + settings.EMBEDDING_REQUEST_HEDGING_ENABLED = False + + mock_embedding.return_value.to_dict.return_value = { + "data": [{"embedding": [0.1, 0.2, 0.3]}] + } + encoder = LiteLLMEncoder("test_model") + res = encoder.embed_query("search query") + + assert res == [0.1, 0.2, 0.3] + assert mock_embedding.call_count == 1 + + +@patch("vector_search.encoders.litellm.embedding") +def test_litellm_encoder_embed_query_sets_request_timeout(mock_embedding, settings): + """ + Test that hedged requests carry an explicit timeout so losing requests + release their worker instead of occupying it indefinitely + """ + settings.EMBEDDING_REQUEST_HEDGING_ENABLED = True + settings.EMBEDDING_HEDGE_COUNT = 2 + settings.EMBEDDING_HEDGE_DELAY_SECONDS = 0 + settings.EMBEDDING_HEDGE_REQUEST_TIMEOUT_SECONDS = 3.5 + + mock_embedding.side_effect = _timed_responses(0.0, 0.0) + encoder = LiteLLMEncoder("test_model") + encoder.embed_query("search query") + + assert mock_embedding.call_args_list + assert all(call.kwargs["timeout"] == 3.5 for call in mock_embedding.call_args_list) + + +@patch("vector_search.encoders.litellm.embedding") +def test_litellm_encoder_embed_query_inline_when_primary_pool_full( + mock_embedding, settings +): + """ + Test that a query runs inline on the calling thread rather than queueing + behind abandoned requests when the primary pool is saturated + """ + settings.EMBEDDING_REQUEST_HEDGING_ENABLED = True + settings.EMBEDDING_HEDGE_COUNT = 2 + settings.EMBEDDING_HEDGE_DELAY_SECONDS = 0 + settings.EMBEDDING_QUERY_MAX_WORKERS = 1 + + release = threading.Event() + calling_threads = [] + + def side_effect(**_kwargs): + calling_threads.append(threading.current_thread().name) + return _embedding_response([0.1, 0.2, 0.3]) + + mock_embedding.side_effect = side_effect + executor = get_primary_executor() + # occupy the only worker, the pool must then reject rather than queue + assert executor.submit(release.wait, 10) is not None + assert executor.submit(release.wait, 10) is None + + encoder = LiteLLMEncoder("test_model") + try: + res = encoder.embed_query("search query") + finally: + release.set() + + assert res == [0.1, 0.2, 0.3] + assert mock_embedding.call_count == 1 + assert calling_threads == [threading.current_thread().name] + + +@patch("vector_search.encoders.litellm.embedding") +def test_litellm_encoder_embed_query_skips_hedges_when_hedge_pool_full( + mock_embedding, settings +): + """ + Test that backup requests are skipped when the hedge pool is saturated + """ + settings.EMBEDDING_REQUEST_HEDGING_ENABLED = True + settings.EMBEDDING_HEDGE_COUNT = 2 + settings.EMBEDDING_HEDGE_DELAY_SECONDS = 0 + settings.EMBEDDING_HEDGE_MAX_WORKERS = 1 + + release = threading.Event() + assert get_hedge_executor().submit(release.wait, 10) is not None + + mock_embedding.side_effect = _timed_responses(0.05, 0.0) + encoder = LiteLLMEncoder("test_model") + try: + res = encoder.embed_query("search query") + finally: + release.set() + + assert res == [0.0, 0.2, 0.3] + assert mock_embedding.call_count == 1 + + +@patch("vector_search.encoders.litellm.embedding") +def test_litellm_encoder_embed_query_bounded_under_saturation(mock_embedding, settings): + """ + Test that concurrent queries stay bounded by the configured pool sizes and + still complete while earlier requests are in flight + """ + settings.EMBEDDING_REQUEST_HEDGING_ENABLED = True + settings.EMBEDDING_HEDGE_COUNT = 2 + settings.EMBEDDING_HEDGE_DELAY_SECONDS = 0 + settings.EMBEDDING_QUERY_MAX_WORKERS = 2 + settings.EMBEDDING_HEDGE_MAX_WORKERS = 2 + + query_count = 8 + release = threading.Event() + lock = threading.Lock() + in_flight = Counter() + peak = Counter() + + def _pool_of(thread_name): + if thread_name.startswith("embed-query"): + return "primary" + if thread_name.startswith("embed-hedge"): + return "hedge" + return "inline" + + def side_effect(**_kwargs): + pool = _pool_of(threading.current_thread().name) + with lock: + in_flight[pool] += 1 + peak[pool] = max(peak[pool], in_flight[pool]) + try: + release.wait(10) + return _embedding_response([0.1, 0.2, 0.3]) + finally: + with lock: + in_flight[pool] -= 1 + + mock_embedding.side_effect = side_effect + encoder = LiteLLMEncoder("test_model") + + with concurrent.futures.ThreadPoolExecutor(max_workers=query_count) as callers: + futures = [ + callers.submit(encoder.embed_query, f"search query {idx}") + for idx in range(query_count) + ] + # wait for every caller to have a request in flight before releasing + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + with lock: + if sum(in_flight.values()) >= query_count: + break + time.sleep(0.01) + release.set() + results = [fut.result(timeout=10) for fut in futures] + + assert results == [[0.1, 0.2, 0.3]] * query_count + # no unbounded queueing - each pool stays within its configured size + assert peak["primary"] <= settings.EMBEDDING_QUERY_MAX_WORKERS + assert peak["hedge"] <= settings.EMBEDDING_HEDGE_MAX_WORKERS + # queries that could not get a pool slot degraded to running inline + assert peak["inline"] > 0 + + +@patch("vector_search.encoders.litellm.embedding") +def test_litellm_encoder_embed_query_hedge_count_one(mock_embedding, settings): + """ + Test that a hedge count of 1 short circuits to a plain un-hedged request + """ + settings.EMBEDDING_REQUEST_HEDGING_ENABLED = True + settings.EMBEDDING_HEDGE_COUNT = 1 + settings.EMBEDDING_HEDGE_DELAY_SECONDS = 0 + + calling_threads = [] + + def side_effect(**_kwargs): + calling_threads.append(threading.current_thread().name) + return _embedding_response([0.1, 0.2, 0.3]) + + mock_embedding.side_effect = side_effect + encoder = LiteLLMEncoder("test_model") + res = encoder.embed_query("search query") + + assert res == [0.1, 0.2, 0.3] + assert mock_embedding.call_count == 1 + # no thread pool and no hedge request timeout are involved + assert calling_threads == [threading.current_thread().name] + assert "timeout" not in mock_embedding.call_args.kwargs + + +@patch("vector_search.encoders.litellm.embedding") +def test_litellm_encoder_embed_query_hedge_count_three(mock_embedding, settings): + """ + Test that a hedge count of 3 sends two backups and returns the first success + """ + settings.EMBEDDING_REQUEST_HEDGING_ENABLED = True + settings.EMBEDDING_HEDGE_COUNT = 3 + settings.EMBEDDING_HEDGE_DELAY_SECONDS = 0 + + slow = 5.0 + lock = threading.Lock() + calls = [] + + def side_effect(**_kwargs): + with lock: + calls.append(threading.current_thread().name) + # only the primary request is slow, both backups are fast + is_primary = len(calls) == 1 + if is_primary: + time.sleep(slow) + return _embedding_response([0.1, 0.2, 0.3]) + + mock_embedding.side_effect = side_effect + encoder = LiteLLMEncoder("test_model") + + start = time.monotonic() + res = encoder.embed_query("search query") + elapsed = time.monotonic() - start + + assert res == [0.1, 0.2, 0.3] + assert elapsed < slow / 2 + + # both backups are sent, they may still be starting up when the winner returns + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + with lock: + if len(calls) == 3: + break + time.sleep(0.01) + assert len(calls) == 3 + assert sum(name.startswith("embed-hedge") for name in calls) == 2 From 9b211b027fef91021f3eced0367878914df59be5 Mon Sep 17 00:00:00 2001 From: Tobias Macey Date: Tue, 1 Sep 2026 13:55:27 -0400 Subject: [PATCH 03/12] Move the shared media players into page-components (#3875) page-components/TiptapEditor embeds both VideoResourcePlayer and PodcastEmbedPlayer, but both lived in app-pages, which the import/no-restricted-paths zone in frontends/.eslintrc.js forbids page-components from importing. eslint-plugin-import 2.29 missed it; 2.31 (in #3851) reports all four. Moving only the two player entry points would have relocated the violation rather than fixed it, since each pulls in siblings, so each cluster moved with the modules it needs: page-components/VideoPlayer/ VideoResourcePlayer, VideoJsPlayer, YouTubeIframePlayer, videoSources page-components/PodcastEmbedPlayer/ PodcastEmbedPlayer, useAudioPlayer, usePlaybackRecovery, AudioPlayer.styled Only what the players need moved. The old shared.styled.ts keeps its seven page-chrome exports and gives up NoVideoMessage and ScreenReaderOnly; PodcastPlayer stays on the podcast page and now imports the hooks from page-components, which is an allowed direction. getEpisodeAudioUrl went to common/podcasts.ts instead of either player folder: the listing page and the embed player both need it, and common/ is already where this kind of pure helper lives. helpers.ts re-exports it so its existing importers and tests are untouched. Claude-Session: https://claude.ai/code/session_01FyJ72HRTyQf6jVYmw3GZSt Co-authored-by: Claude Opus 5 --- .../HomePage/VideoShortsModal.test.tsx | 2 +- .../app-pages/HomePage/VideoShortsModal.tsx | 2 +- .../app-pages/PodcastPage/PodcastPlayer.tsx | 12 ++++++--- .../PodcastsListingPage/helpers.ts | 24 +++-------------- .../VideoEmbedPage/VideoEmbedPage.test.tsx | 2 +- .../VideoEmbedPage/VideoEmbedPage.tsx | 2 +- .../VideoDetailPage.styled.ts | 4 +-- .../VideoDetailPage.test.tsx | 2 +- .../VideoDetailPage.tsx | 2 +- .../VideoSeriesDetailPage.styled.ts | 6 ++--- .../VideoSeriesDetailPage.test.tsx | 2 +- .../VideoSeriesDetailPage.tsx | 4 +-- .../VideoShareButton.test.tsx | 2 +- .../VideoShareButton.tsx | 2 +- .../shared.styled.ts | 23 ---------------- .../(embed)/podcast/embed/[id]/page.test.tsx | 2 +- .../app/(embed)/podcast/embed/[id]/page.tsx | 2 +- frontends/main/src/common/podcasts.ts | 26 +++++++++++++++++++ .../PodcastEmbedPlayer}/AudioPlayer.styled.ts | 0 .../PodcastEmbedPlayer.test.tsx | 0 .../PodcastEmbedPlayer.tsx | 2 +- .../PodcastEmbedPlayer}/useAudioPlayer.ts | 0 .../usePlaybackRecovery.ts | 0 .../node/MediaEmbed/MediaEmbedNodeView.tsx | 2 +- .../node/MediaEmbed/MediaEmbedViewer.tsx | 2 +- .../PodcastEpisodeEmbedNode.tsx | 2 +- .../PodcastEpisodeEmbedViewer.tsx | 2 +- .../VideoPlayer}/VideoJsPlayer.tsx | 0 .../VideoPlayer}/VideoResourcePlayer.test.tsx | 0 .../VideoPlayer}/VideoResourcePlayer.tsx | 0 .../VideoPlayer}/YouTubeIframePlayer.tsx | 0 .../VideoPlayer/shared.styled.ts | 24 +++++++++++++++++ .../VideoPlayer}/videoSources.test.ts | 0 .../VideoPlayer}/videoSources.ts | 0 34 files changed, 85 insertions(+), 70 deletions(-) create mode 100644 frontends/main/src/common/podcasts.ts rename frontends/main/src/{app-pages/PodcastPage => page-components/PodcastEmbedPlayer}/AudioPlayer.styled.ts (100%) rename frontends/main/src/{app-pages/PodcastPage => page-components/PodcastEmbedPlayer}/PodcastEmbedPlayer.test.tsx (100%) rename frontends/main/src/{app-pages/PodcastPage => page-components/PodcastEmbedPlayer}/PodcastEmbedPlayer.tsx (99%) rename frontends/main/src/{app-pages/PodcastPage => page-components/PodcastEmbedPlayer}/useAudioPlayer.ts (100%) rename frontends/main/src/{app-pages/PodcastPage => page-components/PodcastEmbedPlayer}/usePlaybackRecovery.ts (100%) rename frontends/main/src/{app-pages/VideoPlaylistCollectionPage => page-components/VideoPlayer}/VideoJsPlayer.tsx (100%) rename frontends/main/src/{app-pages/VideoPlaylistCollectionPage => page-components/VideoPlayer}/VideoResourcePlayer.test.tsx (100%) rename frontends/main/src/{app-pages/VideoPlaylistCollectionPage => page-components/VideoPlayer}/VideoResourcePlayer.tsx (100%) rename frontends/main/src/{app-pages/VideoPlaylistCollectionPage => page-components/VideoPlayer}/YouTubeIframePlayer.tsx (100%) create mode 100644 frontends/main/src/page-components/VideoPlayer/shared.styled.ts rename frontends/main/src/{app-pages/VideoPlaylistCollectionPage => page-components/VideoPlayer}/videoSources.test.ts (100%) rename frontends/main/src/{app-pages/VideoPlaylistCollectionPage => page-components/VideoPlayer}/videoSources.ts (100%) diff --git a/frontends/main/src/app-pages/HomePage/VideoShortsModal.test.tsx b/frontends/main/src/app-pages/HomePage/VideoShortsModal.test.tsx index a9588145ee..98133e687b 100644 --- a/frontends/main/src/app-pages/HomePage/VideoShortsModal.test.tsx +++ b/frontends/main/src/app-pages/HomePage/VideoShortsModal.test.tsx @@ -42,7 +42,7 @@ const makeMockPlayer = () => ({ const mockHandles: MockPlayerHandle[] = [] -jest.mock("@/app-pages/VideoPlaylistCollectionPage/VideoJsPlayer", () => ({ +jest.mock("@/page-components/VideoPlayer/VideoJsPlayer", () => ({ __esModule: true, default: ({ sources, diff --git a/frontends/main/src/app-pages/HomePage/VideoShortsModal.tsx b/frontends/main/src/app-pages/HomePage/VideoShortsModal.tsx index 5386d74ce8..0be030a911 100644 --- a/frontends/main/src/app-pages/HomePage/VideoShortsModal.tsx +++ b/frontends/main/src/app-pages/HomePage/VideoShortsModal.tsx @@ -13,7 +13,7 @@ import { ActionButton, VisuallyHidden } from "@mitodl/smoot-design" import { useWindowDimensions } from "ol-utilities" import type { VideoResource } from "api/v1" import MITOpenLearningLogo from "@/public/images/mit-open-learning-logo.svg" -import VideoJsPlayer from "@/app-pages/VideoPlaylistCollectionPage/VideoJsPlayer" +import VideoJsPlayer from "@/page-components/VideoPlayer/VideoJsPlayer" import type Player from "video.js/dist/types/player" import { FocusTrap } from "@mui/base/FocusTrap" import { usePostHog } from "posthog-js/react" diff --git a/frontends/main/src/app-pages/PodcastPage/PodcastPlayer.tsx b/frontends/main/src/app-pages/PodcastPage/PodcastPlayer.tsx index edaf96be9b..277f0a1b9d 100644 --- a/frontends/main/src/app-pages/PodcastPage/PodcastPlayer.tsx +++ b/frontends/main/src/app-pages/PodcastPage/PodcastPlayer.tsx @@ -11,8 +11,14 @@ import { RiCloseLine, RiErrorWarningLine, } from "@remixicon/react" -import { useAudioPlayer, formatClockTime } from "./useAudioPlayer" -import { usePlaybackRecovery, RETRYING_STATUS } from "./usePlaybackRecovery" +import { + useAudioPlayer, + formatClockTime, +} from "@/page-components/PodcastEmbedPlayer/useAudioPlayer" +import { + usePlaybackRecovery, + RETRYING_STATUS, +} from "@/page-components/PodcastEmbedPlayer/usePlaybackRecovery" import { TrackInfo as TrackInfoBase, TrackTitle as TrackTitleBase, @@ -28,7 +34,7 @@ import { PlaybackError, PlaybackErrorText, RetryButton, -} from "./AudioPlayer.styled" +} from "@/page-components/PodcastEmbedPlayer/AudioPlayer.styled" // ─── Types ──────────────────────────────────────────────────────────────────── diff --git a/frontends/main/src/app-pages/PodcastPage/PodcastsListingPage/helpers.ts b/frontends/main/src/app-pages/PodcastPage/PodcastsListingPage/helpers.ts index a689791671..d888051b74 100644 --- a/frontends/main/src/app-pages/PodcastPage/PodcastsListingPage/helpers.ts +++ b/frontends/main/src/app-pages/PodcastPage/PodcastsListingPage/helpers.ts @@ -2,29 +2,13 @@ import moment from "moment" import { ResourceTypeEnum } from "api/v1" import type { LearningResource, PodcastEpisodeParent } from "api/v1" +// Defined in common/ so page-components can use it too; re-exported here to +// keep this module the single helper import for the podcast pages. +export { getEpisodeAudioUrl } from "@/common/podcasts" + export const formatApproxCount = (count: number): string => count >= 100 ? `${Math.floor(count / 100) * 100}+` : String(count) -/** - * The URL to play/link for an episode. - * - * Defaults to the direct `audio_url`, falling back to `episode_link`. Pass - * `{ allowEpisodeLink: false }` when the URL is fed straight into an `