diff --git a/backend/inbox/semaphore.py b/backend/inbox/semaphore.py new file mode 100644 index 0000000..d20aa29 --- /dev/null +++ b/backend/inbox/semaphore.py @@ -0,0 +1,41 @@ +"""Bounded Graph/mailbox HTTP concurrency for the upstream Email API.""" + +from __future__ import annotations + +import asyncio + + +class GraphSemaphore: + """Cap concurrent Graph-backed calls to one mailbox. + + A free slot starts the next waiter immediately. On 429 the wait happens + outside the semaphore so in-flight stays under the cap; a retry only runs + after re-acquiring. + """ + + def __init__(self,concurrency=4,max_retries=5): + self.concurrency=max(int(concurrency),1) + self.max_retries=max(int(max_retries),0) + self._slots=asyncio.Semaphore(self.concurrency) + + def _retry_after_seconds(self,response,attempt): + raw=(response.headers.get("Retry-After") or "").strip() + if raw: + try: + return max(float(raw),0.1) + except ValueError: + pass + # ~1–2s base, doubles each attempt, soft cap so a wedged mailbox cannot sleep forever. + return min(1.5*(2**attempt),30.0) + + async def get(self,client,url,*,params=None,headers=None): + """GET under this semaphore; on 429 release, wait, then retry.""" + attempt=0 + while True: + async with self._slots: + response=await client.get(url,params=params,headers=headers) + if response.status_code!=429 or attempt>=self.max_retries: + return response + wait=self._retry_after_seconds(response,attempt) + await asyncio.sleep(wait) + attempt+=1 diff --git a/backend/inbox/views.py b/backend/inbox/views.py index 1e07af5..874c9e6 100644 --- a/backend/inbox/views.py +++ b/backend/inbox/views.py @@ -14,6 +14,8 @@ from inbox.plugins import ( request_email_confirmation, send_mail, ) +from inbox.semaphore import GraphSemaphore + from inbox_classifier.decorators import is_manual_upload,triage_fields from inbox_classifier.execute_agent import classify_email from inbox_classifier.plugins import ( @@ -42,6 +44,7 @@ class Email: self.session=session self.get_url=os.getenv("EMAIL_URL") self.token=token or EMAIL_API_TOKEN + self.graph_slots=GraphSemaphore(concurrency=4,max_retries=5) self.pending_match_ids:list[str]=[] self.pending_confirmation_emails:list[str]=[] # Upstream ids the intake gate judged not to be job applications. They get a @@ -62,27 +65,32 @@ class Email: async def service_email(self,top,skip): async with httpx.AsyncClient() as client: try: - response=await client.get(f"{self.get_url}/emails", - params={"skip":skip,"top":top}, - headers={"Authorization":f"Bearer {self.token}"} + response=await self.graph_slots.get( + client,f"{self.get_url}/emails", + params={"skip":skip,"top":top}, + headers={"Authorization":f"Bearer {self.token}"}, ) if response.status_code==200: return response.json() else: raise HTTPException(status_code=response.status_code,detail=response.text) + except HTTPException: + raise except Exception as e: raise HTTPException(status_code=500,detail=str(e)) - + async def fetch_message(self,message_id): """GET /emails/{id} on the upstream Email API -> the Graph payload.""" async with httpx.AsyncClient() as client: - response=await client.get(f"{self.get_url}/emails/{message_id}", - headers={"Authorization":f"Bearer {self.token}"} + response=await self.graph_slots.get( + client,f"{self.get_url}/emails/{message_id}", + headers={"Authorization":f"Bearer {self.token}"}, ) if response.status_code!=200: raise HTTPException(status_code=response.status_code,detail=response.text) return response.json() + async def triage_round(self,message_ids): """Fetch and classify a whole /email/fetch page, bounded by a semaphore.