start instant

routers/gemini.py

@@ -75,7 +75,7 @@ async def create_job(
     input_data: dict
 ) -> GeminiJob:
     """Create a new job in the queue with auto-assigned priority."""
-    from services.gemini_job_worker import get_priority_for_job_type
+    from services.gemini_job_worker import get_priority_for_job_type, get_pool
     
     job_id = f"job_{uuid.uuid4().hex[:16]}"
     priority = get_priority_for_job_type(job_type)
@@ -92,6 +92,9 @@ async def create_job(
     await db.commit()
     await db.refresh(job)
     
+    # Notify workers immediately so they wake up and process this job
+    get_pool().notify_new_job(priority)
+    
     return job
 
 

services/gemini_job_worker.py

@@ -112,20 +112,39 @@ class GeminiJobProcessor(JobProcessor[GeminiJob]):
                 if status_result.get("status") == "completed":
                     video_url = status_result.get("video_url")
                     if video_url:
-                        filename = await service.download_video(video_url, job.job_id)
-                        job.status = "completed"
-                        job.output_data = {"filename": filename}
-                        success = True
+                        try:
+                            filename = await service.download_video(video_url, job.job_id)
+                            job.status = "completed"
+                            job.output_data = {"filename": filename}
+                            job.error_message = None  # Clear any previous error
+                            success = True
+                        except Exception as download_error:
+                            # Download failed - track separately
+                            error_msg = f"Download failed: {download_error}"
+                            logger.error(f"Error downloading video for {job.job_id}: {download_error}")
+                            
+                            # Fail job after 5 download attempts (don't retry forever)
+                            if job.retry_count >= 5:
+                                job.status = "failed"
+                                job.error_message = error_msg
+                                job.completed_at = datetime.utcnow()
+                            else:
+                                # Retry
+                                job.retry_count += 1
+                                job.error_message = f"Download attempt {job.retry_count} failed: {download_error}"
+                                config = WorkerConfig.from_env()
+                                interval = get_interval_for_priority(job.priority, config)
+                                job.next_process_at = datetime.utcnow() + timedelta(seconds=interval)
                     else:
                         job.status = "failed"
                         job.error_message = "No video URL returned"
+                        job.completed_at = datetime.utcnow()
                         error_msg = job.error_message
                 else:
                     job.status = "failed"
                     job.error_message = status_result.get("error", "Unknown error")
+                    job.completed_at = datetime.utcnow()
                     error_msg = job.error_message
-                
-                job.completed_at = datetime.utcnow()
             else:
                 # Not done - reschedule
                 job.retry_count += 1
@@ -137,6 +156,7 @@ class GeminiJobProcessor(JobProcessor[GeminiJob]):
         except Exception as e:
             logger.error(f"Error checking video status for {job.job_id}: {e}")
             job.retry_count += 1
+            job.error_message = f"Status check failed: {e}"
             config = WorkerConfig.from_env()
             interval = get_interval_for_priority(job.priority, config)
             job.next_process_at = datetime.utcnow() + timedelta(seconds=interval)

services/gemini_service.py

@@ -212,10 +212,13 @@ class GeminiService:
         Returns status and video URL if complete.
         """
         try:
-            # Get operation status
+            # Get operation status using the operation object
+            # First, we need to recreate the operation from the name
+            from google.genai.types import GenerateVideosOperation
+            
             operation = await asyncio.to_thread(
                 self.client.operations.get,
-                name=gemini_operation_name
+                GenerateVideosOperation(name=gemini_operation_name, done=False)
             )
             
             if not operation.done:
@@ -225,39 +228,36 @@ class GeminiService:
                     "status": "pending"
                 }
             
+            # Check for error - handle both string and object types
             if operation.error:
+                error_msg = operation.error
+                if hasattr(operation.error, 'message'):
+                    error_msg = operation.error.message
                 return {
                     "gemini_operation_name": gemini_operation_name,
                     "done": True,
                     "status": "failed",
-                    "error": operation.error.message or "Unknown error"
-                }
-            
-            # Extract video URI
-            generated_videos = getattr(operation.response, 'generated_videos', None)
-            if not generated_videos:
-                return {
-                    "gemini_operation_name": gemini_operation_name,
-                    "done": True,
-                    "status": "failed",
-                    "error": "No video URI returned. May be due to safety filters."
+                    "error": str(error_msg) or "Unknown error"
                 }
             
-            video_uri = generated_videos[0].video.uri if generated_videos[0].video else None
-            if not video_uri:
-                return {
-                    "gemini_operation_name": gemini_operation_name,
-                    "done": True,
-                    "status": "failed",
-                    "error": "No video URI in response."
-                }
+            # Extract video URI from result
+            result = operation.result
+            if result and hasattr(result, 'generated_videos') and result.generated_videos:
+                video = result.generated_videos[0]
+                if hasattr(video, 'video') and video.video and hasattr(video.video, 'uri'):
+                    video_uri = video.video.uri
+                    return {
+                        "gemini_operation_name": gemini_operation_name,
+                        "done": True,
+                        "status": "completed",
+                        "video_url": f"{video_uri}&key={self.api_key}"
+                    }
             
-            # Return success with video URL (internal - will be downloaded by router)
             return {
                 "gemini_operation_name": gemini_operation_name,
                 "done": True,
-                "status": "completed",
-                "video_url": f"{video_uri}&key={self.api_key}"
+                "status": "failed",
+                "error": "No video URI returned. May be due to safety filters."
             }
             
         except Exception as error:
@@ -280,7 +280,8 @@ class GeminiService:
         filepath = os.path.join(DOWNLOADS_DIR, filename)
         
         try:
-            async with httpx.AsyncClient(timeout=120.0) as client:
+            # follow_redirects=True is required as Gemini returns 302 redirects
+            async with httpx.AsyncClient(timeout=120.0, follow_redirects=True) as client:
                 response = await client.get(video_url)
                 response.raise_for_status()
                 

services/priority_worker_pool.py

@@ -149,7 +149,8 @@ class PriorityWorker(Generic[JobType]):
         session_maker: async_sessionmaker,
         job_model: type,
         job_processor: JobProcessor[JobType],
-        max_retries: int = 60
+        max_retries: int = 60,
+        wake_event: Optional[asyncio.Event] = None
     ):
         self.worker_id = worker_id
         self.priority = priority
@@ -160,6 +161,7 @@ class PriorityWorker(Generic[JobType]):
         self.max_retries = max_retries
         self._running = False
         self._current_job_id: Optional[str] = None
+        self._wake_event = wake_event  # Event to wake worker immediately when new jobs arrive
     
     async def start(self):
         """Start the worker polling loop."""
@@ -173,16 +175,44 @@ class PriorityWorker(Generic[JobType]):
         logger.info(f"Worker {self.worker_id} ({self.priority}) stopped")
     
     async def _poll_loop(self):
-        """Main polling loop."""
+        """Main polling loop with optimized scheduling.
+        
+        Optimizations:
+        - When no jobs are found, sleep for poll_interval before checking again
+        - When a job is processed, immediately check for the next job (no waiting)
+        - This ensures first job starts immediately when queue was empty
+        - This ensures next job starts immediately after current job finishes
+        """
         while self._running:
+            job_found = False
             try:
-                await self._process_one_job()
+                job_found = await self._process_one_job()
             except Exception as e:
                 logger.error(f"Worker {self.worker_id}: Error in poll loop: {e}")
-            await asyncio.sleep(self.poll_interval)
+            
+            # Only sleep if no job was found - otherwise immediately look for next job
+            if not job_found:
+                # Wait on event with timeout - allows immediate wake-up when new job arrives
+                if self._wake_event:
+                    try:
+                        # Wait for event or timeout (whichever comes first)
+                        await asyncio.wait_for(
+                            self._wake_event.wait(),
+                            timeout=self.poll_interval
+                        )
+                        # Clear event after waking (we'll check for jobs)
+                        self._wake_event.clear()
+                    except asyncio.TimeoutError:
+                        pass  # Normal timeout, check for jobs
+                else:
+                    await asyncio.sleep(self.poll_interval)
     
-    async def _process_one_job(self):
-        """Find and process one job."""
+    async def _process_one_job(self) -> bool:
+        """Find and process one job.
+        
+        Returns:
+            True if a job was found and processed, False if no jobs available
+        """
         async with self.session_maker() as session:
             now = datetime.utcnow()
             
@@ -202,18 +232,20 @@ class PriorityWorker(Generic[JobType]):
             job = result.scalar_one_or_none()
             
             if not job:
-                return
+                return False
             
             self._current_job_id = job.job_id
             
             try:
                 await self._process_job(session, job)
+                return True
             except Exception as e:
                 logger.error(f"Worker {self.worker_id}: Error processing job {job.job_id}: {e}")
                 job.status = "failed"
                 job.error_message = str(e)
                 job.completed_at = datetime.utcnow()
                 await session.commit()
+                return True  # Job was found, even though it failed
             finally:
                 self._current_job_id = None
     
@@ -326,6 +358,13 @@ class PriorityWorkerPool(Generic[JobType]):
         )
         self.workers: List[PriorityWorker] = []
         self._running = False
+        
+        # Wake events for each priority tier - allows immediate job notification
+        self._wake_events: dict[str, asyncio.Event] = {
+            "fast": asyncio.Event(),
+            "medium": asyncio.Event(),
+            "slow": asyncio.Event()
+        }
     
     async def start(self):
         """Start all workers."""
@@ -341,7 +380,8 @@ class PriorityWorkerPool(Generic[JobType]):
                 session_maker=self.session_maker,
                 job_model=self.job_model,
                 job_processor=self.job_processor,
-                max_retries=self.config.max_retries
+                max_retries=self.config.max_retries,
+                wake_event=self._wake_events["fast"]
             )
             self.workers.append(worker)
             await worker.start()
@@ -356,7 +396,8 @@ class PriorityWorkerPool(Generic[JobType]):
                 session_maker=self.session_maker,
                 job_model=self.job_model,
                 job_processor=self.job_processor,
-                max_retries=self.config.max_retries
+                max_retries=self.config.max_retries,
+                wake_event=self._wake_events["medium"]
             )
             self.workers.append(worker)
             await worker.start()
@@ -371,7 +412,8 @@ class PriorityWorkerPool(Generic[JobType]):
                 session_maker=self.session_maker,
                 job_model=self.job_model,
                 job_processor=self.job_processor,
-                max_retries=self.config.max_retries
+                max_retries=self.config.max_retries,
+                wake_event=self._wake_events["slow"]
             )
             self.workers.append(worker)
             await worker.start()
@@ -383,6 +425,18 @@ class PriorityWorkerPool(Generic[JobType]):
             f"{self.config.fast_workers} fast, {self.config.medium_workers} medium, {self.config.slow_workers} slow"
         )
     
+    def notify_new_job(self, priority: str):
+        """
+        Wake sleeping workers of the specified priority tier.
+        Call this when a new job is created to start processing immediately.
+        
+        Args:
+            priority: Priority tier ("fast", "medium", or "slow")
+        """
+        if priority in self._wake_events:
+            self._wake_events[priority].set()
+            logger.debug(f"Notified {priority} workers of new job")
+    
     async def stop(self):
         """Stop all workers."""
         self._running = False