fix: api announcements are not deleted from user subscriptions (#1748)

Author: davidgamez

.github/workflows/api-deploy-developer.yml

@@ -225,7 +225,7 @@ jobs:
             --update-secrets  "FEEDS_DATABASE_URL=DEV_FEEDS_DATABASE_URL:latest,USERS_DATABASE_URL=DEV_USERS_DATABASE_URL:latest,S2S_JWT_SECRET=DEV_S2S_JWT_SECRET:latest" \
             --set-env-vars    "PROJECT_ID=${{ env.PROJECT_ID }}" \
             --vpc-connector   "projects/${{ env.VPC_CONNECTOR_PROJECT }}/locations/${{ env.REGION }}/connectors/${{ env.VPC_CONNECTOR }}" \
-            --vpc-egress      all
+            --vpc-egress      private-ranges-only
 
       - name: Get service URL
         id: get_url

api/.openapi-generator/FILES

@@ -47,6 +47,8 @@ src/user_service/impl/__init__.py
 src/user_service_gen/apis/__init__.py
 src/user_service_gen/apis/notifications_api.py
 src/user_service_gen/apis/notifications_api_base.py
+src/user_service_gen/apis/subscriptions_api.py
+src/user_service_gen/apis/subscriptions_api_base.py
 src/user_service_gen/apis/users_api.py
 src/user_service_gen/apis/users_api_base.py
 src/user_service_gen/main.py
@@ -60,5 +62,3 @@ src/user_service_gen/models/update_notification_subscription_request.py
 src/user_service_gen/models/update_user_request.py
 src/user_service_gen/models/user_profile.py
 src/user_service_gen/security_api.py
-src/user_service_gen/apis/subscriptions_api.py
-src/user_service_gen/apis/subscriptions_api_base.py

api/src/main.py

@@ -35,6 +35,7 @@
 
 from middleware.request_context_middleware import RequestContextMiddleware
 from utils.logger import global_logging_setup
+from utils.route_offload import offload_blocking_routes
 
 app = FastAPI(
     title="Mobility Data Catalog API",
@@ -58,9 +59,12 @@
 app.include_router(MetadataApiRouter)
 app.include_router(SearchApiRouter)
 app.include_router(LicensesApiRouter)
-app.include_router(UsersApiRouter)
-app.include_router(NotificationsApiRouter)
-app.include_router(SubscriptionsApiRouter)
+# The user-service routes are generated as ``async def`` but call blocking
+# synchronous impls (database + Brevo HTTP). Offload them to the threadpool so a
+# slow Brevo call cannot freeze the event loop for every other request.
+app.include_router(offload_blocking_routes(UsersApiRouter))
+app.include_router(offload_blocking_routes(NotificationsApiRouter))
+app.include_router(offload_blocking_routes(SubscriptionsApiRouter))
 
 
 @app.on_event("startup")

api/src/shared/common/brevo.py

@@ -30,9 +30,21 @@
 from enum import Enum
 
 import sib_api_v3_sdk
+import urllib3
 
 logger = logging.getLogger(__name__)
 
+# (connect, read) timeout in seconds for every Brevo API call. Without this the SDK uses no
+# timeout, so when Brevo is unreachable the underlying urllib3 retries hang for a long time while
+# holding the caller's DB connection from the pool, which can exhaust the pool and make requests
+# appear "stuck". A short connect timeout makes the failure fast and bounded.
+BREVO_REQUEST_TIMEOUT = (3.05, 10)
+
+# Do not retry on connection failures. The Brevo call runs inside an async FastAPI route (which
+# executes on the event loop), so a long retry loop would block the whole API, not just this
+# request. With no retries, an unreachable Brevo fails within ~the connect timeout above.
+_BREVO_RETRIES = urllib3.Retry(total=0, connect=0, read=0, redirect=0, status=0)
+
 
 class BrevoSubscriptionStatus(Enum):
     SUBSCRIBED = "subscribed"
@@ -47,7 +59,10 @@ def _get_contacts_api() -> "sib_api_v3_sdk.ContactsApi":
         raise RuntimeError("BREVO_API_KEY environment variable is not set")
     configuration = sib_api_v3_sdk.Configuration()
     configuration.api_key["api-key"] = api_key
-    return sib_api_v3_sdk.ContactsApi(sib_api_v3_sdk.ApiClient(configuration))
+    api = sib_api_v3_sdk.ContactsApi(sib_api_v3_sdk.ApiClient(configuration))
+    # Disable urllib3 retries so a connection failure fails fast instead of looping.
+    api.api_client.rest_client.pool_manager.connection_pool_kw["retries"] = _BREVO_RETRIES
+    return api
 
 
 def get_announcements_list_id() -> int:
@@ -71,15 +86,20 @@ def add_contact_to_list(email: str, list_id: int, subscription_id: str) -> None:
             attributes={"MDB_SUBSCRIPTION_ID": subscription_id},
             list_ids=[list_id],
             update_enabled=True,
-        )
+        ),
+        _request_timeout=BREVO_REQUEST_TIMEOUT,
     )
 
 
 def remove_contact_from_list(email: str, list_id: int) -> None:
     """Remove a Brevo contact from the list. No-op if the contact is not on the list."""
     api = _get_contacts_api()
     try:
-        api.remove_contact_from_list(list_id, sib_api_v3_sdk.RemoveContactFromList(emails=[email]))
+        api.remove_contact_from_list(
+            list_id,
+            sib_api_v3_sdk.RemoveContactFromList(emails=[email]),
+            _request_timeout=BREVO_REQUEST_TIMEOUT,
+        )
     except sib_api_v3_sdk.rest.ApiException as exc:
         # 400 "Contact already removed from list" / 404 contact-not-found are idempotent no-ops.
         if exc.status in (400, 404):
@@ -107,7 +127,7 @@ def get_contact_subscription_status(
     api = _get_contacts_api()
 
     try:
-        contact = api.get_contact_info(email)
+        contact = api.get_contact_info(email, _request_timeout=BREVO_REQUEST_TIMEOUT)
     except sib_api_v3_sdk.rest.ApiException as exc:
         if exc.status == 404:
             return BrevoSubscriptionStatus.NOT_FOUND

api/src/shared/database/users_database.py

@@ -20,8 +20,8 @@
 from contextlib import contextmanager
 
 from dotenv import load_dotenv
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy import create_engine, event
+from sqlalchemy.orm import sessionmaker, mapper
 
 from shared.common.logging_utils import get_env_logging_level
 
@@ -115,3 +115,24 @@ def wrapper(*args, **kwargs):
 
     wrapper.__wrapped__ = func
     return wrapper
+
+
+@event.listens_for(mapper, "mapper_configured")
+def _configure_users_passive_deletes(mapper_, class_):
+    """Enable ``passive_deletes`` on ``NotificationSubscription.notification_logs``.
+
+    Deleting a subscription then relies on the database ``ON DELETE CASCADE`` to remove its
+    ``notification_log`` rows, instead of SQLAlchemy trying to NULL the NOT NULL
+    ``notification_log.subscription_id`` (which would raise a NotNullViolation).
+
+    This mirrors the ``set_cascade`` mechanism in ``shared.database.database`` but is scoped to
+    the users models, so ``users_database`` stays independent of the feeds ``database_gen``.
+
+    The mapper is matched by class name rather than by importing the model, so importing this
+    module does not require ``shared.users_database_gen`` to be present (functions that symlink
+    ``shared`` without the users models must still be able to import ``users_database``).
+    """
+    if class_.__name__ == "NotificationSubscription" and "notification_logs" in mapper_.relationships:
+        rel = mapper_.relationships["notification_logs"]
+        rel.cascade = "all, delete-orphan"
+        rel.passive_deletes = True

api/src/shared/db_models/notification_subscription_impl.py

@@ -19,6 +19,5 @@ def from_orm(cls, sub: NotificationSubscriptionOrm | None) -> NotificationSubscr
             user_id=sub.user_id,
             notification_id=sub.notification_type_id,
             active=sub.active,
-            last_notified_at=sub.last_notified_at,
             created_at=sub.created_at,
         )

api/src/user_service/impl/subscription_helpers.py

@@ -20,6 +20,7 @@
 from fastapi import HTTPException
 
 import sib_api_v3_sdk
+import urllib3
 from shared.common.brevo import add_contact_to_list, get_announcements_list_id, remove_contact_from_list
 
 logger = logging.getLogger(__name__)
@@ -34,6 +35,8 @@ def sync_announcements(email: str, subscribe: bool, subscription_id: str | None
             add_contact_to_list(email, get_announcements_list_id(), subscription_id)
         else:
             remove_contact_from_list(email, get_announcements_list_id())
-    except (RuntimeError, sib_api_v3_sdk.rest.ApiException) as exc:
+    except (RuntimeError, sib_api_v3_sdk.rest.ApiException, urllib3.exceptions.HTTPError, OSError) as exc:
+        # urllib3.exceptions.HTTPError / OSError cover connection failures and timeouts (e.g. Brevo
+        # unreachable), so the request fails fast with a 502 instead of hanging on retries.
         logger.error("Brevo sync failed for %s: %s", email, exc)
         raise HTTPException(status_code=502, detail="Failed to sync subscription with email provider.")

api/src/user_service/impl/subscriptions_api_impl.py

@@ -22,7 +22,10 @@
     AppUser,
     NotificationSubscription as NotificationSubscriptionOrm,
 )
-from user_service.impl.subscription_helpers import ANNOUNCEMENTS_NOTIFICATION_TYPE_ID, sync_announcements
+from user_service.impl.subscription_helpers import (
+    ANNOUNCEMENTS_NOTIFICATION_TYPE_ID,
+    sync_announcements,
+)
 from user_service_gen.apis.subscriptions_api_base import BaseSubscriptionsApi
 from user_service_gen.models.notification_subscription import NotificationSubscription
 
@@ -42,14 +45,24 @@ def get_subscription(self, id: str, db_session=None) -> NotificationSubscription
 
     @with_users_db_session
     def delete_subscription(self, id: str, db_session=None) -> None:
+        """Removes a subscription by ID.
+
+        The announcements subscription cannot be deleted; it is disabled instead.
+        """
         sub = db_session.get(NotificationSubscriptionOrm, id)
         if sub is None:
             raise HTTPException(status_code=404, detail="Subscription not found.")
 
         if sub.notification_type_id == ANNOUNCEMENTS_NOTIFICATION_TYPE_ID:
             user = db_session.get(AppUser, sub.user_id)
-            if user is not None:
-                sync_announcements(user.email, subscribe=False)
-
-        db_session.delete(sub)
+            email = user.email if user is not None else None
+            # Release the pooled DB connection before the (potentially slow) Brevo call so a slow
+            # or unreachable provider never holds a connection while we talk to it. The read above
+            # took no row locks, so committing here only returns the connection to the pool.
+            db_session.commit()
+            if email is not None:
+                sync_announcements(email, subscribe=False)
+            sub.active = False
+        else:
+            db_session.delete(sub)
         db_session.flush()

api/src/user_service/impl/users_api_impl.py

@@ -34,7 +34,10 @@
     NotificationSubscription as NotificationSubscriptionOrm,
     NotificationType,
 )
-from user_service.impl.subscription_helpers import ANNOUNCEMENTS_NOTIFICATION_TYPE_ID, sync_announcements
+from user_service.impl.subscription_helpers import (
+    ANNOUNCEMENTS_NOTIFICATION_TYPE_ID,
+    sync_announcements,
+)
 from user_service_gen.apis.users_api_base import BaseUsersApi
 from user_service_gen.models.create_notification_subscription_request import (
     CreateNotificationSubscriptionRequest,
@@ -192,15 +195,23 @@ def update_user_subscription(
 
     @with_users_db_session
     def delete_user_subscription(self, id: str, db_session=None) -> None:
-        """Removes a notification subscription by ID."""
+        """Removes a notification subscription by ID.
+
+        The announcements subscription cannot be deleted; it is disabled instead.
+        """
         user_id = self._require_user_id()
         sub = self._get_owned_subscription(db_session, id, user_id)
 
         if sub.notification_type_id == ANNOUNCEMENTS_NOTIFICATION_TYPE_ID:
-            user = db_session.get(AppUser, user_id)
-            sync_announcements(user.email, subscribe=False)
-
-        db_session.delete(sub)
+            email = db_session.get(AppUser, user_id).email
+            # Release the pooled DB connection before the (potentially slow) Brevo call so a slow
+            # or unreachable provider never holds a connection while we talk to it. The reads above
+            # took no row locks, so committing here only returns the connection to the pool.
+            db_session.commit()
+            sync_announcements(email, subscribe=False)
+            sub.active = False
+        else:
+            db_session.delete(sub)
         db_session.flush()
 
     # ── Helpers ──────────────────────────────────────────────────────────────

api/src/utils/route_offload.py

@@ -0,0 +1,72 @@
+#
+#   MobilityData 2024
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#   You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+#
+"""Utilities to keep blocking endpoints off the event loop.
+
+The code generated for the user service declares every route as ``async def`` but
+the route bodies are fully synchronous: they simply call into a synchronous impl
+that performs blocking I/O (database access and Brevo HTTP calls). FastAPI runs
+``async def`` path operations directly on the event loop, so a single slow Brevo
+call freezes the entire API for every concurrent request.
+
+``offload_blocking_routes`` rewrites those coroutine endpoints into plain
+synchronous functions. FastAPI then dispatches them through the anyio threadpool
+(see ``fastapi.routing.run_endpoint_function``), so blocking work no longer stalls
+the event loop and other requests stay responsive.
+"""
+
+import asyncio
+import functools
+
+from fastapi import APIRouter
+from fastapi.routing import APIRoute
+
+
+def _to_sync_endpoint(coroutine_endpoint):
+    """Wrap a coroutine endpoint whose body is synchronous into a plain function.
+
+    The generated user-service endpoints contain no ``await`` expressions, so the
+    underlying coroutine completes on the first step. Driving it once and reading
+    the ``StopIteration`` value yields the same result while letting FastAPI run
+    the wrapper in a threadpool instead of on the event loop.
+    """
+
+    @functools.wraps(coroutine_endpoint)
+    def sync_endpoint(*args, **kwargs):
+        coro = coroutine_endpoint(*args, **kwargs)
+        try:
+            coro.send(None)
+        except StopIteration as stop:
+            return stop.value
+        else:
+            coro.close()
+            raise RuntimeError(
+                f"Endpoint {coroutine_endpoint.__qualname__} awaited; it cannot be "
+                "offloaded to a threadpool. Remove it from offload_blocking_routes."
+            )
+
+    return sync_endpoint
+
+
+def offload_blocking_routes(router: APIRouter) -> APIRouter:
+    """Convert every coroutine route in ``router`` to run in the threadpool.
+
+    Must be called before ``app.include_router(router)`` so FastAPI builds the
+    route handlers from the synchronous endpoints.
+    """
+    for route in router.routes:
+        if isinstance(route, APIRoute) and asyncio.iscoroutinefunction(route.endpoint):
+            route.endpoint = _to_sync_endpoint(route.endpoint)
+    return router