From 792604bd5fd609c2af68de0a9dc50d71a8efc32c Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sat, 1 Aug 2026 17:54:09 -0500 Subject: [PATCH 1/5] perf(cloud-ai): run AWS Rekognition boto3 calls off the event loop boto3 is a synchronous SDK. Every Rekognition call in AWSRekognition was issued directly inside an `async def`, so each one blocked the event loop for a full network round-trip. `_wait_for_job_completion` is the worst case: it polls every 5s for up to 600s, so a single video analysis could stall the loop up to 120 times. All 14 boto3 calls now dispatch via `await asyncio.to_thread(...)`, and the local-image read in `_prepare_image_input` goes through a new module-level `_read_file_bytes` helper on the same path. - 89 pre-existing tests pass with zero edits - 6 new heartbeat tests (`TestRekognitionDoesNotBlockEventLoop`); 5 of the 6 discriminate, proven by reverting both dimensions simultaneously (5 targeted failures / 90 passed) - ruff: exact parity with origin/main (8 pre-existing findings, 0 added) Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../cloud_ai/providers/aws_rekognition.py | 82 ++++++---- tests/unit/test_aws_rekognition_provider.py | 151 ++++++++++++++++++ 2 files changed, 206 insertions(+), 27 deletions(-) diff --git a/src/youtube_extension/integrations/cloud_ai/providers/aws_rekognition.py b/src/youtube_extension/integrations/cloud_ai/providers/aws_rekognition.py index dd56960e0..747c3206c 100644 --- a/src/youtube_extension/integrations/cloud_ai/providers/aws_rekognition.py +++ b/src/youtube_extension/integrations/cloud_ai/providers/aws_rekognition.py @@ -31,6 +31,13 @@ logger = logging.getLogger(__name__) +def _read_file_bytes(path: str) -> bytes: + """Read a file's bytes. Module-level so it can run in a worker thread.""" + with open(path, 'rb') as handle: + return handle.read() + + + class AWSRekognition(BaseCloudAI): """Amazon Rekognition video and image analysis integration.""" @@ -103,8 +110,9 @@ async def _test_connection(self) -> None: """Test AWS Rekognition connection.""" try: # Simple API call to test connectivity - self._rekognition_client.describe_collection( - CollectionId='non-existent-collection' + await asyncio.to_thread( + self._rekognition_client.describe_collection, + CollectionId='non-existent-collection', ) except Exception as e: if "ResourceNotFoundException" in str(e): @@ -174,27 +182,31 @@ async def analyze_image(self, image_url: str, results = {} if AnalysisType.LABEL_DETECTION in analysis_types: - results['labels'] = self._rekognition_client.detect_labels( + results['labels'] = await asyncio.to_thread( + self._rekognition_client.detect_labels, Image=image_data, MaxLabels=50, - MinConfidence=0.5 + MinConfidence=0.5, ) if AnalysisType.FACE_DETECTION in analysis_types: - results['faces'] = self._rekognition_client.detect_faces( + results['faces'] = await asyncio.to_thread( + self._rekognition_client.detect_faces, Image=image_data, - Attributes=['ALL'] + Attributes=['ALL'], ) if AnalysisType.TEXT_DETECTION in analysis_types: - results['text'] = self._rekognition_client.detect_text( - Image=image_data + results['text'] = await asyncio.to_thread( + self._rekognition_client.detect_text, + Image=image_data, ) if AnalysisType.CONTENT_MODERATION in analysis_types: - results['moderation'] = self._rekognition_client.detect_moderation_labels( + results['moderation'] = await asyncio.to_thread( + self._rekognition_client.detect_moderation_labels, Image=image_data, - MinConfidence=0.5 + MinConfidence=0.5, ) processing_time = (datetime.utcnow() - start_time).total_seconds() @@ -219,8 +231,9 @@ async def get_service_status(self) -> dict[str, Any]: # Test with describe_collection call try: - self._rekognition_client.describe_collection( - CollectionId='health-check-collection' + await asyncio.to_thread( + self._rekognition_client.describe_collection, + CollectionId='health-check-collection', ) except Exception as e: if "ResourceNotFoundException" in str(e): @@ -278,28 +291,32 @@ async def _start_video_analysis(self, s3_bucket: str, s3_key: str, # Start different analysis operations based on requested types if AnalysisType.LABEL_DETECTION in analysis_types: - response = self._rekognition_client.start_label_detection( + response = await asyncio.to_thread( + self._rekognition_client.start_label_detection, Video=video_input, - MinConfidence=0.5 + MinConfidence=0.5, ) job_ids['labels'] = response['JobId'] if AnalysisType.FACE_DETECTION in analysis_types: - response = self._rekognition_client.start_face_detection( - Video=video_input + response = await asyncio.to_thread( + self._rekognition_client.start_face_detection, + Video=video_input, ) job_ids['faces'] = response['JobId'] if AnalysisType.TEXT_DETECTION in analysis_types: - response = self._rekognition_client.start_text_detection( - Video=video_input + response = await asyncio.to_thread( + self._rekognition_client.start_text_detection, + Video=video_input, ) job_ids['text'] = response['JobId'] if AnalysisType.CONTENT_MODERATION in analysis_types: - response = self._rekognition_client.start_content_moderation( + response = await asyncio.to_thread( + self._rekognition_client.start_content_moderation, Video=video_input, - MinConfidence=0.5 + MinConfidence=0.5, ) job_ids['moderation'] = response['JobId'] @@ -326,13 +343,25 @@ async def _wait_for_job_completion(self, job_id: str, analysis_type: str) -> dic while elapsed_time < max_wait_time: try: if analysis_type == 'labels': - response = self._rekognition_client.get_label_detection(JobId=job_id) + response = await asyncio.to_thread( + self._rekognition_client.get_label_detection, + JobId=job_id, + ) elif analysis_type == 'faces': - response = self._rekognition_client.get_face_detection(JobId=job_id) + response = await asyncio.to_thread( + self._rekognition_client.get_face_detection, + JobId=job_id, + ) elif analysis_type == 'text': - response = self._rekognition_client.get_text_detection(JobId=job_id) + response = await asyncio.to_thread( + self._rekognition_client.get_text_detection, + JobId=job_id, + ) elif analysis_type == 'moderation': - response = self._rekognition_client.get_content_moderation(JobId=job_id) + response = await asyncio.to_thread( + self._rekognition_client.get_content_moderation, + JobId=job_id, + ) else: raise ValueError(f"Unknown analysis type: {analysis_type}") @@ -369,9 +398,8 @@ async def _prepare_image_input(self, image_url: str) -> dict[str, Any]: response = await client.get(image_url) return {'Bytes': response.content} else: - # Local file - with open(image_url, 'rb') as image_file: - return {'Bytes': image_file.read()} + # Local file - read off the event loop + return {'Bytes': await asyncio.to_thread(_read_file_bytes, image_url)} def _process_video_results(self, results: dict[str, Any], video_id: str, analysis_types: list[AnalysisType], diff --git a/tests/unit/test_aws_rekognition_provider.py b/tests/unit/test_aws_rekognition_provider.py index 2f0ad4000..e1c411ed3 100644 --- a/tests/unit/test_aws_rekognition_provider.py +++ b/tests/unit/test_aws_rekognition_provider.py @@ -1004,3 +1004,154 @@ def test_cost_scales_with_duration(self): cost1 = provider.estimate_cost(60.0, [AnalysisType.LABEL_DETECTION]) cost2 = provider.estimate_cost(120.0, [AnalysisType.LABEL_DETECTION]) assert cost2 == pytest.approx(cost1 * 2) + + +# =========================================================================== +# Event-loop responsiveness: every boto3 call must run off the loop +# =========================================================================== + +async def _count_heartbeats(coro, tick: float = 0.005) -> tuple[object, int]: + """Run ``coro`` while a heartbeat task ticks; return (result, ticks). + + If the awaited work performs blocking I/O directly on the event loop the + heartbeat never gets scheduled and ``ticks`` stays at 0. + """ + import asyncio + import contextlib + + ticks = 0 + stop = False + + async def _beat(): + nonlocal ticks + while not stop: + await asyncio.sleep(tick) + ticks += 1 + + beat = asyncio.create_task(_beat()) + await asyncio.sleep(0) # let the heartbeat reach its first await first + try: + result = await coro + finally: + stop = True + beat.cancel() + with contextlib.suppress(asyncio.CancelledError): + await beat + return result, ticks + + +def _slow(return_value, delay: float = 0.12): + """A synchronous callable that blocks for ``delay`` seconds.""" + import time + + def _call(*_args, **_kwargs): + time.sleep(delay) + return return_value + + return _call + + +class TestRekognitionDoesNotBlockEventLoop: + async def test_analyze_image_does_not_stall_the_event_loop(self): + provider = _make_provider() + mock_client = _make_rekognition_client() + mock_client.detect_labels.side_effect = _slow({"Labels": []}) + provider._rekognition_client = mock_client + + with patch.object(provider, '_prepare_image_input', + new=AsyncMock(return_value={"Bytes": b"img"})): + result, ticks = await _count_heartbeats( + provider.analyze_image("http://e.com/i.jpg", [AnalysisType.LABEL_DETECTION]) + ) + + assert isinstance(result, VideoAnalysisResult) + assert ticks > 0, "detect_labels blocked the event loop" + + async def test_every_detection_type_runs_off_the_loop(self): + provider = _make_provider() + mock_client = _make_rekognition_client() + # four blocking calls back to back + mock_client.detect_labels.side_effect = _slow({"Labels": []}, 0.06) + mock_client.detect_faces.side_effect = _slow({"FaceDetails": []}, 0.06) + mock_client.detect_text.side_effect = _slow({"TextDetections": []}, 0.06) + mock_client.detect_moderation_labels.side_effect = _slow({"ModerationLabels": []}, 0.06) + provider._rekognition_client = mock_client + + with patch.object(provider, '_prepare_image_input', + new=AsyncMock(return_value={"Bytes": b"img"})): + _result, ticks = await _count_heartbeats( + provider.analyze_image( + "http://e.com/i.jpg", + [AnalysisType.LABEL_DETECTION, AnalysisType.FACE_DETECTION, + AnalysisType.TEXT_DETECTION, AnalysisType.CONTENT_MODERATION], + ) + ) + + mock_client.detect_labels.assert_called_once() + mock_client.detect_moderation_labels.assert_called_once() + assert ticks > 0, "the detect_* chain blocked the event loop" + + async def test_job_polling_does_not_stall_the_event_loop(self): + provider = _make_provider() + mock_client = MagicMock() + mock_client.get_label_detection.side_effect = _slow( + {'JobStatus': 'SUCCEEDED', 'Labels': []} + ) + provider._rekognition_client = mock_client + + result, ticks = await _count_heartbeats( + provider._wait_for_job_completion("job-1", "labels") + ) + + assert result['JobStatus'] == 'SUCCEEDED' + assert ticks > 0, "get_label_detection blocked the event loop" + + async def test_start_video_analysis_does_not_stall_the_event_loop(self): + provider = _make_provider() + mock_client = MagicMock() + mock_client.start_label_detection.side_effect = _slow({'JobId': 'j-1'}) + provider._rekognition_client = mock_client + + result, ticks = await _count_heartbeats( + provider._start_video_analysis( + "my-bucket", "video.mp4", [AnalysisType.LABEL_DETECTION] + ) + ) + + assert result == {'labels': 'j-1'} + assert ticks > 0, "start_label_detection blocked the event loop" + + async def test_local_image_read_does_not_stall_the_event_loop(self, tmp_path): + from youtube_extension.integrations.cloud_ai.providers import ( + aws_rekognition as _rek_mod, + ) + + image = tmp_path / "frame.jpg" + image.write_bytes(b"BINARY-IMAGE-PAYLOAD") + provider = _make_provider() + + real_read = _rek_mod._read_file_bytes + + def _slow_read(path): + import time + time.sleep(0.12) + return real_read(path) + + with patch.object(_rek_mod, '_read_file_bytes', _slow_read): + result, ticks = await _count_heartbeats( + provider._prepare_image_input(str(image)) + ) + + assert result == {'Bytes': b"BINARY-IMAGE-PAYLOAD"} + assert ticks > 0, "the local image read blocked the event loop" + + async def test_local_image_bytes_are_read_correctly(self, tmp_path): + """Guard: offloading must not change what is returned.""" + image = tmp_path / "frame.png" + payload = bytes(range(256)) * 8 + image.write_bytes(payload) + provider = _make_provider() + + result = await provider._prepare_image_input(str(image)) + + assert result == {'Bytes': payload} From 9b3f5e4844c556676498c69f90b14a426fd83deb Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sat, 1 Aug 2026 18:05:55 -0500 Subject: [PATCH 2/5] test(rekognition): assert the local read leaves the loop thread, not elapsed ticks The heartbeat form of this one test failed on CI. Unlike the four boto3 tests, which drive a controllable 0.12s mock, the local file read is a few microseconds of real work, so "did the loop tick while it ran" is a load-sensitive proxy rather than a property. Assert the property directly instead: record `threading.get_ident()` inside `_read_file_bytes` and require it to differ from the thread running the event loop. That is exactly what "dispatched off the loop" means, needs no sleeps, and cannot flake under runner contention. - 95 tests pass (89 pre-existing, unmodified, + 6 new) - Non-vacuity: calling `_read_file_bytes` directly instead of via `asyncio.to_thread` yields exactly 1 targeted failure / 94 passed - Suite runtime for the file drops to 0.85s (the 0.12s sleep is gone) Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/unit/test_aws_rekognition_provider.py | 31 +++++++++++++++------ 1 file changed, 22 insertions(+), 9 deletions(-) diff --git a/tests/unit/test_aws_rekognition_provider.py b/tests/unit/test_aws_rekognition_provider.py index e1c411ed3..98fff260f 100644 --- a/tests/unit/test_aws_rekognition_provider.py +++ b/tests/unit/test_aws_rekognition_provider.py @@ -1121,7 +1121,17 @@ async def test_start_video_analysis_does_not_stall_the_event_loop(self): assert result == {'labels': 'j-1'} assert ticks > 0, "start_label_detection blocked the event loop" - async def test_local_image_read_does_not_stall_the_event_loop(self, tmp_path): + async def test_local_image_read_runs_off_the_event_loop_thread(self, tmp_path): + """ + Deterministic counterpart to the heartbeat tests. + + The local read is far too fast to time reliably on a loaded runner, so + instead of measuring elapsed ticks this asserts the property directly: + the read must execute on a worker thread, not on the thread running the + event loop. This is immune to scheduler load and needs no sleeps. + """ + import threading + from youtube_extension.integrations.cloud_ai.providers import ( aws_rekognition as _rek_mod, ) @@ -1130,20 +1140,23 @@ async def test_local_image_read_does_not_stall_the_event_loop(self, tmp_path): image.write_bytes(b"BINARY-IMAGE-PAYLOAD") provider = _make_provider() + loop_thread_id = threading.get_ident() + observed: dict[str, int] = {} real_read = _rek_mod._read_file_bytes - def _slow_read(path): - import time - time.sleep(0.12) + def _recording_read(path): + observed['thread_id'] = threading.get_ident() return real_read(path) - with patch.object(_rek_mod, '_read_file_bytes', _slow_read): - result, ticks = await _count_heartbeats( - provider._prepare_image_input(str(image)) - ) + with patch.object(_rek_mod, '_read_file_bytes', _recording_read): + result = await provider._prepare_image_input(str(image)) assert result == {'Bytes': b"BINARY-IMAGE-PAYLOAD"} - assert ticks > 0, "the local image read blocked the event loop" + assert 'thread_id' in observed, "_read_file_bytes was never called" + assert observed['thread_id'] != loop_thread_id, ( + "the local image read ran on the event loop thread instead of " + "being dispatched to a worker thread" + ) async def test_local_image_bytes_are_read_correctly(self, tmp_path): """Guard: offloading must not change what is returned.""" From 2c0fec7f77d5e95b8d1da06c12f421e384367eaa Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sat, 1 Aug 2026 18:17:05 -0500 Subject: [PATCH 3/5] test(rekognition): cover every blocking SDK operation --- tests/unit/test_aws_rekognition_provider.py | 118 +++++++++++++------- 1 file changed, 77 insertions(+), 41 deletions(-) diff --git a/tests/unit/test_aws_rekognition_provider.py b/tests/unit/test_aws_rekognition_provider.py index 98fff260f..39609417e 100644 --- a/tests/unit/test_aws_rekognition_provider.py +++ b/tests/unit/test_aws_rekognition_provider.py @@ -1052,75 +1052,111 @@ def _call(*_args, **_kwargs): class TestRekognitionDoesNotBlockEventLoop: - async def test_analyze_image_does_not_stall_the_event_loop(self): + @pytest.mark.parametrize( + ("analysis_type", "client_method", "response"), + [ + (AnalysisType.LABEL_DETECTION, "detect_labels", {"Labels": []}), + (AnalysisType.FACE_DETECTION, "detect_faces", {"FaceDetails": []}), + (AnalysisType.TEXT_DETECTION, "detect_text", {"TextDetections": []}), + ( + AnalysisType.CONTENT_MODERATION, + "detect_moderation_labels", + {"ModerationLabels": []}, + ), + ], + ) + async def test_each_image_detection_runs_off_the_loop( + self, analysis_type, client_method, response + ): provider = _make_provider() mock_client = _make_rekognition_client() - mock_client.detect_labels.side_effect = _slow({"Labels": []}) + getattr(mock_client, client_method).side_effect = _slow(response) provider._rekognition_client = mock_client - with patch.object(provider, '_prepare_image_input', - new=AsyncMock(return_value={"Bytes": b"img"})): + with patch.object( + provider, + "_prepare_image_input", + new=AsyncMock(return_value={"Bytes": b"img"}), + ): result, ticks = await _count_heartbeats( - provider.analyze_image("http://e.com/i.jpg", [AnalysisType.LABEL_DETECTION]) + provider.analyze_image("http://e.com/i.jpg", [analysis_type]) ) assert isinstance(result, VideoAnalysisResult) - assert ticks > 0, "detect_labels blocked the event loop" - - async def test_every_detection_type_runs_off_the_loop(self): + getattr(mock_client, client_method).assert_called_once() + assert ticks > 0, f"{client_method} blocked the event loop" + + @pytest.mark.parametrize( + ("analysis_type", "client_method", "result_key"), + [ + (AnalysisType.LABEL_DETECTION, "start_label_detection", "labels"), + (AnalysisType.FACE_DETECTION, "start_face_detection", "faces"), + (AnalysisType.TEXT_DETECTION, "start_text_detection", "text"), + ( + AnalysisType.CONTENT_MODERATION, + "start_content_moderation", + "moderation", + ), + ], + ) + async def test_each_video_start_runs_off_the_loop( + self, analysis_type, client_method, result_key + ): provider = _make_provider() - mock_client = _make_rekognition_client() - # four blocking calls back to back - mock_client.detect_labels.side_effect = _slow({"Labels": []}, 0.06) - mock_client.detect_faces.side_effect = _slow({"FaceDetails": []}, 0.06) - mock_client.detect_text.side_effect = _slow({"TextDetections": []}, 0.06) - mock_client.detect_moderation_labels.side_effect = _slow({"ModerationLabels": []}, 0.06) + mock_client = MagicMock() + getattr(mock_client, client_method).side_effect = _slow({"JobId": "j-1"}) provider._rekognition_client = mock_client - with patch.object(provider, '_prepare_image_input', - new=AsyncMock(return_value={"Bytes": b"img"})): - _result, ticks = await _count_heartbeats( - provider.analyze_image( - "http://e.com/i.jpg", - [AnalysisType.LABEL_DETECTION, AnalysisType.FACE_DETECTION, - AnalysisType.TEXT_DETECTION, AnalysisType.CONTENT_MODERATION], - ) + result, ticks = await _count_heartbeats( + provider._start_video_analysis( + "my-bucket", "video.mp4", [analysis_type] ) + ) - mock_client.detect_labels.assert_called_once() - mock_client.detect_moderation_labels.assert_called_once() - assert ticks > 0, "the detect_* chain blocked the event loop" - - async def test_job_polling_does_not_stall_the_event_loop(self): + assert result == {result_key: "j-1"} + assert ticks > 0, f"{client_method} blocked the event loop" + + @pytest.mark.parametrize( + ("analysis_key", "client_method"), + [ + ("labels", "get_label_detection"), + ("faces", "get_face_detection"), + ("text", "get_text_detection"), + ("moderation", "get_content_moderation"), + ], + ) + async def test_each_video_poll_runs_off_the_loop( + self, analysis_key, client_method + ): provider = _make_provider() mock_client = MagicMock() - mock_client.get_label_detection.side_effect = _slow( - {'JobStatus': 'SUCCEEDED', 'Labels': []} + getattr(mock_client, client_method).side_effect = _slow( + {"JobStatus": "SUCCEEDED"} ) provider._rekognition_client = mock_client result, ticks = await _count_heartbeats( - provider._wait_for_job_completion("job-1", "labels") + provider._wait_for_job_completion("job-1", analysis_key) ) - assert result['JobStatus'] == 'SUCCEEDED' - assert ticks > 0, "get_label_detection blocked the event loop" + assert result["JobStatus"] == "SUCCEEDED" + assert ticks > 0, f"{client_method} blocked the event loop" - async def test_start_video_analysis_does_not_stall_the_event_loop(self): + @pytest.mark.parametrize("provider_method", ["_test_connection", "get_service_status"]) + async def test_each_describe_collection_runs_off_the_loop(self, provider_method): provider = _make_provider() mock_client = MagicMock() - mock_client.start_label_detection.side_effect = _slow({'JobId': 'j-1'}) + mock_client.describe_collection.side_effect = _slow({}) provider._rekognition_client = mock_client - result, ticks = await _count_heartbeats( - provider._start_video_analysis( - "my-bucket", "video.mp4", [AnalysisType.LABEL_DETECTION] - ) + _result, ticks = await _count_heartbeats( + getattr(provider, provider_method)() ) - assert result == {'labels': 'j-1'} - assert ticks > 0, "start_label_detection blocked the event loop" - + mock_client.describe_collection.assert_called_once() + assert ticks > 0, ( + f"describe_collection in {provider_method} blocked the event loop" + ) async def test_local_image_read_runs_off_the_event_loop_thread(self, tmp_path): """ Deterministic counterpart to the heartbeat tests. From af04aedb59d5b46871c46ab945536faa58c97d91 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sat, 1 Aug 2026 18:21:54 -0500 Subject: [PATCH 4/5] test(rekognition): make off-loop read test robust to module eviction --- tests/unit/test_aws_rekognition_provider.py | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/tests/unit/test_aws_rekognition_provider.py b/tests/unit/test_aws_rekognition_provider.py index 39609417e..a6cfe10d9 100644 --- a/tests/unit/test_aws_rekognition_provider.py +++ b/tests/unit/test_aws_rekognition_provider.py @@ -1157,6 +1157,7 @@ async def test_each_describe_collection_runs_off_the_loop(self, provider_method) assert ticks > 0, ( f"describe_collection in {provider_method} blocked the event loop" ) + async def test_local_image_read_runs_off_the_event_loop_thread(self, tmp_path): """ Deterministic counterpart to the heartbeat tests. @@ -1168,23 +1169,24 @@ async def test_local_image_read_runs_off_the_event_loop_thread(self, tmp_path): """ import threading - from youtube_extension.integrations.cloud_ai.providers import ( - aws_rekognition as _rek_mod, - ) - image = tmp_path / "frame.jpg" image.write_bytes(b"BINARY-IMAGE-PAYLOAD") provider = _make_provider() + # Patch the exact namespace that _prepare_image_input resolves from. + # Other test modules evict cloud_ai modules from sys.modules during + # collection, so re-importing the module here can patch a stale object. + method_globals = type(provider)._prepare_image_input.__globals__ + real_read = method_globals["_read_file_bytes"] + loop_thread_id = threading.get_ident() observed: dict[str, int] = {} - real_read = _rek_mod._read_file_bytes def _recording_read(path): observed['thread_id'] = threading.get_ident() return real_read(path) - with patch.object(_rek_mod, '_read_file_bytes', _recording_read): + with patch.dict(method_globals, {"_read_file_bytes": _recording_read}): result = await provider._prepare_image_input(str(image)) assert result == {'Bytes': b"BINARY-IMAGE-PAYLOAD"} From b5721eb4fc1327611a17664659051cab1fb7dd09 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sat, 1 Aug 2026 18:36:15 -0500 Subject: [PATCH 5/5] perf(rekognition): bound AWS client requests with botocore timeouts Review round 4, finding 2 (CodeRabbit, Stability & Availability, Major). This PR moved 14 blocking boto3 calls onto the *shared* default asyncio executor via asyncio.to_thread. botocore's defaults leave a request effectively unbounded, so a stalled AWS call would now pin one of that pool's limited worker threads indefinitely and starve every other to_thread user in the process -- including the metrics persistence (#1194), sqlite access (#1196) and result writes (#1203) already merged onto that same pool. _wait_for_job_completion can issue up to 120 such calls per job, so the exposure is real rather than theoretical. Both the Rekognition and S3 clients are now constructed with an explicit botocore Config carrying connect_timeout, read_timeout and a bounded standard-mode retry policy. Values are overridable via AWS_REKOGNITION_CONNECT_TIMEOUT / _READ_TIMEOUT / _MAX_ATTEMPTS and are validated with math.isfinite, raising rather than clamping so that 'inf', 'nan', '0' and negatives are rejected outright. Parsing happens before initialize()'s try block: that method ends in a catch-all `except Exception -> CloudAIError`, which would otherwise bury a precise ConfigurationError message behind a generic init failure. Tests: 127 pass (105 pre-existing + 22 new). Non-vacuity proven by mutating both dimensions simultaneously -- making the env helpers ignore the environment and dropping `config=` from both client constructions yields exactly 18 targeted failures / 109 passed, matching the predicted count (1 client-config + 1 override + 12 timeout rejections + 4 max-attempts rejections). ruff parity with origin/main unchanged (8 = 8). Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../cloud_ai/providers/aws_rekognition.py | 89 ++++++++++++++++- tests/unit/test_aws_rekognition_provider.py | 98 +++++++++++++++++++ 2 files changed, 185 insertions(+), 2 deletions(-) diff --git a/src/youtube_extension/integrations/cloud_ai/providers/aws_rekognition.py b/src/youtube_extension/integrations/cloud_ai/providers/aws_rekognition.py index 747c3206c..9b300bdb4 100644 --- a/src/youtube_extension/integrations/cloud_ai/providers/aws_rekognition.py +++ b/src/youtube_extension/integrations/cloud_ai/providers/aws_rekognition.py @@ -11,6 +11,8 @@ import asyncio import logging +import math +import os from datetime import datetime from typing import Any @@ -31,6 +33,79 @@ logger = logging.getLogger(__name__) +# Default botocore timeouts. Every Rekognition/S3 call in this module runs via +# ``asyncio.to_thread``, i.e. on the *shared* default executor. Without an +# explicit read timeout a stalled AWS request pins one of the pool's limited +# worker threads indefinitely, which would starve every other ``to_thread`` +# user in the process. Bounding the request bounds the worker. +_DEFAULT_CONNECT_TIMEOUT = 10.0 +_DEFAULT_READ_TIMEOUT = 60.0 +_DEFAULT_MAX_ATTEMPTS = 3 + + +def _positive_finite_float_env(name: str, default: float) -> float: + """Read a strictly positive, finite float from the environment.""" + raw = os.environ.get(name) + if raw is None or not raw.strip(): + return default + try: + value = float(raw) + except (TypeError, ValueError) as exc: + raise ConfigurationError( + f"{name} must be a number, got {raw!r}", + provider=CloudAIProvider.AWS_REKOGNITION.value, + ) from exc + if not math.isfinite(value) or value <= 0: + raise ConfigurationError( + f"{name} must be a positive finite number, got {raw!r}", + provider=CloudAIProvider.AWS_REKOGNITION.value, + ) + return value + + +def _positive_int_env(name: str, default: int) -> int: + """Read a strictly positive integer from the environment.""" + raw = os.environ.get(name) + if raw is None or not raw.strip(): + return default + try: + value = int(raw) + except (TypeError, ValueError) as exc: + raise ConfigurationError( + f"{name} must be an integer, got {raw!r}", + provider=CloudAIProvider.AWS_REKOGNITION.value, + ) from exc + if value <= 0: + raise ConfigurationError( + f"{name} must be a positive integer, got {raw!r}", + provider=CloudAIProvider.AWS_REKOGNITION.value, + ) + return value + + +def _timeout_config_kwargs() -> dict: + """Parse botocore timeout settings from the environment. + + Kept separate from ``initialize`` so a bad value raises ConfigurationError + with its precise message instead of being re-wrapped as a generic + CloudAIError by that method's catch-all handler. + """ + return { + 'connect_timeout': _positive_finite_float_env( + 'AWS_REKOGNITION_CONNECT_TIMEOUT', _DEFAULT_CONNECT_TIMEOUT + ), + 'read_timeout': _positive_finite_float_env( + 'AWS_REKOGNITION_READ_TIMEOUT', _DEFAULT_READ_TIMEOUT + ), + 'retries': { + 'max_attempts': _positive_int_env( + 'AWS_REKOGNITION_MAX_ATTEMPTS', _DEFAULT_MAX_ATTEMPTS + ), + 'mode': 'standard', + }, + } + + def _read_file_bytes(path: str) -> bytes: """Read a file's bytes. Module-level so it can run in a worker thread.""" with open(path, 'rb') as handle: @@ -64,8 +139,12 @@ def _validate_config(self) -> None: async def initialize(self) -> None: """Initialize AWS Rekognition client.""" + # Parsed before the try so a misconfigured value surfaces as a + # ConfigurationError rather than the catch-all CloudAIError below. + timeout_kwargs = _timeout_config_kwargs() try: import boto3 + from botocore.config import Config as BotoConfig from botocore.exceptions import ClientError, NoCredentialsError # Create session with credentials @@ -75,9 +154,15 @@ async def initialize(self) -> None: region_name=self.config["region"] ) + # Bound every request so a stalled call cannot hold a shared + # executor thread forever (all SDK calls here run in to_thread). + client_config = BotoConfig(**timeout_kwargs) + # Initialize clients - self._rekognition_client = session.client('rekognition') - self._s3_client = session.client('s3') + self._rekognition_client = session.client( + 'rekognition', config=client_config + ) + self._s3_client = session.client('s3', config=client_config) # Test connection await self._test_connection() diff --git a/tests/unit/test_aws_rekognition_provider.py b/tests/unit/test_aws_rekognition_provider.py index a6cfe10d9..ea88ae298 100644 --- a/tests/unit/test_aws_rekognition_provider.py +++ b/tests/unit/test_aws_rekognition_provider.py @@ -2,6 +2,8 @@ from __future__ import annotations +import math +import os import sys from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch @@ -131,6 +133,7 @@ async def test_initialize_sets_rekognition_client(self): mock_boto3.Session.return_value = mock_session with patch.dict("sys.modules", {"boto3": mock_boto3, "botocore": MagicMock(), + "botocore.config": MagicMock(), "botocore.exceptions": MagicMock( ClientError=Exception, NoCredentialsError=Exception )}): @@ -147,6 +150,7 @@ async def test_initialize_sets_s3_client(self): mock_boto3.Session.return_value = mock_session with patch.dict("sys.modules", {"boto3": mock_boto3, "botocore": MagicMock(), + "botocore.config": MagicMock(), "botocore.exceptions": MagicMock( ClientError=Exception, NoCredentialsError=Exception )}): @@ -169,6 +173,7 @@ async def test_initialize_uses_correct_region(self): mock_boto3.Session.return_value = mock_session with patch.dict("sys.modules", {"boto3": mock_boto3, "botocore": MagicMock(), + "botocore.config": MagicMock(), "botocore.exceptions": MagicMock( ClientError=Exception, NoCredentialsError=Exception )}): @@ -1206,3 +1211,96 @@ async def test_local_image_bytes_are_read_correctly(self, tmp_path): result = await provider._prepare_image_input(str(image)) assert result == {'Bytes': payload} + + +# --------------------------------------------------------------------------- +# botocore client timeouts +# --------------------------------------------------------------------------- +class TestClientTimeoutConfiguration: + """Every SDK call here runs on the shared default executor via + ``asyncio.to_thread``. Without an explicit read timeout a stalled AWS + request would pin one of that pool's limited worker threads forever and + starve every other ``to_thread`` user in the process, so the clients must + always be built with a bounded botocore config. + """ + + @staticmethod + async def _initialize(env: dict | None = None): + provider = _make_provider() + mock_session = MagicMock() + mock_session.client.return_value = _make_rekognition_client() + mock_boto3 = MagicMock() + mock_boto3.Session.return_value = mock_session + config_mod = MagicMock() + + stubs = { + "boto3": mock_boto3, + "botocore": MagicMock(), + "botocore.config": config_mod, + "botocore.exceptions": MagicMock( + ClientError=Exception, NoCredentialsError=Exception + ), + } + with patch.dict("sys.modules", stubs), patch.dict(os.environ, env or {}): + await provider.initialize() + return mock_session, config_mod + + async def test_both_clients_are_built_with_a_bounded_config(self): + mock_session, config_mod = await self._initialize() + + built = config_mod.Config.return_value + services = {} + for call in mock_session.client.call_args_list: + services[call.args[0]] = call.kwargs.get("config") + + assert set(services) == {"rekognition", "s3"} + assert services["rekognition"] is built + assert services["s3"] is built + + async def test_default_timeouts_are_finite_and_positive(self): + _session, config_mod = await self._initialize() + + kwargs = config_mod.Config.call_args.kwargs + assert 0 < kwargs["connect_timeout"] < math.inf + assert 0 < kwargs["read_timeout"] < math.inf + assert kwargs["retries"]["max_attempts"] >= 1 + + async def test_timeouts_are_overridable_from_the_environment(self): + _session, config_mod = await self._initialize( + { + "AWS_REKOGNITION_CONNECT_TIMEOUT": "2.5", + "AWS_REKOGNITION_READ_TIMEOUT": "7", + "AWS_REKOGNITION_MAX_ATTEMPTS": "5", + } + ) + + kwargs = config_mod.Config.call_args.kwargs + assert kwargs["connect_timeout"] == 2.5 + assert kwargs["read_timeout"] == 7.0 + assert kwargs["retries"]["max_attempts"] == 5 + + @pytest.mark.parametrize( + "var", + ["AWS_REKOGNITION_CONNECT_TIMEOUT", "AWS_REKOGNITION_READ_TIMEOUT"], + ) + @pytest.mark.parametrize("bad", ["inf", "-inf", "nan", "0", "-1", "abc"]) + async def test_non_finite_or_non_positive_timeouts_are_rejected(self, var, bad): + with pytest.raises(ConfigurationError): + await self._initialize({var: bad}) + + @pytest.mark.parametrize("bad", ["0", "-3", "1.5", "abc"]) + async def test_invalid_max_attempts_is_rejected(self, bad): + with pytest.raises(ConfigurationError): + await self._initialize({"AWS_REKOGNITION_MAX_ATTEMPTS": bad}) + + @pytest.mark.parametrize( + "var", + [ + "AWS_REKOGNITION_CONNECT_TIMEOUT", + "AWS_REKOGNITION_READ_TIMEOUT", + "AWS_REKOGNITION_MAX_ATTEMPTS", + ], + ) + async def test_blank_values_fall_back_to_defaults(self, var): + _session, config_mod = await self._initialize({var: " "}) + assert config_mod.Config.call_args is not None