Merge pull request #79 from christianlouis/copilot/fix-process-enabled-mail-accounts
Fix "Exception terminating connection" in Celery async tasks
This commit is contained in:
@@ -18,6 +18,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||
- `auth_service.py` `get_google_user_info` now returns `access_token`, `refresh_token`, `expires_in`, and `scope` alongside user info so the login endpoint can persist Gmail credentials in the same request.
|
||||
|
||||
### Fixed
|
||||
- Fixed `Exception terminating connection` error logged by Celery workers after every task run. The error was caused by `asyncio.run()` closing the event loop while the asyncpg connection pool still held open idle connections. The fix calls `await engine.dispose()` inside the task's `_run()` coroutine (within the same event loop) so all pooled connections are closed cleanly before the loop is torn down.
|
||||
- Gmail API `verify_access()` returning 403 for tokens that lacked a read-capable scope: added `gmail.readonly` to all scope lists.
|
||||
|
||||
- Fixed three ESLint errors that caused CI to fail: removed unused `_setUser` store binding and unused `useAuthStore` import from `login/page.tsx`; replaced unused `_err` catch binding with a bare `catch {}` in `login/page.tsx`; removed a `useEffect` in `settings/page.tsx` that called `setProfileForm` synchronously (flagged by `react-hooks/set-state-in-effect`) — the effect was redundant because `useState` already initialises the form from the auth store's `user` object, which is the same value passed as `initialData` to `useQuery`.
|
||||
|
||||
@@ -8,7 +8,7 @@ from celery import Task
|
||||
import logging
|
||||
|
||||
from app.workers.celery_app import celery_app
|
||||
from app.core.database import async_session_maker
|
||||
from app.core.database import async_session_maker, engine
|
||||
from app.core.security import decrypt_credential, encrypt_credential
|
||||
from app.models.database_models import (
|
||||
MailAccount,
|
||||
@@ -34,8 +34,20 @@ class AsyncTask(Task):
|
||||
|
||||
def __call__(self, *args, **kwargs):
|
||||
"""Run async task in event loop"""
|
||||
# Use asyncio.run() for better event loop management
|
||||
return asyncio.run(self.run(*args, **kwargs))
|
||||
|
||||
async def _run():
|
||||
try:
|
||||
return await self.run(*args, **kwargs)
|
||||
finally:
|
||||
# Dispose the connection pool before the event loop closes.
|
||||
# Each asyncio.run() creates a fresh event loop; if pooled
|
||||
# asyncpg connections are still open when the loop is torn
|
||||
# down, asyncpg raises "Exception terminating connection".
|
||||
# Disposing the engine here closes those connections cleanly
|
||||
# inside the same loop, before asyncio.run() shuts it down.
|
||||
await engine.dispose()
|
||||
|
||||
return asyncio.run(_run())
|
||||
|
||||
|
||||
@celery_app.task(base=AsyncTask, name="app.workers.tasks.process_mail_account")
|
||||
|
||||
@@ -16,6 +16,7 @@ Comprehensive task breakdown for repository improvements and production readines
|
||||
- [ ] Enable rate limiting per user/tier
|
||||
- [x] Fix bare exception handlers throughout codebase
|
||||
- [x] Update datetime usage to timezone-aware (`DateTime(timezone=True)` columns and `lambda: datetime.now(timezone.utc)` defaults; fixes `DBAPIError` from asyncpg on timezone-naive columns)
|
||||
- [x] Fix `Exception terminating connection` in Celery workers: call `await engine.dispose()` inside task coroutine so pooled asyncpg connections are closed before the event loop is torn down
|
||||
- [ ] Validate redirect_uri to prevent open redirect vulnerabilities
|
||||
- [ ] Add per-user random salt for encryption (currently deterministic)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user