diff --git a/.github/workflows/fastapi-test.yml b/.github/workflows/fastapi-test.yml index e3242152..417d90b4 100644 --- a/.github/workflows/fastapi-test.yml +++ b/.github/workflows/fastapi-test.yml @@ -49,7 +49,7 @@ jobs: run: hatch run lint:format-check - name: Install pnpm - uses: pnpm/action-setup@008330803749db0355799c700092d9a85fd074e9 + uses: pnpm/action-setup@008330803749db0355799c700092d9a85fd074e9 # v4.0.0 - name: Install Node.js uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 @@ -76,7 +76,7 @@ jobs: run: yamllint --config-file .yamllint .github/ infra/ docker-compose*.yml - name: Run ShellCheck - run: shellcheck --severity=warning scripts/*.sh entrypoint.sh migrate.sh + run: shellcheck --severity=warning scripts/*.sh entrypoint.sh migrate.sh worker/entrypoint-worker.sh # ============================================ # TYPE CHECK - Static type analysis @@ -164,7 +164,7 @@ jobs: image: postgres:15 env: POSTGRES_USER: test_user - POSTGRES_PASSWORD: test_pass + POSTGRES_PASSWORD: test_pw POSTGRES_DB: test_db options: >- --health-cmd pg_isready --health-interval 10s --health-timeout 5s --health-retries 5 @@ -200,20 +200,24 @@ jobs: - name: Create test environment run: | + SK="$(openssl rand -hex 32)" cp .env.example .env - sed -i 's|^DB_PASSWORD=.*|DB_PASSWORD=test_pass|' .env - sed -i 's|^SECRET_KEY=$|SECRET_KEY=test-secret-key-for-ci-at-least-32-chars-long|' .env + sed -i "s|^DB_PASSWORD=.*|DB_PASSWORD=test_pw|" .env + sed -i "s|^SECRET_KEY=$|SECRET_KEY=${SK}|" .env + echo "PGPASSWORD=test_pw" >> "$GITHUB_ENV" - name: Create database schema run: | uv run python -c " import asyncio + import os from sqlalchemy.ext.asyncio import create_async_engine from core.database import Base from core.models import User, DownloadJob async def init_db(): - engine = create_async_engine('postgresql+asyncpg://test_user:test_pass@localhost:5432/test_db') + db_pass = os.environ['PGPASSWORD'] + engine = create_async_engine(f'postgresql+asyncpg://test_user:{db_pass}@localhost:5432/test_db') async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) await engine.dispose() diff --git a/.gitignore b/.gitignore index 024244e7..7428ed0e 100644 --- a/.gitignore +++ b/.gitignore @@ -180,3 +180,15 @@ docs/CRITIQUE-ISSUES.md .agents/ .kilocode/skills/* .agents/.story-automator-active + +# Local scan and audit artifacts (not checked in) +.betterleaks.toml +.betterleaksignore +betterleaks-setup.md +betterleaks.sarif +SOTA-AUDIT-REPORT.md +fail-console.md +fail-result.md +fail.har +# HAR files and local exports that are ephemeral +notegpt-clone/ diff --git a/.secrets.baseline b/.secrets.baseline index 127212f7..2e712109 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -146,14 +146,14 @@ { "type": "Secret Keyword", "filename": ".github/workflows/fastapi-test.yml", - "hashed_secret": "c94d65f02a652d11c2e5c2e1ccf38dce5a076e1e", + "hashed_secret": "d986d7729a3ca93e8fb8d9c0f1962564aed9f829", "is_verified": false, "line_number": 167 }, { "type": "Secret Keyword", "filename": ".github/workflows/fastapi-test.yml", - "hashed_secret": "89edba72d4aef5098771cee787b40b81af666eb3", + "hashed_secret": "fac0e4eec4d11a3bbad583ff2d750d900a70bf72", "is_verified": false, "line_number": 202 } diff --git a/Dockerfile b/Dockerfile index 858f673b..e87213f1 100644 --- a/Dockerfile +++ b/Dockerfile @@ -7,7 +7,7 @@ # ============================================ # Stage 1: Python Dependency Builder # ============================================ -FROM python:3.12-slim AS python-builder +FROM python@sha256:6c4dd321d176d61ea848dc8c73a4f7dbae8f70e0ee48bb411ea2f045b599fa8e AS python-builder ENV PYTHONDONTWRITEBYTECODE=1 \ PYTHONUNBUFFERED=1 @@ -29,7 +29,7 @@ ENV PATH="/opt/venv/bin:$PATH" \ UV_COMPILE_BYTECODE=1 # Install uv binary (single static binary, ~25MB, not copied to final image) -COPY --from=ghcr.io/astral-sh/uv:0.6 /uv /bin/uv +COPY --from=ghcr.io/astral-sh/uv@sha256:4a6c9444b126bd325fba904bff796bf91fb777bf6148d60109c4cb1de2ffc497 /uv /bin/uv # Copy manifest and lockfile first → cacheable dependency layer COPY pyproject.toml uv.lock ./ @@ -41,7 +41,7 @@ RUN --mount=type=cache,target=/root/.cache/uv \ # ============================================ # Stage 2: Frontend Builder # ============================================ -FROM node:20-alpine AS frontend-builder +FROM node@sha256:fb4cd12c85ee03686f6af5362a0b0d56d50c58a04632e6c0fb8363f609372293 AS frontend-builder WORKDIR /app # Install pnpm for package management (version pinned in frontend/package.json packageManager field) @@ -102,7 +102,7 @@ RUN mkdir -p /app/app/static/swagger && \ # ============================================ # Stage 4: Runtime Base # ============================================ -FROM python:3.12-slim AS runtime-base +FROM python@sha256:6c4dd321d176d61ea848dc8c73a4f7dbae8f70e0ee48bb411ea2f045b599fa8e AS runtime-base ENV PYTHONDONTWRITEBYTECODE=1 # Install runtime dependencies with apt cache mounts @@ -115,7 +115,10 @@ RUN --mount=type=cache,target=/var/cache/apt,sharing=locked \ curl \ gnupg \ && mkdir -p /etc/apt/keyrings \ - && curl -fsSL https://deb.nodesource.com/gpgkey/nodesource-repo.gpg.key | gpg --dearmor -o /etc/apt/keyrings/nodesource.gpg \ + && curl -fsSL https://deb.nodesource.com/gpgkey/nodesource-repo.gpg.key -o /tmp/nodesource-repo.gpg.key \ + && echo "b42e0321dabdc24e892115da705cf061167eac12a317f23d329862d0aa0a271d /tmp/nodesource-repo.gpg.key" | sha256sum -c - \ + && gpg --dearmor -o /etc/apt/keyrings/nodesource.gpg /tmp/nodesource-repo.gpg.key \ + && rm /tmp/nodesource-repo.gpg.key \ && echo "deb [signed-by=/etc/apt/keyrings/nodesource.gpg] https://deb.nodesource.com/node_20.x nodistro main" > /etc/apt/sources.list.d/nodesource.list \ && apt-get update \ && apt-get install -y --no-install-recommends nodejs \ diff --git a/app/api/dependencies/__init__.py b/app/api/dependencies/__init__.py index e5746d24..d0328e37 100644 --- a/app/api/dependencies/__init__.py +++ b/app/api/dependencies/__init__.py @@ -77,7 +77,7 @@ async def get_current_user_from_cookie( if credentials is not None: token = credentials.credentials else: - token = request.cookies.get("access_token") + token = request.cookies.get("__Host-access_token") return await _resolve_user_from_token(db, token, expected_type=ACCESS_TOKEN_TYPE) diff --git a/app/api/docs.py b/app/api/docs.py index 9dea1de8..580d6661 100644 --- a/app/api/docs.py +++ b/app/api/docs.py @@ -30,7 +30,7 @@ def register_docs_routes(app: FastAPI) -> None: """Register custom Swagger UI and ReDoc routes.""" @app.get("/docs", include_in_schema=False) - async def custom_docs(request: Request): + async def custom_docs(request: Request) -> HTMLResponse: nonce = request.state.nonce swagger_dir = APP_DIR / "static" / "swagger" if swagger_dir.exists(): @@ -69,7 +69,7 @@ async def custom_docs(request: Request): return docs_response @app.get("/redoc", include_in_schema=False) - async def custom_redoc(request: Request): + async def custom_redoc(request: Request) -> HTMLResponse: nonce = request.state.nonce redoc_dir = APP_DIR / "static" / "redoc" if redoc_dir.exists(): @@ -97,7 +97,7 @@ async def custom_redoc(request: Request): def _inject_inline_script_nonce(html: str, nonce: str) -> str: """Add the request nonce to FastAPI's generated inline docs script.""" return html.replace( - " + {% endblock %} diff --git a/app/templates/slides/presentation.html b/app/templates/slides/presentation.html index 13a6ed45..1bf37e0c 100644 --- a/app/templates/slides/presentation.html +++ b/app/templates/slides/presentation.html @@ -9,7 +9,7 @@ - @@ -281,7 +286,7 @@
-
+

Every downloader
@@ -323,7 +328,7 @@

No Observability

-
+
Resilience Architecture — Recovery Flow
@@ -331,7 +336,7 @@

No Observability

Happy Path — Normal Operation
-
+
@@ -524,7 +529,7 @@

No Observability

-' in response.text @@ -178,7 +178,7 @@ async def test_dashboard_route_renders_scoped_download_form_contract(): """The rendered dashboard exposes the scoped HTMX download form contract.""" async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: access_token = await create_test_user_and_login(client) - response = await client.get("/web/downloads", cookies={"access_token": access_token}) + response = await client.get("/web/downloads", cookies={"__Host-access_token": access_token}) assert response.status_code == 200 assert 'id="download-form"' in response.text @@ -198,7 +198,7 @@ async def test_non_dashboard_pages_do_not_render_sse_extension(): register_response = await client.get("/web/register") access_token = await create_test_user_and_login(client) settings_response = await client.get( - "/web/settings", cookies={"access_token": access_token} + "/web/settings", cookies={"__Host-access_token": access_token} ) for response in (login_response, register_response, settings_response): diff --git a/tests/test_story_8_5_missing_ui_states.py b/tests/test_story_8_5_missing_ui_states.py index 5047bd2b..05629d17 100644 --- a/tests/test_story_8_5_missing_ui_states.py +++ b/tests/test_story_8_5_missing_ui_states.py @@ -64,7 +64,7 @@ async def test_settings_username_save_button_has_scoped_loading_contract(): async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: access_token = await create_test_user_and_login(client) - response = await client.get("/web/settings", cookies={"access_token": access_token}) + response = await client.get("/web/settings", cookies={"__Host-access_token": access_token}) assert response.status_code == 200 assert 'id="username-settings-form"' in response.text @@ -93,7 +93,7 @@ async def test_settings_username_htmx_fragments_cover_success_and_error_states() """The username save endpoint returns HTMX fragments for success and critical errors.""" async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: access_token = await create_test_user_and_login(client) - client.cookies.set("access_token", access_token) + client.cookies.set("__Host-access_token", access_token) settings_response = await client.get("/web/settings") csrf_token = _csrf_token(client, settings_response) diff --git a/tests/test_story_8_6_accessibility_audit.py b/tests/test_story_8_6_accessibility_audit.py index f064a8b1..0500c742 100644 --- a/tests/test_story_8_6_accessibility_audit.py +++ b/tests/test_story_8_6_accessibility_audit.py @@ -254,7 +254,7 @@ async def test_authenticated_dashboard_and_settings_render_accessible_controls() """Authenticated pages render keyboard-reachable controls without gray-500 text.""" async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: access_token = await create_test_user_and_login(client) - cookies = {"access_token": access_token} + cookies = {"__Host-access_token": access_token} dashboard_response = await client.get("/web/downloads", cookies=cookies) settings_response = await client.get("/web/settings", cookies=cookies) @@ -297,7 +297,7 @@ async def test_settings_error_state_renders_accessible_error_contract(): access_token = await create_test_user_and_login(client) response = await client.get( "/web/settings?error=bad_current_password", - cookies={"access_token": access_token}, + cookies={"__Host-access_token": access_token}, ) assert response.status_code == 200 diff --git a/tests/test_worker/test_browser_executor.py b/tests/test_worker/test_browser_executor.py new file mode 100644 index 00000000..5e9a271c --- /dev/null +++ b/tests/test_worker/test_browser_executor.py @@ -0,0 +1,341 @@ +"""Tests for worker.browser_executor module. + +Covers the I/O matrix from spec-gh-140-p2-worker-integration.md: +- success path returns (file_path, file_name, None) +- error code → ErrorCategory mapping for every documented microservice code +- HTTP transport failures map to TRANSIENT/TIMEOUT +- circuit breaker integration (open circuit → TRANSIENT, no HTTP call) +- no httpx exception leaks past extract_media +""" + +from __future__ import annotations + +import httpx +import pytest + +from app.services.circuit_breaker import CircuitState +from app.services.error_classifier import ErrorCategory +from worker.browser_executor import ( + BrowserExecutorError, + _map_response_to_category, + extract_media, + get_browser_downloader_circuit_breaker, + select_executor, +) + + +@pytest.fixture(autouse=True) +def _reset_breaker_state(): + """Reset the module-level circuit breaker between tests. + + The breaker is a singleton; without this fixture a test that opens the + breaker (e.g. via `record_failure` loops) leaks state into subsequent + tests and triggers spurious `circuit_open` errors. + """ + breaker = get_browser_downloader_circuit_breaker() + breaker._state = CircuitState.CLOSED + breaker._failure_count = 0 + breaker._success_count = 0 + breaker._last_failure_time = None + breaker._half_open_calls = 0 + yield + breaker._state = CircuitState.CLOSED + breaker._failure_count = 0 + breaker._success_count = 0 + breaker._last_failure_time = None + breaker._half_open_calls = 0 + + +# -- select_executor: hostname routing ----------------------------------- + + +class TestSelectExecutor: + """Hostname-based dispatch — pure function, no settings touch.""" + + @pytest.mark.unit + @pytest.mark.parametrize( + "url", + [ + "https://www.tiktok.com/@user/video/123", + "https://tiktok.com/@u/v/1", + "https://m.tiktok.com/v/1.html", + "https://tiktokv.com/share/video/1", + "https://vm.tiktok.com/abcdef", + "https://www.instagram.com/reel/abc", + "https://instagram.com/p/xyz", + "https://instagr.am/p/abc", + "https://twitter.com/user/status/1", + "https://x.com/user/status/1", + "https://t.co/abc", + ], + ) + def test_browser_platforms_route_to_browser(self, url: str) -> None: + assert select_executor(url) == "browser" + + @pytest.mark.unit + def test_fqdn_trailing_dot_routes_to_browser(self) -> None: + # Some DNS resolvers return FQDN form (with trailing dot). + # We must still match. + assert select_executor("https://www.tiktok.com./@u/v/1") == "browser" + assert select_executor("https://instagram.com./p/x") == "browser" + + @pytest.mark.unit + @pytest.mark.parametrize( + "url", + [ + "https://www.youtube.com/watch?v=abc", + "https://youtu.be/abc", + "https://example.com/foo", + "https://vimeo.com/123", + "", + ], + ) + def test_non_browser_platforms_route_to_youtube(self, url: str) -> None: + assert select_executor(url) == "youtube" + + @pytest.mark.unit + def test_unparseable_url_falls_through_to_youtube(self) -> None: + # Malformed URL is treated as unknown → yt-dlp (current behavior) + assert select_executor("not a url at all") == "youtube" + + +# -- _map_response_to_category: error code → ErrorCategory --------------- + + +class TestMapResponseToCategory: + """Single source of truth for microservice error codes.""" + + @pytest.mark.unit + def test_drm_detected_is_blocked(self) -> None: + assert _map_response_to_category("drm_detected") == ErrorCategory.BLOCKED + + @pytest.mark.unit + def test_anti_bot_block_is_blocked(self) -> None: + assert _map_response_to_category("anti_bot_block") == ErrorCategory.BLOCKED + + @pytest.mark.unit + def test_no_media_found_is_not_found(self) -> None: + assert _map_response_to_category("no_media_found") == ErrorCategory.NOT_FOUND + + @pytest.mark.unit + def test_network_error_is_transient(self) -> None: + assert _map_response_to_category("network_error") == ErrorCategory.TRANSIENT + + @pytest.mark.unit + def test_timeout_is_timeout(self) -> None: + assert _map_response_to_category("request_timeout") == ErrorCategory.TIMEOUT + + @pytest.mark.unit + def test_http_5xx_is_transient(self) -> None: + assert _map_response_to_category("http_503") == ErrorCategory.TRANSIENT + + @pytest.mark.unit + def test_http_4xx_unknown_is_blocked(self) -> None: + assert _map_response_to_category("http_400") == ErrorCategory.BLOCKED + + @pytest.mark.unit + def test_invalid_request_is_blocked(self) -> None: + assert _map_response_to_category("invalid_request") == ErrorCategory.BLOCKED + + @pytest.mark.unit + def test_http_429_rate_limit_is_transient(self) -> None: + # 429 from the microservice (in the error code) is rate limiting, + # not a platform-level block. Retries should kick in. + assert _map_response_to_category("http_429") == ErrorCategory.TRANSIENT + + @pytest.mark.unit + def test_unknown_code_defaults_to_transient(self) -> None: + assert _map_response_to_category("something_new") == ErrorCategory.TRANSIENT + + +# -- extract_media: HTTP path -------------------------------------------- + + +def _make_mock_client(response_status: int, body: dict | str) -> httpx.AsyncClient: + """Build an httpx.AsyncClient whose single POST returns the given response. + + Uses httpx.MockTransport (httpx's built-in test utility — no external + mocking library required). + """ + + def handler(request: httpx.Request) -> httpx.Response: + if isinstance(body, str): + return httpx.Response(response_status, text=body) + return httpx.Response(response_status, json=body) + + return httpx.AsyncClient(transport=httpx.MockTransport(handler)) + + +class TestExtractMediaSuccess: + @pytest.mark.asyncio + @pytest.mark.unit + async def test_success_returns_tuple_with_filename_derived_from_path(self) -> None: + client = _make_mock_client( + 200, + {"status": "success", "file_path": "/storage/abc-123.mp4", "tier_used": 1}, + ) + result = await extract_media( + "https://tiktok.com/@u/v/1", + "/storage", + client=client, + ) + assert result == ("/storage/abc-123.mp4", "abc-123.mp4", None) + + +class TestExtractMediaErrorCodes: + @pytest.mark.asyncio + @pytest.mark.unit + async def test_drm_detected_maps_to_blocked(self) -> None: + client = _make_mock_client( + 502, + {"status": "failed", "error": "drm_detected"}, + ) + with pytest.raises(BrowserExecutorError) as exc: + await extract_media("https://tiktok.com/@u/v/1", "/storage", client=client) + assert exc.value.category == ErrorCategory.BLOCKED + assert exc.value.signal == "drm_detected" + + @pytest.mark.asyncio + @pytest.mark.unit + async def test_anti_bot_block_maps_to_blocked(self) -> None: + client = _make_mock_client( + 502, + {"status": "failed", "error": "anti_bot_block"}, + ) + with pytest.raises(BrowserExecutorError) as exc: + await extract_media("https://instagram.com/p/x", "/storage", client=client) + assert exc.value.category == ErrorCategory.BLOCKED + + @pytest.mark.asyncio + @pytest.mark.unit + async def test_no_media_found_maps_to_not_found(self) -> None: + client = _make_mock_client( + 502, + {"status": "failed", "error": "no_media_found"}, + ) + with pytest.raises(BrowserExecutorError) as exc: + await extract_media("https://tiktok.com/@u/v/1", "/storage", client=client) + assert exc.value.category == ErrorCategory.NOT_FOUND + + @pytest.mark.asyncio + @pytest.mark.unit + async def test_http_5xx_with_json_error_maps_to_transient(self) -> None: + client = _make_mock_client( + 503, + {"status": "failed", "error": "concurrency_limit"}, + ) + with pytest.raises(BrowserExecutorError) as exc: + await extract_media("https://tiktok.com/@u/v/1", "/storage", client=client) + # 'concurrency_limit' is not a known code → falls through to TRANSIENT + assert exc.value.category == ErrorCategory.TRANSIENT + assert exc.value.signal == "concurrency_limit" + + @pytest.mark.asyncio + @pytest.mark.unit + async def test_http_503_empty_body_maps_to_transient_with_status_signal(self) -> None: + """AC4: 503 with an empty body should still classify as TRANSIENT + with the synthetic http_ signal — covers the case where + the microservice is overloaded and closes the response without + writing JSON. + """ + client = _make_mock_client(503, "") + with pytest.raises(BrowserExecutorError) as exc: + await extract_media("https://tiktok.com/@u/v/1", "/storage", client=client) + assert exc.value.category == ErrorCategory.TRANSIENT + assert exc.value.signal == "http_503" + + @pytest.mark.asyncio + @pytest.mark.unit + async def test_non_json_error_body_maps_to_transient(self) -> None: + client = _make_mock_client(502, "internal server error") + with pytest.raises(BrowserExecutorError) as exc: + await extract_media("https://tiktok.com/@u/v/1", "/storage", client=client) + assert exc.value.category == ErrorCategory.TRANSIENT + assert exc.value.signal == "http_502" + + @pytest.mark.asyncio + @pytest.mark.unit + async def test_http_400_maps_to_blocked(self) -> None: + client = _make_mock_client( + 400, + {"status": "failed", "error": "invalid_request", "message": "bad url"}, + ) + with pytest.raises(BrowserExecutorError) as exc: + await extract_media("https://tiktok.com/@u/v/1", "/storage", client=client) + assert exc.value.category == ErrorCategory.BLOCKED + + @pytest.mark.asyncio + @pytest.mark.unit + async def test_200_with_missing_file_path_maps_to_transient(self) -> None: + client = _make_mock_client(200, {"status": "success"}) + with pytest.raises(BrowserExecutorError) as exc: + await extract_media("https://tiktok.com/@u/v/1", "/storage", client=client) + assert exc.value.category == ErrorCategory.TRANSIENT + assert exc.value.signal == "missing_file_path" + + @pytest.mark.asyncio + @pytest.mark.unit + async def test_200_with_non_dict_json_body_maps_to_transient(self) -> None: + """Microservice contract violation: 200 OK with a JSON list/null/scalar body.""" + client = _make_mock_client(200, [1, 2, 3]) + with pytest.raises(BrowserExecutorError) as exc: + await extract_media("https://tiktok.com/@u/v/1", "/storage", client=client) + assert exc.value.category == ErrorCategory.TRANSIENT + assert exc.value.signal == "invalid_response_shape" + + +class TestExtractMediaCircuitBreaker: + @pytest.mark.asyncio + @pytest.mark.unit + async def test_open_circuit_raises_transient_without_http_call(self) -> None: + # When the breaker is OPEN, no HTTP call should be made. We use a + # transport that would raise if invoked, proving the call was skipped. + # Phase 2 fix: CircuitBreakerOpenError propagates raw so the processor's + # deferred-job path handles it (worker/processor.py:_handle_circuit_open). + from app.services.circuit_breaker import CircuitBreakerOpenError + + called = False + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal called + called = True + return httpx.Response(200, json={"status": "success", "file_path": "/x.mp4"}) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + + # Force the breaker open by recording 5 consecutive failures + breaker = get_browser_downloader_circuit_breaker() + for _ in range(breaker.failure_threshold): + await breaker.record_failure(RuntimeError("boom")) + + with pytest.raises(CircuitBreakerOpenError): + await extract_media( + "https://tiktok.com/@u/v/1", + "/storage", + client=client, + ) + assert called is False, "HTTP transport was called despite open breaker" + + @pytest.mark.asyncio + @pytest.mark.unit + async def test_no_httpx_exception_leaks_past_extract_media(self) -> None: + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ReadError("stream broke") + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + with pytest.raises(BrowserExecutorError): + await extract_media("https://tiktok.com/@u/v/1", "/storage", client=client) + # The exception must be BrowserExecutorError, not httpx.HTTPError + # (defensive — the type annotation in extract_media's docstring). + + +# -- Circuit breaker singleton ------------------------------------------- + + +class TestBreakerSingleton: + @pytest.mark.unit + def test_get_breaker_returns_same_instance(self) -> None: + a = get_browser_downloader_circuit_breaker() + b = get_browser_downloader_circuit_breaker() + assert a is b + assert a.name == "browser_downloader" diff --git a/tests/test_worker/test_job_executor_routing.py b/tests/test_worker/test_job_executor_routing.py new file mode 100644 index 00000000..a02be382 --- /dev/null +++ b/tests/test_worker/test_job_executor_routing.py @@ -0,0 +1,203 @@ +"""Tests for the Phase 2 routing decision in worker.job_executor. + +The I/O matrix in spec-gh-140-p2-worker-integration.md: +- tiktok/instagram/twitter/x URL + feature on → browser executor +- youtube URL → yt-dlp +- unknown host → yt-dlp (fallthrough, current behavior) +- feature off forces yt-dlp even for TikTok +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, patch + +import pytest + +from worker import job_executor +from worker.job_executor import _resolve_executor_kind, select_executor + + +class TestSelectExecutor: + """select_executor is the pure routing function. It does not touch settings.""" + + @pytest.mark.unit + @pytest.mark.parametrize( + "url", + [ + "https://www.tiktok.com/@user/video/123", + "https://vm.tiktok.com/x", + "https://www.instagram.com/reel/abc", + "https://instagr.am/p/x", + "https://twitter.com/u/status/1", + "https://x.com/u/status/1", + "https://t.co/abc", + ], + ) + def test_browser_platforms(self, url: str) -> None: + assert select_executor(url) == "browser" + + @pytest.mark.unit + @pytest.mark.parametrize( + "url", + [ + "https://www.youtube.com/watch?v=abc", + "https://youtu.be/abc", + "https://example.com/foo", + "https://vimeo.com/123", + ], + ) + def test_youtube_or_unknown(self, url: str) -> None: + assert select_executor(url) == "youtube" + + +class TestResolveExecutorKind: + """_resolve_executor_kind respects the browser_downloader_enabled flag.""" + + @pytest.mark.unit + def test_tiktok_url_with_feature_off_returns_youtube(self) -> None: + with patch.object(job_executor.settings, "browser_downloader_enabled", False): + assert _resolve_executor_kind("https://tiktok.com/@u/v/1") == "youtube" + + @pytest.mark.unit + def test_tiktok_url_with_feature_on_returns_browser(self) -> None: + with patch.object(job_executor.settings, "browser_downloader_enabled", True): + assert _resolve_executor_kind("https://tiktok.com/@u/v/1") == "browser" + + @pytest.mark.unit + def test_youtube_url_with_feature_on_returns_youtube(self) -> None: + with patch.object(job_executor.settings, "browser_downloader_enabled", True): + assert _resolve_executor_kind("https://youtu.be/abc") == "youtube" + + @pytest.mark.unit + def test_unknown_host_with_feature_on_returns_youtube(self) -> None: + with patch.object(job_executor.settings, "browser_downloader_enabled", True): + assert _resolve_executor_kind("https://example.com/foo") == "youtube" + + @pytest.mark.unit + def test_instagram_url_with_feature_off_returns_youtube(self) -> None: + with patch.object(job_executor.settings, "browser_downloader_enabled", False): + assert _resolve_executor_kind("https://instagram.com/p/x") == "youtube" + + +class TestExecuteRoutesToBrowserExecutor: + """End-to-end: `execute()` invokes the browser path for TikTok + feature on.""" + + @pytest.mark.asyncio + @pytest.mark.unit + async def test_browser_path_invokes_browser_executor_for_tiktok(self) -> None: + from core.models.download_job import DownloadJob + from worker.job_executor import execute + + job = DownloadJob( + id="550e8400-e29b-41d4-a716-446655440000", + user_id="550e8400-e29b-41d4-a716-446655440005", + url="https://tiktok.com/@u/v/1", + status="processing", + retry_count=0, + ) + + mock_browser = AsyncMock( + return_value=("/storage/abc.mp4", "abc.mp4", None), + ) + with ( + patch.object(job_executor.settings, "browser_downloader_enabled", True), + patch.object(job_executor.settings, "feature_throttle_preemptive_enabled", False), + patch.object(job_executor, "extract_media_browser", mock_browser), + patch.object( + job_executor, + "extract_media_with_circuit_breaker", + AsyncMock(), + ) as mock_ytdlp, + ): + db = AsyncMock() + db.execute = AsyncMock() + db.commit = AsyncMock() + # First db.execute call updates the row to completed; subsequent + # calls re-select. We return a fake result with rowcount=1. + update_result = AsyncMock() + update_result.rowcount = 1 + update_result.scalar_one_or_none = AsyncMock(return_value=job) + db.execute.return_value = update_result + + await execute(db, job, start_time=0.0) + + mock_browser.assert_awaited_once() + mock_ytdlp.assert_not_awaited() + + @pytest.mark.asyncio + @pytest.mark.unit + async def test_youtube_path_skips_browser_executor(self) -> None: + from core.models.download_job import DownloadJob + from worker.job_executor import execute + + job = DownloadJob( + id="550e8400-e29b-41d4-a716-446655440000", + user_id="550e8400-e29b-41d4-a716-446655440005", + url="https://youtu.be/abc", + status="processing", + retry_count=0, + ) + + mock_ytdlp = AsyncMock( + return_value=("/storage/yt.mp4", "yt.mp4", "Title"), + ) + with ( + patch.object(job_executor.settings, "browser_downloader_enabled", True), + patch.object(job_executor.settings, "feature_throttle_preemptive_enabled", False), + patch.object(job_executor, "extract_media_browser", AsyncMock()) as mock_browser, + patch.object( + job_executor, + "extract_media_with_circuit_breaker", + mock_ytdlp, + ), + ): + db = AsyncMock() + update_result = AsyncMock() + update_result.rowcount = 1 + update_result.scalar_one_or_none = AsyncMock(return_value=job) + db.execute = AsyncMock(return_value=update_result) + db.commit = AsyncMock() + + await execute(db, job, start_time=0.0) + + mock_ytdlp.assert_awaited_once() + mock_browser.assert_not_awaited() + + @pytest.mark.asyncio + @pytest.mark.unit + async def test_feature_off_routes_tiktok_to_ytdlp(self) -> None: + from core.models.download_job import DownloadJob + from worker.job_executor import execute + + job = DownloadJob( + id="550e8400-e29b-41d4-a716-446655440000", + user_id="550e8400-e29b-41d4-a716-446655440005", + url="https://tiktok.com/@u/v/1", + status="processing", + retry_count=0, + ) + + mock_ytdlp = AsyncMock( + return_value=("/storage/yt.mp4", "yt.mp4", None), + ) + with ( + patch.object(job_executor.settings, "browser_downloader_enabled", False), + patch.object(job_executor.settings, "feature_throttle_preemptive_enabled", False), + patch.object(job_executor, "extract_media_browser", AsyncMock()) as mock_browser, + patch.object( + job_executor, + "extract_media_with_circuit_breaker", + mock_ytdlp, + ), + ): + db = AsyncMock() + update_result = AsyncMock() + update_result.rowcount = 1 + update_result.scalar_one_or_none = AsyncMock(return_value=job) + db.execute = AsyncMock(return_value=update_result) + db.commit = AsyncMock() + + await execute(db, job, start_time=0.0) + + mock_ytdlp.assert_awaited_once() + mock_browser.assert_not_awaited() diff --git a/tests/test_worker/test_outbox_recovery.py b/tests/test_worker/test_outbox_recovery.py index 1685f334..906f5039 100644 --- a/tests/test_worker/test_outbox_recovery.py +++ b/tests/test_worker/test_outbox_recovery.py @@ -92,13 +92,18 @@ async def test_sync_outbox_recovers_pending_entries(self, db_session, job_id, us synced = await sync_outbox_to_queue(batch_size=10) assert synced == 1 - # Outbox entry is DELETED after successful sync (not updated to enqueued) + # Outbox entry is marked 'processed' (not deleted) after successful sync. + # Expire the test session so the commit from the relay's separate session + # is visible through the same async engine. + await db_session.commit() + db_session.expire_all() from sqlalchemy import select - await db_session.commit() result = await db_session.execute(select(Outbox).where(Outbox.id == outbox_id)) - deleted_entry = result.scalar_one_or_none() - assert deleted_entry is None + processed_entry = result.scalar_one_or_none() + assert processed_entry is not None + assert processed_entry.status == "processed" + assert processed_entry.processed_at is not None @pytest.mark.unit async def test_sync_outbox_with_retry_scheduled(self, db_session, job_id, user_id): @@ -247,8 +252,16 @@ async def test_sync_outbox_with_for_update_skip_locked(self, db_session): class TestOutboxIdempotency: @pytest.mark.unit - async def test_duplicate_outbox_entry_prevented(self, db_session, job_id, user_id): - """Test that duplicate outbox entries are prevented by idempotent check.""" + async def test_duplicate_pending_outbox_entry_rejected_by_db(self, db_session, job_id, user_id): + """Test that a second 'pending' outbox row for the same job_id is rejected + by the partial unique index ``uq_outbox_pending_job_id``. + + The DB-enforced constraint is the primary guard; the application-layer + SELECT-then-INSERT check inside ``write_job_to_outbox`` provides a + cheaper fast-path for the common case. + """ + from sqlalchemy.exc import IntegrityError + job = DownloadJob( id=job_id, user_id=user_id, @@ -285,11 +298,9 @@ async def test_duplicate_outbox_entry_prevented(self, db_session, job_id, user_i status="pending", ) db_session.add(outbox2) - await db_session.commit() - - result = await db_session.execute(select(Outbox).where(Outbox.job_id == job_id)) - all_entries = result.scalars().all() - assert len(all_entries) == 2 + with pytest.raises(IntegrityError): + await db_session.commit() + await db_session.rollback() class TestOutboxBatchProcessing: @@ -327,11 +338,15 @@ async def test_sync_respects_batch_size(self, db_session): synced = await sync_outbox_to_queue(batch_size=2) assert synced == 2 - # Entries are DELETED after successful sync (batch_size=2 means 2 deleted) + # Entries are marked 'processed' after successful sync (batch_size=2 means 2 processed) await db_session.commit() result = await db_session.execute(select(Outbox)) remaining = result.scalars().all() - assert len(remaining) == 3 # 5 - 2 = 3 remain + assert len(remaining) == 5 + processed = [r for r in remaining if r.status == "processed"] + pending = [r for r in remaining if r.status == "pending"] + assert len(processed) == 2 + assert len(pending) == 3 class TestOutboxCrashRecoveryScenarios: @@ -367,13 +382,16 @@ async def test_job_created_but_not_enqueued(self, db_session, job_id, user_id): synced = await sync_outbox_to_queue(batch_size=10) assert synced == 1 - # Entry is DELETED after successful sync + # Entry is marked 'processed' after successful sync, retained for audit. await db_session.commit() + db_session.expire_all() from sqlalchemy import select result = await db_session.execute(select(Outbox).where(Outbox.id == outbox_id)) - deleted = result.scalar_one_or_none() - assert deleted is None + processed = result.scalar_one_or_none() + assert processed is not None + assert processed.status == "processed" + assert processed.processed_at is not None @pytest.mark.unit async def test_job_enqueued_twice_prevented(self, db_session, job_id, user_id): diff --git a/worker/browser_executor.py b/worker/browser_executor.py new file mode 100644 index 00000000..60e47656 --- /dev/null +++ b/worker/browser_executor.py @@ -0,0 +1,373 @@ +"""Browser downloader microservice HTTP client. + +Phase 2 worker integration: the worker calls the standalone Node.js +microservice at `packages/browser-downloader/` (Phase 0) over HTTP for +platforms that yt-dlp cannot handle (TikTok, Instagram, Twitter/X). + +This module owns three concerns: +1. Translate `POST :3000/download` responses into the worker's existing + `(file_path, file_name, title)` tuple shape. +2. Map the microservice's structured error codes into the existing + `app.services.error_classifier.ErrorCategory` values so the existing + retry/DLQ pipeline handles them. +3. Wrap every call in a named circuit breaker so a flaky downstream + cannot stall the worker. + +Design notes (KEEP): +- Single `BrowserExecutorError` exception type carries `category` and + `signal`. The job executor passes these to the existing retry machinery + unchanged. +- The error-code → category mapping is centralized in + `_map_response_to_category` — every failure path goes through it. +- The circuit breaker is a named singleton alongside + `get_youtube_circuit_breaker` in `app.services.circuit_breaker`. +- No progress streaming (microservice is single-shot HTTP); the worker + passes `progress_callback=None` from the caller. +""" + +from __future__ import annotations + +import json +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from typing import Any, ClassVar +from urllib.parse import urlparse + +import httpx + +from app.services.circuit_breaker import ( + CircuitBreaker, + CircuitBreakerOpenError, +) +from app.services.error_classifier import ErrorCategory +from core.config import settings +from core.logging_config import get_logger + +logger = get_logger(__name__) + + +class BrowserExecutorError(Exception): + """Failure with a classified retry category and a stable signal. + + The `category` value is consumed by the existing retry/DLQ pipeline in + `worker/job_executor.py`; the `signal` value is the original microservice + error code (or a synthetic marker for transport failures) preserved for + observability and tests. + + The `str(...)` representation embeds a marker that the existing + `app.services.error_classifier.classify_error` regex patterns can match, + so the typed `category` is corroborated by the string-based classifier + used in `worker/retry_scheduler.evaluate`. + """ + + _CATEGORY_MARKERS: ClassVar[dict[ErrorCategory, str]] = { + ErrorCategory.BLOCKED: "blocked (anti-bot or DRM)", + ErrorCategory.NOT_FOUND: "404 not found", + ErrorCategory.TIMEOUT: "Request timeout", + ErrorCategory.TRANSIENT: "network error (transient)", + ErrorCategory.UNKNOWN: "unknown error", + } + + def __init__(self, category: ErrorCategory, signal: str) -> None: + self.category = category + self.signal = signal + marker = self._CATEGORY_MARKERS.get(category, "transient error") + super().__init__(f"{marker}: {signal}") + + +@dataclass(slots=True) +class _HttpResponse: + """Minimal shape for httpx responses we read here. + + Allows tests to inject a fake without standing up an httpx transport. + The real path passes through `httpx.Response` (duck-typed). + """ + + status_code: int + body: bytes + + +ProgressCallback = Callable[[dict], Awaitable[None]] + + +# -- Singleton client & breaker ------------------------------------------- + +_client: httpx.AsyncClient | None = None +_breaker: CircuitBreaker | None = None + + +def _get_client() -> httpx.AsyncClient: + """Lazy singleton httpx client. + + `trust_env=False` so the worker does not pick up a stray HTTP_PROXY + from the environment and route internal service-to-service traffic + through an unrelated proxy. Operators who DO need proxy support can + add a `browser_downloader_proxy` setting in P4. + + Timeout model: 30s connect (matches the spec's "existing 30s connect" + pattern; httpx's default of 5s is too tight for a slow microservice + cold start) plus the configurable `browser_downloader_timeout` for + read/write/pool (default 300s — covers the full Tier 1 + Tier 2 + window in the microservice). + """ + global _client + if _client is None: + _client = httpx.AsyncClient( + timeout=httpx.Timeout( + connect=30.0, + read=settings.browser_downloader_timeout, + write=settings.browser_downloader_timeout, + pool=settings.browser_downloader_timeout, + ), + trust_env=False, + ) + return _client + + +def get_browser_downloader_circuit_breaker() -> CircuitBreaker: + """Lazy singleton circuit breaker for the browser-downloader service.""" + global _breaker + if _breaker is None: + _breaker = CircuitBreaker( + name="browser_downloader", + failure_threshold=5, + success_threshold=3, + reset_timeout=30.0, + half_open_max_calls=3, + use_redis_distributed=settings.browser_downloader_cb_use_redis, + ) + return _breaker + + +# -- Public API ------------------------------------------------------------ + + +def select_executor(url: str) -> str: + """Pick the executor kind for a job URL. + + Returns the literal string `"browser"` for known browser-only platforms + (TikTok, Instagram, Twitter/X) and `"youtube"` for everything else. The + feature flag (`browser_downloader_enabled`) is enforced by the caller in + `worker/job_executor.py` — this function is a pure hostname lookup so it + can be unit-tested without touching settings. + + Hostname matching uses suffix equality (e.g. `www.tiktok.com`, + `m.tiktok.com`, `vm.tiktok.com` all match `tiktok.com`). This keeps the + set small while still catching mobile subdomains that platforms + frequently serve on. + + Unknown hosts fall through to `"youtube"` to preserve pre-Phase-2 + behavior (yt-dlp attempts and fails as today). + """ + try: + hostname = (urlparse(url).hostname or "").lower() + except (ValueError, TypeError): + return "youtube" + # Strip a leading "www." so www.tiktok.com → tiktok.com. + if hostname.startswith("www."): + hostname = hostname[4:] + # FQDN form (trailing dot) — "tiktok.com." → "tiktok.com" before matching. + if hostname.endswith("."): + hostname = hostname[:-1] + browser_suffixes = ( + "tiktok.com", + "tiktokv.com", + "instagram.com", + "instagr.am", + "twitter.com", + "x.com", + "t.co", + ) + for suffix in browser_suffixes: + if hostname == suffix or hostname.endswith("." + suffix): + return "browser" + return "youtube" + + +async def extract_media( + url: str, + storage_path: str, + *, + progress_callback: ProgressCallback | None = None, + client: httpx.AsyncClient | None = None, +) -> tuple[str, str, str | None]: + """Call the browser-downloader microservice and return `(file_path, file_name, title)`. + + `title` is always `None` for browser-platform downloads in Phase 2 (the + microservice does not extract titles). The DB column is nullable so this + is safe to insert directly. + + Raises: + BrowserExecutorError: every failure mode is wrapped in this with a + classified `ErrorCategory`. No raw `httpx` exception leaks. + """ + _ = progress_callback # Microservice is single-shot; no progress stream. + http_client = client or _get_client() + breaker = get_browser_downloader_circuit_breaker() + + request_body = {"url": url, "output_dir": storage_path} + endpoint = settings.browser_downloader_endpoint.rstrip("/") + "/download" + + try: + return await breaker.execute(_call_service, http_client, endpoint, request_body) + except CircuitBreakerOpenError: + # Let CircuitBreakerOpenError propagate so the processor's dedicated + # deferred-job path can handle it (worker/processor.py:_handle_circuit_open). + raise + except BrowserExecutorError: + raise + except Exception as exc: + logger.error("browser_downloader_unexpected_error", error=str(exc), exc_info=True) + raise BrowserExecutorError( + category=ErrorCategory.UNKNOWN, signal="unexpected_error" + ) from exc + + +# -- Internal helpers ------------------------------------------------------ + + +async def _call_service( + http_client: httpx.AsyncClient, endpoint: str, body: dict[str, Any] +) -> tuple[str, str, str | None]: + """Single HTTP attempt. Returns the success tuple or raises BrowserExecutorError. + + Split out from `extract_media` so the circuit breaker can wrap exactly + one HTTP round-trip per recorded success/failure. + """ + try: + response = await http_client.post(endpoint, json=body) + except httpx.TimeoutException as exc: + logger.warning("browser_downloader_timeout", error=str(exc)) + raise BrowserExecutorError( + category=ErrorCategory.TIMEOUT, signal="request_timeout" + ) from exc + except httpx.ConnectError as exc: + logger.warning("browser_downloader_connect_error", error=str(exc)) + raise BrowserExecutorError( + category=ErrorCategory.TRANSIENT, signal="connect_error" + ) from exc + except httpx.HTTPError as exc: + logger.warning("browser_downloader_http_error", error=str(exc)) + raise BrowserExecutorError(category=ErrorCategory.TRANSIENT, signal="http_error") from exc + + if response.status_code == 200: + return _parse_success(response) + return _parse_failure_response(response) + + +def _parse_success(response: httpx.Response) -> tuple[str, str, str | None]: + """Parse a 200 OK response into the worker's tuple shape.""" + try: + payload = response.json() + except (json.JSONDecodeError, ValueError) as exc: + logger.warning("browser_downloader_non_json_success", status=response.status_code) + raise BrowserExecutorError( + category=ErrorCategory.TRANSIENT, signal="non_json_response" + ) from exc + + if not isinstance(payload, dict): + logger.warning( + "browser_downloader_invalid_response_shape", + payload_type=type(payload).__name__, + ) + raise BrowserExecutorError( + category=ErrorCategory.TRANSIENT, signal="invalid_response_shape" + ) + + if payload.get("status") != "success": + # 200 OK with a failed status is a microservice-side failure + # (e.g. a graceful degraded path). Reuse the failure parser. + return _parse_failure_payload(payload) + + file_path = payload.get("file_path") + if not isinstance(file_path, str) or not file_path: + logger.warning( + "browser_downloader_missing_file_path", + payload_keys=list(payload.keys()), + ) + raise BrowserExecutorError(category=ErrorCategory.TRANSIENT, signal="missing_file_path") + file_name = file_path.rsplit("/", 1)[-1] or file_path + return file_path, file_name, None + + +def _parse_failure_response(response: httpx.Response) -> tuple[str, str, str | None]: + """Parse a non-200 response, mapping HTTP status + JSON error code to a category.""" + signal = f"http_{response.status_code}" + try: + payload = response.json() + except (json.JSONDecodeError, ValueError) as err: + # Non-JSON error body — use the synthesized HTTP status signal so 404/403/429 + # are categorized correctly even when the body is empty or non-JSON. + logger.warning( + "browser_downloader_non_json_error", + status=response.status_code, + ) + raise BrowserExecutorError( + category=_map_response_to_category(signal), + signal=signal, + ) from err + + if not isinstance(payload, dict): + logger.warning( + "browser_downloader_invalid_response_shape", + payload_type=type(payload).__name__, + ) + raise BrowserExecutorError( + category=_map_response_to_category(signal), + signal=signal, + ) + + return _parse_failure_payload(payload, fallback_code=signal) + + +def _parse_failure_payload( + payload: dict[str, Any], + *, + fallback_code: str = "unknown_error", +) -> tuple[str, str, str | None]: + """Map a structured failure payload to an error category. + + Accepts both 200-with-failed-status and non-200 responses. + Raises BrowserExecutorError so the circuit breaker records a failure. + + When the JSON body lacks an explicit ``error`` field, ``fallback_code`` + (typically the synthesized ``http_`` signal) is used so HTTP + 404/403/429 still map to their correct categories. + """ + code: str = payload.get("error") # type: ignore[assignment] + if not isinstance(code, str) or not code: + code = fallback_code + category = _map_response_to_category(code, payload) + raise BrowserExecutorError(category=category, signal=code) + + +def _map_response_to_category(code: str, payload: dict[str, Any] | None = None) -> ErrorCategory: + """Single source of truth for microservice error code → ErrorCategory. + + Mapping is intentionally narrow: anything not explicitly recognized is + TRANSIENT (retries are safe; the worker can reclassify later if needed). + """ + # Terminal categories first — these never retry + if code in {"drm_detected", "anti_bot_block"}: + return ErrorCategory.BLOCKED + if code in {"no_media_found", "not_found", "private_content", "http_404"}: + return ErrorCategory.NOT_FOUND + # 429 rate-limit (microservice body) — TRANSIENT so the existing + # backoff machinery applies; falling through to BLOCKED here would + # send rate-limited jobs straight to the DLQ. + if code in {"http_429", "http_503"}: + return ErrorCategory.TRANSIENT + if code in {"network_error", "http_error", "connect_error", "non_json_response"}: + return ErrorCategory.TRANSIENT + if code in {"timeout", "request_timeout"}: + return ErrorCategory.TIMEOUT + # Generic 4xx (other than the BLOCKED/NOT_FOUND codes above) → BLOCKED + if code.startswith("http_4"): + return ErrorCategory.BLOCKED + if code.startswith("http_5"): + return ErrorCategory.TRANSIENT + if code == "circuit_open": + return ErrorCategory.TRANSIENT + if code == "invalid_request": + return ErrorCategory.BLOCKED + return ErrorCategory.TRANSIENT diff --git a/worker/dlq_manager.py b/worker/dlq_manager.py index 5b66b753..f78b0c04 100644 --- a/worker/dlq_manager.py +++ b/worker/dlq_manager.py @@ -87,7 +87,7 @@ async def mark_failed_and_move_to_dlq( last_error=accumulated, error_category=decision.category.value, completed_at=datetime.now(UTC), - ) + ), ) if int(getattr(failed_result, "rowcount", 0) or 0) == 0: logger.warning( @@ -139,7 +139,7 @@ async def reset_stuck_jobs(timeout_minutes: int = 10) -> int: updated_at=datetime.now(UTC), ) .returning(DownloadJob.id, DownloadJob.user_id) - .execution_options(synchronize_session=False) + .execution_options(synchronize_session=False), ) affected = result.fetchall() if not affected: diff --git a/worker/health.py b/worker/health.py index 2d652958..b0303698 100644 --- a/worker/health.py +++ b/worker/health.py @@ -233,6 +233,8 @@ def start_health_server(port: int | None = None) -> uvicorn.Server | None: if port is None: port = int(env_port) + health_host = os.environ.get("WORKER_HEALTH_HOST", "0.0.0.0") + if port == 0: logger.info("worker_health_http_disabled") return None @@ -244,7 +246,7 @@ def start_health_server(port: int | None = None) -> uvicorn.Server | None: config = uvicorn.Config( health_app, - host="0.0.0.0", + host=health_host, port=port, log_level=os.environ.get("LOG_LEVEL", "info").lower(), access_log=False, diff --git a/worker/job_claimer.py b/worker/job_claimer.py index d4c01b76..0581d5ef 100644 --- a/worker/job_claimer.py +++ b/worker/job_claimer.py @@ -54,7 +54,7 @@ async def claim_next(db: AsyncSession, job_id: UUID | str | bytes) -> DownloadJo .where(DownloadJob.id == normalized_job_id, DownloadJob.status == "pending") .values(status="processing", updated_at=datetime.now(UTC)) .returning(DownloadJob) - .execution_options(synchronize_session=False) + .execution_options(synchronize_session=False), ) job = result.scalar_one_or_none() await db.commit() @@ -64,7 +64,7 @@ async def claim_next(db: AsyncSession, job_id: UUID | str | bytes) -> DownloadJo async def heartbeat(db: AsyncSession, job_id: UUID) -> None: """Update a processing job heartbeat timestamp.""" await db.execute( - update(DownloadJob).where(DownloadJob.id == job_id).values(updated_at=datetime.now(UTC)) + update(DownloadJob).where(DownloadJob.id == job_id).values(updated_at=datetime.now(UTC)), ) await db.commit() @@ -89,7 +89,7 @@ async def periodic_heartbeat( await hb_db.execute( update(DownloadJob) .where(DownloadJob.id == job_id) - .values(updated_at=datetime.now(UTC)) + .values(updated_at=datetime.now(UTC)), ) await hb_db.commit() except asyncio.CancelledError: diff --git a/worker/job_executor.py b/worker/job_executor.py index 6b8538c4..f8e97991 100644 --- a/worker/job_executor.py +++ b/worker/job_executor.py @@ -27,6 +27,12 @@ from core.models.download_job import DownloadJob from core.models.outbox import Outbox from core.queue import redis_client +from worker.browser_executor import ( + extract_media as extract_media_browser, +) +from worker.browser_executor import ( + select_executor, +) from worker.health import update_worker_state from worker.job_claimer import heartbeat, periodic_heartbeat @@ -53,6 +59,20 @@ class ExecutionResult: completed: bool = False +def _resolve_executor_kind(url: str) -> str: + """Phase 2: pick the executor kind, respecting the browser feature flag. + + `select_executor` does the pure hostname-based routing. The feature + flag forces a fallback to the existing yt-dlp path when the microservice + is disabled, preserving pre-Phase-2 behavior. Tests for the routing + decision should patch `select_executor` directly to avoid touching + settings; this wrapper is the integration point in `execute()`. + """ + if not settings.browser_downloader_enabled: + return "youtube" + return select_executor(url) + + async def publish_job_status(job: DownloadJob) -> None: """Publish the current job status to the user's pub/sub channel.""" try: @@ -85,7 +105,7 @@ async def requeue_job(job_id: UUID, db) -> bool: { "retry_count": 0, "next_retry_at": datetime.now(UTC).isoformat(), - } + }, ), status="pending", ) @@ -100,7 +120,7 @@ async def requeue_job(job_id: UUID, db) -> bool: .values( status="pending", updated_at=datetime.now(UTC), - ) + ), ) await db.commit() if result.rowcount == 0: @@ -138,7 +158,7 @@ async def check_chaos_injection(db, job_id: UUID, start_time: float) -> bool: await db.execute( update(DownloadJob) .where(DownloadJob.id == job_id) - .values(status="pending", updated_at=datetime.now(UTC)) + .values(status="pending", updated_at=datetime.now(UTC)), ) await db.commit() RECOVERIES.labels(reason="zombie_sweep_recovery").inc() @@ -148,7 +168,7 @@ async def check_chaos_injection(db, job_id: UUID, start_time: float) -> bool: JOB_DURATION_SECONDS.observe(time.time() - start_time) return True except Exception: - pass + logger.debug("zombie_sweep_recovery skipped (non-critical)", exc_info=True) try: if await redis_client.exists("chaos:db_failover"): @@ -161,15 +181,15 @@ async def check_chaos_injection(db, job_id: UUID, start_time: float) -> bool: except OperationalError: raise except Exception: - pass + logger.debug("chaos_db_failover check skipped (non-critical)", exc_info=True) try: if await redis_client.exists("chaos:slow_processing"): - delay = random.uniform(5.0, 20.0) + delay = random.uniform(5.0, 20.0) # noqa: S311 — chaos testing, not crypto logger.info("chaos_slow_processing", job_id=str(job_id), delay_seconds=round(delay, 1)) await asyncio.sleep(delay) except Exception: - pass + logger.debug("chaos_slow_processing skipped (non-critical)", exc_info=True) if shutdown_event.is_set(): logger.info("Shutdown requested, requeueing job %s", job_id) @@ -201,11 +221,19 @@ async def execute( if await check_chaos_injection(db, job_id, start_time): return ExecutionResult(ExecutionStatus.CONSUMED, job_id, job=job) - if settings.feature_throttle_preemptive_enabled: + # Phase 2: route the job to the right executor. Browser-platform + # jobs skip the throttle predictor (yt-dlp-specific signal) and the + # progress callback (microservice is single-shot HTTP). The feature + # flag forces a fallback to yt-dlp when the microservice is disabled. + executor_kind = _resolve_executor_kind(job.url) + + if executor_kind == "youtube" and settings.feature_throttle_preemptive_enabled: throttle_risk = await get_risk_score("youtube") if throttle_risk >= 1.0: logger.warning( - "preemptive_throttle_block", job_id=str(job_id), risk_score=throttle_risk + "preemptive_throttle_block", + job_id=str(job_id), + risk_score=throttle_risk, ) await requeue_job(job_id, db) JOBS_COMPLETED.labels(status="deferred").inc() @@ -274,28 +302,43 @@ async def progress_callback(progress_data: dict) -> None: stop_hb = asyncio.Event() hb_task = asyncio.create_task( - periodic_heartbeat(get_async_session_factory(), job_id, stop_hb) + periodic_heartbeat(get_async_session_factory(), job_id, stop_hb), ) loop = asyncio.get_running_loop() - extract_task = loop.create_task( - extract_media_with_circuit_breaker( - job.url, - settings.storage_path, - progress_callback=progress_callback, + if executor_kind == "browser": + logger.info( + "job_routed_to_browser_executor", + job_id=str(job_id), + url=job.url, + ) + extract_task = loop.create_task( + extract_media_browser( + job.url, + settings.storage_path, + progress_callback=None, + ), + ) + else: + extract_task = loop.create_task( + extract_media_with_circuit_breaker( + job.url, + settings.storage_path, + progress_callback=progress_callback, + ), ) - ) try: file_path, file_name, title = await asyncio.wait_for( - extract_task, timeout=attempt_timeout + extract_task, + timeout=attempt_timeout, ) except TimeoutError: extract_task.cancel() try: await extract_task except (asyncio.CancelledError, Exception): - pass + logger.debug("extract_task cancel cleanup (non-critical)", exc_info=True) if getattr(worker_main_module, "shutdown_requested_at", None) is not None: await requeue_job(job_id, db) @@ -306,7 +349,7 @@ async def progress_callback(progress_data: dict) -> None: return ExecutionResult(ExecutionStatus.REQUEUED, job_id, job=job) raise TimeoutError( - f"Extraction timed out after {attempt_timeout}s (attempt {job.retry_count + 1})" + f"Extraction timed out after {attempt_timeout}s (attempt {job.retry_count + 1})", ) from None if shutdown_event.is_set(): @@ -331,7 +374,7 @@ async def progress_callback(progress_data: dict) -> None: title=title, completed_at=datetime.now(UTC), expires_at=datetime.now(UTC) + timedelta(hours=settings.file_expire_hours), - ) + ), ) await db.commit() if result.rowcount == 0: diff --git a/worker/main.py b/worker/main.py index 96819482..93699e6a 100644 --- a/worker/main.py +++ b/worker/main.py @@ -80,7 +80,7 @@ async def _update_circuit_deferred_depth() -> None: depth = await redis_client.zcard("circuit_deferred_queue") CIRCUIT_DEFERRED_DEPTH.set(depth) except Exception: - pass + logger.debug("circuit_deferred_depth update skipped (non-critical)", exc_info=True) async def cleanup_expired_jobs() -> int: @@ -93,8 +93,9 @@ async def cleanup_expired_jobs() -> int: result = await db.execute( select(DownloadJob).where( - DownloadJob.expires_at < now, DownloadJob.status == "completed" - ) + DownloadJob.expires_at < now, + DownloadJob.status == "completed", + ), ) expired_jobs = result.scalars().all() @@ -111,17 +112,21 @@ async def cleanup_expired_jobs() -> int: ) continue - if os.path.exists(safe_path): + if os.path.exists(safe_path): # noqa: ASYNC240 — local filesystem, negligible latency try: os.remove(safe_path) logger.info( - "cleaned_up_expired_file", file_path=safe_path, job_id=str(job.id) + "cleaned_up_expired_file", + file_path=safe_path, + job_id=str(job.id), ) await db.delete(job) cleanup_count += 1 except OSError as e: logger.warning( - "failed_to_delete_expired_file", file_path=job.file_path, error=str(e) + "failed_to_delete_expired_file", + file_path=job.file_path, + error=str(e), ) else: logger.info("file_already_deleted", job_id=str(job.id), file_path=job.file_path) @@ -133,7 +138,10 @@ async def cleanup_expired_jobs() -> int: cleanup_count += 1 except Exception as db_err: logger.warning( - "failed_to_delete_db_row", job_id=job.id, error=str(db_err), exc_info=True + "failed_to_delete_db_row", + job_id=job.id, + error=str(db_err), + exc_info=True, ) try: @@ -156,7 +164,11 @@ async def _update_queue_depth() -> None: return dl + rt + cd """ total = await redis_client.eval( - lua_script, 3, "download_queue", "retry_queue", "circuit_deferred_queue" + lua_script, + 3, + "download_queue", + "retry_queue", + "circuit_deferred_queue", ) QUEUE_DEPTH.set(int(total)) except Exception as e: @@ -256,9 +268,9 @@ async def main() -> None: last_cleanup = datetime.now(UTC) - cleanup_interval try: - outbox_sync_interval_seconds = int(os.environ.get("OUTBOX_SYNC_INTERVAL_SECONDS", "30")) + outbox_sync_interval_seconds = int(os.environ.get("OUTBOX_SYNC_INTERVAL_SECONDS", "2")) except (ValueError, TypeError): - outbox_sync_interval_seconds = 30 + outbox_sync_interval_seconds = 2 outbox_sync_interval_seconds = max(1, min(outbox_sync_interval_seconds, 3600)) outbox_sync_interval = timedelta(seconds=outbox_sync_interval_seconds) last_outbox_sync = datetime.now(UTC) - outbox_sync_interval @@ -300,7 +312,12 @@ async def main() -> None: return #due_jobs """ moved_count = await redis_client.eval( - lua_script, 2, "retry_queue", "download_queue", now_ts, MAX_RETRY_BATCH + lua_script, + 2, + "retry_queue", + "download_queue", + now_ts, + MAX_RETRY_BATCH, ) if moved_count and moved_count > 0: logger.info("retry_jobs_moved", moved_count=moved_count) diff --git a/worker/outbox_relay.py b/worker/outbox_relay.py index 9572742d..b41a62df 100644 --- a/worker/outbox_relay.py +++ b/worker/outbox_relay.py @@ -1,17 +1,33 @@ -"""Transactional outbox relay for Redis queue recovery.""" +"""Transactional outbox relay for Redis queue recovery. + +Lifecycle: outbox rows are written with status='pending' inside the same DB +transaction that mutates the domain entity (DownloadJob). The relay claims +pending rows, pushes the corresponding Redis message, and transitions the row +to status='processed' (with processed_at). Rows are retained for observability +and reaped by ``cleanup_stale_outbox_entries`` after the retention window. + +This module is also the single source of truth for the +``OUTBOX_OLDEST_PENDING_SECONDS`` and ``OUTBOX_PENDING`` gauges so the +metrics reflect relay reality even if the relay is the only thing running. +""" import json from datetime import UTC, datetime, timedelta -from sqlalchemy import delete, select +from sqlalchemy import delete, func, select, update from core.database import get_async_session_factory from core.logging_config import get_logger +from core.metrics import OUTBOX_OLDEST_PENDING_SECONDS, OUTBOX_PENDING from core.models.outbox import Outbox from core.queue import push_to_download_queue, push_to_retry_queue logger = get_logger(__name__) +_PROCESSED_STATUS = "processed" +_FAILED_STATUS = "failed" +_PENDING_STATUS = "pending" + async def _retry_job_is_already_enqueued(job_id) -> bool: """Return whether a retry job already exists in Redis after a deduplicated push.""" @@ -24,22 +40,55 @@ async def _retry_job_is_already_enqueued(job_id) -> bool: return False +async def _update_staleness_metrics(db) -> None: + """Refresh OUTBOX_PENDING and OUTBOX_OLDEST_PENDING_SECONDS for observability. + + Reads are best-effort. A failure here must not abort the relay cycle. + """ + try: + pending_count = await db.scalar( + select(func.count()).where(Outbox.status == _PENDING_STATUS) + ) + OUTBOX_PENDING.set(float(pending_count or 0)) + + oldest = await db.scalar( + select(func.extract("epoch", func.now() - func.min(Outbox.created_at))).where( + Outbox.status == _PENDING_STATUS, + ), + ) + OUTBOX_OLDEST_PENDING_SECONDS.set(float(oldest or 0)) + except Exception as exc: + logger.warning("outbox_staleness_metric_update_failed", error=str(exc)) + + async def sync_outbox_to_queue(batch_size: int = 100) -> int: - """Sync pending outbox entries to Redis queue.""" + """Sync pending outbox entries to Redis queue and mark them processed. + + Returns the number of outbox entries successfully delivered to Redis. + + Crash-safety: the row's status transitions from 'pending' to 'processed' + in a single SQL UPDATE that runs only after the Redis push has been + confirmed by the queue helper (push_to_download_queue / push_to_retry_queue + return True only after Redis acknowledged the write). If the process dies + between the Redis push and the UPDATE commit, the next sync will re-deliver + the message — duplicates are prevented at the queue side by LREM+LPUSH + for download_queue and ZSCORE-based dedup for retry_queue. + """ session_factory = get_async_session_factory() synced = 0 async with session_factory() as db: claim_result = await db.execute( select(Outbox) - .where(Outbox.status == "pending") + .where(Outbox.status == _PENDING_STATUS) .order_by(Outbox.created_at) .limit(batch_size) - .with_for_update(skip_locked=True) + .with_for_update(skip_locked=True), ) entries = claim_result.scalars().all() if not entries: + await _update_staleness_metrics(db) return 0 processed_entry_ids = [] @@ -68,15 +117,28 @@ async def sync_outbox_to_queue(batch_size: int = 100) -> int: synced += 1 except Exception as e: logger.error( - "failed_to_enqueue_job_from_outbox", job_id=str(entry.job_id), error=str(e) + "failed_to_enqueue_job_from_outbox", + job_id=str(entry.job_id), + error=str(e), ) if processed_entry_ids: + now = datetime.now(UTC) + await db.execute( + update(Outbox) + .where(Outbox.id.in_(processed_entry_ids)) + .values(status=_PROCESSED_STATUS, processed_at=now), + ) try: - await db.execute(delete(Outbox).where(Outbox.id.in_(processed_entry_ids))) await db.commit() except Exception: await db.rollback() + logger.error( + "outbox_processed_status_update_failed", + count=len(processed_entry_ids), + ) + + await _update_staleness_metrics(db) if synced > 0: logger.info("synced_outbox_entries_to_queue", count=synced) @@ -85,15 +147,22 @@ async def sync_outbox_to_queue(batch_size: int = 100) -> int: async def cleanup_stale_outbox_entries(hours: int = 24) -> int: - """Delete old terminal outbox entries while retaining pending crash-recovery rows.""" + """Delete terminal outbox rows older than ``hours`` (default 24). + + Terminal rows are those whose status is ``processed`` or ``failed`` (the + latter is reserved for relay-level delivery failures). ``pending`` rows + are never reaped here — they are the crash-recovery set and must survive + until the relay delivers them. + """ session_factory = get_async_session_factory() cutoff = datetime.now(UTC) - timedelta(hours=hours) async with session_factory() as db: result = await db.execute( delete(Outbox).where( - Outbox.created_at < cutoff, - Outbox.status.in_(["completed", "failed"]), - ) + Outbox.processed_at.is_not(None), + Outbox.processed_at < cutoff, + Outbox.status.in_([_PROCESSED_STATUS, _FAILED_STATUS]), + ), ) await db.commit() count = int(result.rowcount or 0) diff --git a/worker/processor.py b/worker/processor.py index 7e2067ed..b857e949 100644 --- a/worker/processor.py +++ b/worker/processor.py @@ -76,7 +76,7 @@ async def _drain_circuit_deferred(max_batch: int = 10) -> int: result = await db.execute( update(DownloadJob) .where(DownloadJob.id == job_id_str, DownloadJob.status == "deferred") - .values(status="pending", updated_at=datetime.now(UTC)) + .values(status="pending", updated_at=datetime.now(UTC)), ) if result.rowcount != 1: await db.rollback() @@ -196,7 +196,7 @@ async def _handle_circuit_open(db, active_job_id: UUID, cb_error: CircuitBreaker f"deferred until recovery (cooldown: {cb_error.reset_timeout}s)", error_category="transient", updated_at=datetime.now(UTC), - ) + ), ) await db.commit() if result.rowcount == 0: diff --git a/worker/retry_scheduler.py b/worker/retry_scheduler.py index d94dd832..248e3766 100644 --- a/worker/retry_scheduler.py +++ b/worker/retry_scheduler.py @@ -5,7 +5,7 @@ from dataclasses import dataclass from datetime import UTC, datetime, timedelta -from sqlalchemy import delete, select, update +from sqlalchemy import select, update from sqlalchemy.ext.asyncio import AsyncSession from app.services.error_classifier import ( @@ -55,7 +55,7 @@ def evaluate(job: DownloadJob, error: BaseException) -> RetryDecision: error_str = str(error) classification = classify_error(error_str) category = classification.category - job_max_retries = job.max_retries if job.max_retries else 3 + job_max_retries = job.max_retries if job.max_retries is not None else 3 effective_max = min(CATEGORY_POLICIES[category].max_retries, job_max_retries) ERROR_CLASSIFICATION.labels(category=category.value).inc() @@ -139,7 +139,7 @@ async def schedule_retry(db: AsyncSession, job: DownloadJob, decision: RetryDeci last_error=decision.accumulated_error, error_category=decision.category.value, updated_at=datetime.now(UTC), - ) + ), ) if int(getattr(retry_result, "rowcount", 0) or 0) == 0: await db.rollback() @@ -158,7 +158,7 @@ async def schedule_retry(db: AsyncSession, job: DownloadJob, decision: RetryDeci "retry_count": job.retry_count + 1, "category": decision.category.value, "next_retry_at": decision.next_retry_at.isoformat(), - } + }, ), status="pending", ) @@ -169,12 +169,19 @@ async def schedule_retry(db: AsyncSession, job: DownloadJob, decision: RetryDeci try: enqueued = await push_to_retry_queue(active_job_id, decision.next_retry_at.timestamp()) if enqueued: - await db.execute(delete(Outbox).where(Outbox.id == outbox_entry.id)) + from sqlalchemy import update as sqlalchemy_update + + await db.execute( + sqlalchemy_update(Outbox) + .where(Outbox.id == outbox_entry.id) + .values(status="processed", processed_at=datetime.now(UTC)), + ) await db.commit() RETRIES_TOTAL.labels(category=decision.category.value).inc() else: logger.error("job_failed_to_enqueue_for_retry", job_id=str(active_job_id)) except Exception as enqueue_error: + await db.rollback() logger.error( "job_failed_to_enqueue_for_retry", job_id=str(active_job_id), diff --git a/worker/zombie_sweeper.py b/worker/zombie_sweeper.py index e428a80d..96eea63a 100644 --- a/worker/zombie_sweeper.py +++ b/worker/zombie_sweeper.py @@ -53,6 +53,7 @@ async def requeue_stuck_jobs(timeout_minutes: int = 15) -> int: Returns: Number of jobs requeued. + """ # Chaos-aware: if the chaos zombie key is active, clamp to 1 minute max # so the demo doesn't wait 15 minutes for recovery visualization @@ -86,7 +87,7 @@ async def requeue_stuck_jobs(timeout_minutes: int = 15) -> int: status="pending", updated_at=datetime.now(UTC), ) - .returning(DownloadJob.id) + .returning(DownloadJob.id), ) requeued_ids = result.scalars().all() @@ -104,7 +105,7 @@ async def requeue_stuck_jobs(timeout_minutes: int = 15) -> int: event_type="zombie_recovery", payload=json.dumps({"recovered_at": now.isoformat()}), status="pending", - ) + ), ) try: