TASK-028: Add exception logging to fire-and-forget asyncio.create_task()
- Create logged_task() helper in backend/app/utils/async_utils.py to wrap fire-and-forget coroutines with exception logging - Ensures unhandled task exceptions are always logged to structlog instead of silently discarded (Python 3.11+ RuntimeWarning) - Update ban_service.py to use logged_task() for geo_cache.lookup_batch() background resolution - Add comprehensive tests for logged_task() in test_async_utils.py - Document fire-and-forget task conventions in Backend-Development.md The logged_task() wrapper catches any exception raised in a background task, logs it with full traceback context and task name, and never re-raises. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
@@ -309,6 +309,59 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
|
|||||||
await app.state.db.close()
|
await app.state.db.close()
|
||||||
```
|
```
|
||||||
|
|
||||||
|
### Fire-and-Forget Background Tasks
|
||||||
|
|
||||||
|
When you need to spawn a background task that runs independently without waiting for the result, use `asyncio.create_task()` with the `logged_task()` helper from `app.utils.async_utils`. This ensures exceptions in background tasks are always logged and never silently discarded.
|
||||||
|
|
||||||
|
**Why this matters:** In Python 3.11+, unhandled exceptions in fire-and-forget tasks become silent `RuntimeWarning`s. Without logging, background errors (network failures, database writes, API timeouts) become invisible in structured logs and are extremely hard to debug.
|
||||||
|
|
||||||
|
**Pattern:**
|
||||||
|
```python
|
||||||
|
from app.utils.async_utils import logged_task
|
||||||
|
|
||||||
|
# Bad — exceptions are silently discarded
|
||||||
|
asyncio.create_task(some_background_work())
|
||||||
|
|
||||||
|
# Good — exceptions are logged
|
||||||
|
asyncio.create_task(
|
||||||
|
logged_task(some_background_work(), "task_name"),
|
||||||
|
name="task_name",
|
||||||
|
)
|
||||||
|
```
|
||||||
|
|
||||||
|
The `logged_task()` wrapper:
|
||||||
|
- Wraps your coroutine to catch any exception
|
||||||
|
- Logs the exception with `log.exception()` (structlog automatically captures the traceback)
|
||||||
|
- Adds `task_name` to the structured log context
|
||||||
|
- Never re-raises — it's safe to use with `asyncio.create_task()`
|
||||||
|
|
||||||
|
Example:
|
||||||
|
```python
|
||||||
|
import asyncio
|
||||||
|
from app.utils.async_utils import logged_task
|
||||||
|
import structlog
|
||||||
|
|
||||||
|
log = structlog.get_logger()
|
||||||
|
|
||||||
|
async def geo_lookup_batch(ips: list[str]) -> None:
|
||||||
|
"""Look up geolocation data for IPs asynchronously."""
|
||||||
|
try:
|
||||||
|
for ip in ips:
|
||||||
|
# May timeout, fail network call, or fail DB write
|
||||||
|
location = await lookup_ip_location(ip)
|
||||||
|
await db.execute(INSERT_GEO_SQL, (ip, location))
|
||||||
|
await db.commit()
|
||||||
|
except Exception:
|
||||||
|
# All exceptions are automatically logged by logged_task() wrapper
|
||||||
|
raise
|
||||||
|
|
||||||
|
# In your request handler or service:
|
||||||
|
asyncio.create_task(
|
||||||
|
logged_task(geo_lookup_batch(uncached_ips), "geo_cache_batch"),
|
||||||
|
name="geo_cache_batch",
|
||||||
|
)
|
||||||
|
```
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
## 6.1 Database Query Conventions
|
## 6.1 Database Query Conventions
|
||||||
|
|||||||
@@ -1,36 +1,3 @@
|
|||||||
## TASK-027 — Debug compose hardcodes a publicly known weak session secret
|
|
||||||
|
|
||||||
**Severity:** Medium
|
|
||||||
|
|
||||||
### Where found
|
|
||||||
`Docker/compose.debug.yml` line ~63:
|
|
||||||
```yaml
|
|
||||||
BANGUI_SESSION_SECRET: "${BANGUI_SESSION_SECRET:-dev-secret-do-not-use-in-production}"
|
|
||||||
```
|
|
||||||
|
|
||||||
### Why this is needed
|
|
||||||
The fallback value `dev-secret-do-not-use-in-production` is now publicly visible in the repository. If `compose.debug.yml` is used in any environment where `BANGUI_SESSION_SECRET` is not set (e.g., a CI environment or a staging server that uses the debug compose file), all session tokens can be forged by anyone who has seen this repository.
|
|
||||||
|
|
||||||
### Goal
|
|
||||||
Remove the insecure default. Require the secret to be set explicitly before the container starts.
|
|
||||||
|
|
||||||
### What to do
|
|
||||||
1. Change to `BANGUI_SESSION_SECRET: "${BANGUI_SESSION_SECRET:?BANGUI_SESSION_SECRET must be set — generate with: python -c 'import secrets; print(secrets.token_hex(32))'}"`.
|
|
||||||
2. Create a `.env.example` file at the project root with placeholder values and generation instructions.
|
|
||||||
3. Add `.env` to `.gitignore` (verify it is already there).
|
|
||||||
|
|
||||||
### Possible traps and issues
|
|
||||||
- This will break `docker compose -f Docker/compose.debug.yml up` without a `.env` file. Add a clear error message and setup instructions to the README or `Instructions.md`.
|
|
||||||
- `docker-compose.yml` (the legacy file) already uses the `:?` pattern — follow the same approach.
|
|
||||||
|
|
||||||
### Docs changes needed
|
|
||||||
- `Instructions.md` — add first-run setup instructions for the `.env` file.
|
|
||||||
|
|
||||||
### Doc references
|
|
||||||
- [Instructions.md](Instructions.md) — developer setup
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## TASK-028 — Fire-and-forget `asyncio.create_task()` silently discards exceptions
|
## TASK-028 — Fire-and-forget `asyncio.create_task()` silently discards exceptions
|
||||||
|
|
||||||
**Severity:** Low
|
**Severity:** Low
|
||||||
|
|||||||
@@ -41,6 +41,7 @@ from app.models.ban import (
|
|||||||
from app.repositories import fail2ban_db_repo
|
from app.repositories import fail2ban_db_repo
|
||||||
from app.repositories import history_archive_repo as default_history_archive_repo
|
from app.repositories import history_archive_repo as default_history_archive_repo
|
||||||
from app.services.fail2ban_metadata_service import default_fail2ban_metadata_service
|
from app.services.fail2ban_metadata_service import default_fail2ban_metadata_service
|
||||||
|
from app.utils.async_utils import logged_task
|
||||||
from app.utils.constants import (
|
from app.utils.constants import (
|
||||||
DEFAULT_PAGE_SIZE,
|
DEFAULT_PAGE_SIZE,
|
||||||
FAIL2BAN_SOCKET_TIMEOUT,
|
FAIL2BAN_SOCKET_TIMEOUT,
|
||||||
@@ -611,8 +612,11 @@ async def bans_by_country(
|
|||||||
if geo_cache is not None:
|
if geo_cache is not None:
|
||||||
# Fire-and-forget: lookup_batch handles rate-limiting / retries.
|
# Fire-and-forget: lookup_batch handles rate-limiting / retries.
|
||||||
# The dirty-set flush task persists results to the DB.
|
# The dirty-set flush task persists results to the DB.
|
||||||
asyncio.create_task( # noqa: RUF006
|
asyncio.create_task(
|
||||||
geo_cache.lookup_batch(uncached, http_session, db=app_db),
|
logged_task(
|
||||||
|
geo_cache.lookup_batch(uncached, http_session, db=app_db),
|
||||||
|
"geo_bans_by_country",
|
||||||
|
),
|
||||||
name="geo_bans_by_country",
|
name="geo_bans_by_country",
|
||||||
)
|
)
|
||||||
elif geo_enricher is not None and unique_ips:
|
elif geo_enricher is not None and unique_ips:
|
||||||
|
|||||||
@@ -8,12 +8,17 @@ from __future__ import annotations
|
|||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import functools
|
import functools
|
||||||
|
from collections.abc import Callable, Coroutine
|
||||||
from concurrent.futures import ThreadPoolExecutor
|
from concurrent.futures import ThreadPoolExecutor
|
||||||
from typing import Callable, ParamSpec, TypeVar
|
from typing import Any, ParamSpec, TypeVar
|
||||||
|
|
||||||
|
import structlog
|
||||||
|
|
||||||
P = ParamSpec("P")
|
P = ParamSpec("P")
|
||||||
T = TypeVar("T")
|
T = TypeVar("T")
|
||||||
|
|
||||||
|
log: structlog.stdlib.BoundLogger = structlog.get_logger()
|
||||||
|
|
||||||
DEFAULT_BLOCKING_EXECUTOR: ThreadPoolExecutor = ThreadPoolExecutor(
|
DEFAULT_BLOCKING_EXECUTOR: ThreadPoolExecutor = ThreadPoolExecutor(
|
||||||
max_workers=16,
|
max_workers=16,
|
||||||
thread_name_prefix="bangui-blocking",
|
thread_name_prefix="bangui-blocking",
|
||||||
@@ -43,3 +48,27 @@ async def run_blocking(
|
|||||||
func = functools.partial(func, *args, **kwargs)
|
func = functools.partial(func, *args, **kwargs)
|
||||||
return await loop.run_in_executor(executor, func)
|
return await loop.run_in_executor(executor, func)
|
||||||
return await loop.run_in_executor(executor, func, *args)
|
return await loop.run_in_executor(executor, func, *args)
|
||||||
|
|
||||||
|
|
||||||
|
async def logged_task(
|
||||||
|
coro: Coroutine[Any, Any, Any],
|
||||||
|
name: str,
|
||||||
|
) -> None:
|
||||||
|
"""Execute a coroutine with automatic exception logging.
|
||||||
|
|
||||||
|
Wraps fire-and-forget tasks to ensure exceptions are always logged and
|
||||||
|
do not become unhandled task exceptions. Use with asyncio.create_task():
|
||||||
|
|
||||||
|
asyncio.create_task(
|
||||||
|
logged_task(some_coroutine(), "task_name"),
|
||||||
|
name="task_name"
|
||||||
|
)
|
||||||
|
|
||||||
|
Args:
|
||||||
|
coro: Coroutine to execute.
|
||||||
|
name: Task name for logging context.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
await coro
|
||||||
|
except Exception: # noqa: BLE001
|
||||||
|
log.exception("background_task_failed", task_name=name)
|
||||||
|
|||||||
@@ -5,8 +5,12 @@ from __future__ import annotations
|
|||||||
import asyncio
|
import asyncio
|
||||||
import time
|
import time
|
||||||
from concurrent.futures import ThreadPoolExecutor
|
from concurrent.futures import ThreadPoolExecutor
|
||||||
|
from unittest import mock
|
||||||
|
|
||||||
from app.utils.async_utils import run_blocking
|
import pytest
|
||||||
|
import structlog
|
||||||
|
|
||||||
|
from app.utils.async_utils import logged_task, run_blocking
|
||||||
|
|
||||||
|
|
||||||
async def test_run_blocking_executes_callable_in_thread() -> None:
|
async def test_run_blocking_executes_callable_in_thread() -> None:
|
||||||
@@ -46,3 +50,67 @@ async def test_run_blocking_does_not_block_event_loop() -> None:
|
|||||||
await asyncio.sleep(0)
|
await asyncio.sleep(0)
|
||||||
result = await task
|
result = await task
|
||||||
assert result == "done"
|
assert result == "done"
|
||||||
|
|
||||||
|
|
||||||
|
async def test_logged_task_awaits_coroutine() -> None:
|
||||||
|
"""logged_task should await and complete the coroutine."""
|
||||||
|
|
||||||
|
async def dummy_coro() -> str:
|
||||||
|
await asyncio.sleep(0)
|
||||||
|
return "result"
|
||||||
|
|
||||||
|
with mock.patch("app.utils.async_utils.log") as mock_log:
|
||||||
|
await logged_task(dummy_coro(), "test_task")
|
||||||
|
mock_log.exception.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_logged_task_catches_exception_and_logs() -> None:
|
||||||
|
"""logged_task should catch exceptions and log them with task_name context."""
|
||||||
|
|
||||||
|
class CustomError(Exception):
|
||||||
|
pass
|
||||||
|
|
||||||
|
async def failing_coro() -> None:
|
||||||
|
raise CustomError("task failed")
|
||||||
|
|
||||||
|
with mock.patch("app.utils.async_utils.log") as mock_log:
|
||||||
|
await logged_task(failing_coro(), "failing_task")
|
||||||
|
mock_log.exception.assert_called_once_with(
|
||||||
|
"background_task_failed",
|
||||||
|
task_name="failing_task",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def test_logged_task_with_asyncio_create_task() -> None:
|
||||||
|
"""logged_task should work correctly when wrapped with asyncio.create_task."""
|
||||||
|
|
||||||
|
results: list[str] = []
|
||||||
|
|
||||||
|
async def background_work() -> None:
|
||||||
|
await asyncio.sleep(0.01)
|
||||||
|
results.append("done")
|
||||||
|
|
||||||
|
with mock.patch("app.utils.async_utils.log"):
|
||||||
|
task = asyncio.create_task(
|
||||||
|
logged_task(background_work(), "bg_task"),
|
||||||
|
name="bg_task",
|
||||||
|
)
|
||||||
|
await task
|
||||||
|
assert results == ["done"]
|
||||||
|
|
||||||
|
|
||||||
|
async def test_logged_task_preserves_exception_info() -> None:
|
||||||
|
"""logged_task should preserve traceback when logging the exception."""
|
||||||
|
|
||||||
|
async def failing_coro() -> None:
|
||||||
|
raise ValueError("original error message")
|
||||||
|
|
||||||
|
with mock.patch("app.utils.async_utils.log") as mock_log:
|
||||||
|
await logged_task(failing_coro(), "test_task")
|
||||||
|
mock_log.exception.assert_called_once()
|
||||||
|
# Verify the exception context is logged (structlog.exception captures
|
||||||
|
# the traceback automatically)
|
||||||
|
args, kwargs = mock_log.exception.call_args
|
||||||
|
assert args[0] == "background_task_failed"
|
||||||
|
assert kwargs["task_name"] == "test_task"
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user