Fix OG metadata scraping and improve workers

app/incoming_activities.py

@@ -3,7 +3,6 @@ import traceback
 from datetime import datetime
 from datetime import timedelta
 
-import httpx
 from loguru import logger
 from sqlalchemy import func
 from sqlalchemy import select
@@ -108,6 +107,7 @@ async def process_next_incoming_activity(
 
     next_activity.tries = next_activity.tries + 1
     next_activity.last_try = now()
+    await db_session.commit()
 
     if next_activity.ap_object and next_activity.sent_by_ap_actor_id:
         try:
@@ -120,13 +120,16 @@ async def process_next_incoming_activity(
                     ),
                     timeout=60,
                 )
-        except httpx.TimeoutException as exc:
-            url = exc._request.url if exc._request else None
-            logger.error(f"Failed, HTTP timeout when fetching {url}")
+        except asyncio.exceptions.TimeoutError:
+            logger.error("Activity took too long to process")
+            await db_session.rollback()
+            await db_session.refresh(next_activity)
             next_activity.error = traceback.format_exc()
             _set_next_try(next_activity)
         except Exception:
             logger.exception("Failed")
+            await db_session.rollback()
+            await db_session.refresh(next_activity)
             next_activity.error = traceback.format_exc()
             _set_next_try(next_activity)
         else:

app/utils/opengraph.py

@@ -1,12 +1,15 @@
 import asyncio
 import mimetypes
 import re
+import signal
+from concurrent.futures import TimeoutError
 from typing import Any
 from urllib.parse import urlparse
 
 import httpx
 from bs4 import BeautifulSoup  # type: ignore
 from loguru import logger
+from pebble import concurrent  # type: ignore
 from pydantic import BaseModel
 
 from app import activitypub as ap
@@ -29,7 +32,11 @@ class OpenGraphMeta(BaseModel):
     site_name: str
 
 
+@concurrent.process(timeout=5)
 def _scrap_og_meta(url: str, html: str) -> OpenGraphMeta | None:
+    # Prevent SIGTERM to bubble up to the worker
+    signal.signal(signal.SIGTERM, signal.SIG_IGN)
+
     soup = BeautifulSoup(html, "html5lib")
     ogs = {
         og.attrs["property"]: og.attrs.get("content")
@@ -58,6 +65,10 @@ def _scrap_og_meta(url: str, html: str) -> OpenGraphMeta | None:
     return OpenGraphMeta.parse_obj(raw)
 
 
+def scrap_og_meta(url: str, html: str) -> OpenGraphMeta | None:
+    return _scrap_og_meta(url, html).result()
+
+
 async def external_urls(
     db_session: AsyncSession,
     ro: ap_object.RemoteObject | OutboxObject | InboxObject,
@@ -126,7 +137,10 @@ async def _og_meta_from_url(url: str) -> OpenGraphMeta | None:
         return None
 
     try:
-        return _scrap_og_meta(url, resp.text)
+        return scrap_og_meta(url, resp.text)
+    except TimeoutError:
+        logger.info(f"Timed out when scraping OG meta for {url}")
+        return None
     except Exception:
         logger.info(f"Failed to scrap OG meta for {url}")
         return None

app/utils/workers.py

@@ -69,5 +69,5 @@ class Worker(Generic[T]):
         logger.info("stopping loop")
 
     async def _shutdown(self, sig: signal.Signals) -> None:
-        logger.info(f"Caught {signal=}")
+        logger.info(f"Caught {sig=}")
         self._stop_event.set()