Merge pull request 'implemented semaphhore' (#28) from RELIMIt into main

Reviewed-on: #28

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

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",
+                response=await self.graph_slots.get(
-                params={"skip":skip,"top":top},
+                    client,f"{self.get_url}/emails",
-                headers={"Authorization":f"Bearer {self.token}"}
+                    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}",
+            response=await self.graph_slots.get(
-            headers={"Authorization":f"Bearer {self.token}"}
+                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.