Update worker.py

worker.py

@@ -9,65 +9,79 @@ workers_active = []
 
 async def worker_task(worker_id: int):
     log.info(f"Worker {worker_id} started.")
+    
+    # Ensure queue state is completely bound to the active loop thread
     await state.init_queue()
     
     while True:
         try:
-            # Handle STOP state
             if state.status == "STOPPED":
+                log.info(f"Worker {worker_id} stopped via command.")
                 break
                 
-            # Handle PAUSE state
             if state.status == "PAUSED":
                 await asyncio.sleep(1)
                 continue
 
-            # Try to get a word from the in-memory queue
+            # Bulletproof non-blocking queue fetch with polling failsafe
             try:
-                word = state.queue.get_nowait()
-            except asyncio.QueueEmpty:
-                log.info(f"Worker {worker_id} finished. Queue empty.")
-                break
+                word = await asyncio.wait_for(state.queue.get(), timeout=2.0)
+            except asyncio.TimeoutError:
+                # Failsafe check: if scanner is still running but queue is genuinely depleted, exit gracefully
+                if state.queue.empty():
+                    log.info(f"Worker {worker_id} exiting. Queue is empty.")
+                    break
+                continue
 
-            # Pick a proxy if available
+            # Thread-safe proxy slicing
             proxy = random.choice(state.proxies) if state.proxies else None
             
-            # Execute scrape
+            # Execute HTTP scrape request via native aiohttp engine
             status = await scraper.check_fragment(word, proxy_url=proxy)
             
             if status == "ERROR":
-                # Put it back to try again later
-                await state.queue.put(word)
-                await asyncio.sleep(1.5)
+                state.fail_counts[word] = state.fail_counts.get(word, 0) + 1
+                
+                # 3-Strike proxy drops rule
+                if state.fail_counts[word] >= 3:
+                    log.error(f"❌ Word '{word}' failed 3 distinct proxy attempts. Logging to error bin.")
+                    await state.save_result(word, "error")
+                else:
+                    await state.queue.put(word)
+                    await asyncio.sleep(1.0)
             elif status:
                 await state.save_result(word, status)
                 
             state.queue.task_done()
 
         except Exception as e:
-            log.exception(f"Worker {worker_id} crashed:")
+            log.exception(f"⚠️ Worker {worker_id} unhandled crash loop event:")
             await asyncio.sleep(2)
 
 async def manage_workers():
-    """Monitors state and spins up workers as needed."""
     global workers_active
     
     while True:
-        if state.status == "RUNNING" and len(workers_active) < state.concurrency:
-            # Spin up missing workers
-            needed = state.concurrency - len(workers_active)
-            for i in range(needed):
-                t = asyncio.create_task(worker_task(len(workers_active) + i))
-                workers_active.append(t)
-        
-        elif state.status == "STOPPED" and workers_active:
-            # Cancel all workers if stopped
-            for t in workers_active:
-                t.cancel()
-            workers_active.clear()
-            log.info("All workers forcefully stopped.")
+        try:
+            if state.status == "RUNNING":
+                # Dynamically fill out the active worker pool target
+                if len(workers_active) < state.concurrency:
+                    needed = state.concurrency - len(workers_active)
+                    current_count = len(workers_active)
+                    for i in range(needed):
+                        t = asyncio.create_task(worker_task(current_count + i))
+                        workers_active.append(t)
+            
+            elif state.status == "STOPPED" and workers_active:
+                for t in workers_active:
+                    t.cancel()
+                workers_active.clear()
+                log.info("Manager: All active worker pools cleared and forcefully reset.")
+                
+            # Filter and drop dead/finished thread futures cleanly
+            workers_active = [t for t in workers_active if not t.done()]
+            
+        except Exception as e:
+            log.error(f"Manager Engine Fault: {str(e)}")
             
-        # Clean up finished tasks from the active list
-        workers_active = [t for t in workers_active if not t.done()]
-        
         await asyncio.sleep(1)