Update worker.py

worker.py

@@ -8,56 +8,67 @@ log = get_logger()
 workers_active = []
 
 async def worker_task(worker_id: int):
-    log.info(f"Worker {worker_id} started.")
+    log.info(f"Worker {worker_id} successfully mapped to runtime thread loop pool.")
     await state.init_queue()
     
     while True:
-        try:
-            # Safe boundary checks for structural state execution
-            if state.status == "STOPPED":
-                break
-                
-            if state.status == "PAUSED":
-                await asyncio.sleep(1)
-                continue
+        if state.status == "STOPPED":
+            break
+            
+        if state.status == "PAUSED":
+            await asyncio.sleep(1)
+            continue
 
-            try:
-                word = state.queue.get_nowait()
-            except asyncio.QueueEmpty:
-                log.info(f"Worker {worker_id} finished. Queue empty.")
-                break
+        try:
+            # Atomic fetch from asynchronous local memory queues
+            word = state.queue.get_nowait()
+        except asyncio.QueueEmpty:
+            log.info(f"Worker {worker_id} finished execution. Memory loop registry empty.")
+            break
 
-            # Handle rotating proxy allocation safely
-            proxy = random.choice(state.proxies) if state.proxies else None
-            
-            # Execute AJAX search extraction
+        # Explicitly pass None for proxy allocations to isolate proxy-free routing channels
+        proxy = None
+        
+        try:
+            # Connect directly to the direct DOM scanning sequence
             status = await scraper.check_fragment(word, proxy_url=proxy)
             
             if status == "ERROR":
-                # CRITICAL FIX: Ensure property array initialization inside the thread loop space
                 if not hasattr(state, 'fail_counts') or state.fail_counts is None:
                     state.fail_counts = {}
                     
                 state.fail_counts[word] = state.fail_counts.get(word, 0) + 1
                 
-                # Failsafe 3-Strikes Adaptive Moving Sequence
+                # Failsafe 3-Strikes Matrix Boundary
                 if state.fail_counts[word] >= 3:
+                    log.warning(f"⚠️ 3-Strikes Exhausted for '{word}'. Dumping entry safely to results/error.txt")
                     await state.save_result(word, "error")
                 else:
+                    # Defensive Pacing: Sleep the worker task BEFORE putting the item back into circulation
+                    # This prevents rapid looping over the same unproxied connection
+                    stagger_sleep = random.uniform(3.0, 6.0)
+                    log.info(f"🔄 Intercepted execution error for '{word}'. Workers cooling loop path for {stagger_sleep:.2f}s...")
+                    await asyncio.sleep(stagger_sleep)
+                    
                     await state.queue.put(word)
-                    # Dynamic staggered delay before recycling word to prevent loop collisions
-                    await asyncio.sleep(random.uniform(1.5, 3.5))
+                    
             elif status:
-                # Remove word from failure map tracking on successful evaluation
+                # Remove from failure cache logs upon successful classification mapping
                 if hasattr(state, 'fail_counts') and state.fail_counts and word in state.fail_counts:
-                    del state.fail_counts[word]
+                    try: del state.fail_counts[word]
+                    except KeyError: pass
+                    
                 await state.save_result(word, status)
                 
-            state.queue.task_done()
-
         except Exception as e:
-            log.error(f"⚠️ Worker {worker_id} safely intercepted loop exception: {str(e)}")
-            await asyncio.sleep(2)
+            log.error(f"🚨 Critical transaction fault inside Worker {worker_id} processing '{word}': {str(e)}")
+            # Failsafe re-enqueue option to ensure data is never lost mid-flight
+            await state.queue.put(word)
+            await asyncio.sleep(4)
+            
+        finally:
+            # Bulletproof Execution: Always register task completion to keep the event loop balanced
+            state.queue.task_done()
 
 async def manage_workers():
     global workers_active
@@ -74,7 +85,8 @@ async def manage_workers():
             for t in workers_active:
                 t.cancel()
             workers_active.clear()
-            log.info("All workers forcefully stopped.")
+            log.info("All concurrent background workers forcefully torn down.")
             
+        # Keep tracking configurations active by filtering completed execution futures
         workers_active = [t for t in workers_active if not t.done()]
         await asyncio.sleep(1)