proxy pool race condition fixed

freeplay2api.py

@@ -343,63 +343,45 @@ class KeyMaintainer(threading.Thread):
         while True:
             try:
                 logging.info("KeyMaintainer: Starting maintenance cycle.")
+                
+                # First, update balances of existing accounts
                 accounts = self.manager.get_all_accounts()
-
-                # 更新所有账户余额
                 for account in accounts:
                     balance = self.client.check_balance(account["session_id"])
                     if balance != account.get("balance"):
                         account["balance"] = balance
                         self.manager.update_account(account)
-                        logging.info(
-                            f"Account {account['email']} balance updated to ${balance:.4f}"
-                        )
+                        logging.info(f"Account {account['email']} balance updated to ${balance:.4f}")
+
+                # Now, check if we need to register new accounts and loop until we have enough
+                while True:
+                    healthy_accounts = [
+                        acc for acc in self.manager.get_all_accounts()
+                        if acc.get("balance", 0) > self.config["LOW_BALANCE_THRESHOLD"]
+                    ]
+                    needed = self.config["ACTIVE_KEY_THRESHOLD"] - len(healthy_accounts)
 
-                # 检查是否需要注册新账号
-                healthy_accounts = [
-                    acc
-                    for acc in self.manager.get_all_accounts()
-                    if acc.get("balance", 0) > self.config["LOW_BALANCE_THRESHOLD"]
-                ]
-                needed = self.config["ACTIVE_KEY_THRESHOLD"] - len(healthy_accounts)
+                    if needed <= 0:
+                        logging.info(f"Sufficient healthy accounts ({len(healthy_accounts)}). No registration needed for now.")
+                        break
 
-                while needed > 0:
-                    logging.info(
-                        f"Healthy accounts ({len(healthy_accounts)}) below threshold ({self.config['ACTIVE_KEY_THRESHOLD']}). Need to register {needed} new accounts."
-                    )
+                    logging.info(f"Healthy accounts ({len(healthy_accounts)}) below threshold ({self.config['ACTIVE_KEY_THRESHOLD']}). Need to register {needed} new accounts.")
                     
-                    # Determine how many to register in this batch
                     batch_size = min(needed, self.config["REGISTRATION_CONCURRENCY"])
-                    
                     with concurrent.futures.ThreadPoolExecutor(max_workers=batch_size) as executor:
                         futures = [executor.submit(self.client.register) for _ in range(batch_size)]
                         
-                        for future in concurrent.futures.as_completed(futures):
-                            new_account = future.result()
-                            if new_account:
-                                self.manager.add_account(new_account)
-                    
-                    # Re-evaluate after the batch
-                    healthy_accounts = [
-                        acc
-                        for acc in self.manager.get_all_accounts()
-                        if acc.get("balance", 0) > self.config["LOW_BALANCE_THRESHOLD"]
-                    ]
-                    needed = self.config["ACTIVE_KEY_THRESHOLD"] - len(healthy_accounts)
-                    
-                    if needed > 0:
-                        logging.info(f"{needed} more accounts still needed, retrying...")
-                        time.sleep(5) # Wait a bit before the next batch
+                        # Wait for the batch to complete
+                        concurrent.futures.wait(futures)
 
-                if not needed > 0:
-                     logging.info(
-                        f"Sufficient healthy accounts ({len(healthy_accounts)}). No registration needed."
-                    )
+                    logging.info("Registration batch finished. Re-checking healthy account count.")
+                    # The loop will now re-evaluate the 'needed' count
 
             except Exception as e:
                 logging.error(f"Error in KeyMaintainer cycle: {e}")
 
-            time.sleep(self.config["CHECK_INTERVAL_SECONDS"])
+            logging.info(f"KeyMaintainer cycle finished. Waiting for {self.config['CHECK_INTERVAL_SECONDS']} seconds.")
+            time.sleep(self.config['CHECK_INTERVAL_SECONDS'])
 
 
 # --- FastAPI应用 ---
@@ -464,19 +446,10 @@ def initialize_app():
 
         # 5. 启动后台维护线程
         maintainer = KeyMaintainer(account_manager, freeplay_client, config)
-        if not config.get("USE_PROXY_POOL") or (proxy_pool and proxy_pool.is_initialized):
-            maintainer.start()
-            logging.info("Key maintenance service started.")
-        else:
-            # If using proxy pool, delay maintainer start until pool is ready
-            threading.Thread(target=self.wait_for_pool_and_start_maintainer, args=(maintainer, proxy_pool)).start()
-
+        # Always start the maintainer, it will wait for the pool internally if needed.
+        maintainer.start()
+        logging.info("Key maintenance service started.")
 
-def wait_for_pool_and_start_maintainer(self, maintainer, pool):
-    while not pool.is_initialized:
-        time.sleep(2)
-    maintainer.start()
-    logging.info("Key maintenance service started after proxy pool initialization.")
 
 async def authenticate_client(auth: HTTPAuthorizationCredentials = Depends(security)):
     if not auth or auth.credentials not in valid_client_keys: