Fix concurrency: Use Queue + Semaphore pattern

Author: manascb1344

app.py

@@ -7,6 +7,7 @@
 import json
 import time
 import re
+import asyncio
 from urllib.parse import urlparse
 from fastapi import Request, HTTPException
 
@@ -189,20 +190,12 @@ async def verify_signature(request: Request, body: bytes) -> str | None:
     return delivery_id
 
 
-@app.function(image=runner_image, secrets=[github_secret])
-@modal.concurrent(max_inputs=DEFAULT_MAX_PARALLEL)
-async def spawn_runner_sandbox(job_data: dict):
-    """Worker that spawns sandboxes with controlled concurrency using @modal.concurrent.
-
-    This runs with max_inputs=DEFAULT_MAX_PARALLEL to match GitHub's max-parallel,
-    ensuring we don't spawn more sandboxes than GitHub can use.
-    """
+async def _spawn_single_sandbox(job_data: dict):
+    """Helper to spawn a single sandbox and wait for completion."""
     job_id = job_data["job_id"]
-    repo_url = job_data["repo_url"]
-    job_labels = job_data.get("job_labels", [])
     jit_config = job_data["jit_config"]
 
-    logger.info(f"Worker spawning sandbox for job {job_id}")
+    logger.info(f"Spawning sandbox for job {job_id}")
 
     try:
         cmd = "cd /actions-runner && export RUNNER_ALLOW_RUNASROOT=1 && ./run.sh --jitconfig $GHA_JIT_CONFIG"
@@ -219,7 +212,7 @@ async def spawn_runner_sandbox(job_data: dict):
 
         sandbox.set_tags({"job_id": str(job_id)})
 
-        # Wait for sandbox to complete (blocks this worker slot)
+        # Wait for completion
         sandbox.wait()
 
         logger.info(f"Sandbox for job {job_id} completed")
@@ -229,6 +222,47 @@ async def spawn_runner_sandbox(job_data: dict):
         raise
 
 
+@app.function(image=runner_image, secrets=[github_secret])
+async def process_job_batch():
+    """Process jobs from queue with controlled concurrency using asyncio.Semaphore.
+
+    This function runs continuously, pulling jobs from the queue and spawning
+    sandboxes with max DEFAULT_MAX_PARALLEL concurrent executions.
+    """
+    semaphore = asyncio.Semaphore(DEFAULT_MAX_PARALLEL)
+
+    logger.info(
+        f"Starting job processor with max {DEFAULT_MAX_PARALLEL} concurrent sandboxes"
+    )
+
+    while True:
+        try:
+            # Get job from queue (blocks until available)
+            job_data = await job_queue.get.aio()
+
+            if job_data is None:
+                # Sentinel value to stop processing
+                break
+
+            # Acquire semaphore before spawning
+            await semaphore.acquire()
+
+            # Spawn sandbox in background task so we can continue to next job
+            asyncio.create_task(_process_with_semaphore(semaphore, job_data))
+
+        except Exception as e:
+            logger.error(f"Error in job processor: {type(e).__name__}: {e}")
+            await asyncio.sleep(1)
+
+
+async def _process_with_semaphore(semaphore: asyncio.Semaphore, job_data: dict):
+    """Process a job and release semaphore when done."""
+    try:
+        await _spawn_single_sandbox(job_data)
+    finally:
+        semaphore.release()
+
+
 @app.function(image=runner_image, secrets=[github_secret])
 @modal.fastapi_endpoint(method="POST")
 async def github_webhook(request: Request):
@@ -403,15 +437,14 @@ async def github_webhook(request: Request):
             "jit_config": jit_config,
         }
 
-        # Spawn worker function to process this job
-        # @modal.concurrent controls concurrent execution
-        await spawn_runner_sandbox.spawn.aio(job_data)
+        # Put job in queue for processor
+        await job_queue.put.aio(job_data)
 
         logger.info(f"Successfully queued job {job_id}")
 
     except Exception as e:
         logger.error(f"Failed to queue job {job_id}: {type(e).__name__}")
-        raise HTTPException(status_code=500, detail="Failed to spawn runner sandbox")
+        raise HTTPException(status_code=500, detail="Failed to queue job")
 
     logger.info(f"Successfully queued runner for job {job_id}")
     return {"status": "queued", "job_id": job_id}