В моем коде:
# routing.py

import threading
import time
from typing import List, Dict, Optional


class Request:
    def __init__(self, client_id: str, request_id: str, processing_time: float):
        self.client_id = client_id
        self.request_id = request_id
        self.processing_time = processing_time


class Server:
    """
    Represents a server that can process requests.
    """
    def __init__(self, server_id: str, performance_score: int):
        self.server_id = server_id
        self.performance_score = performance_score
        self._alive = True
        self._processed_requests = set()
        self._lock = threading.Lock()

    def process_request(self, request: Request) -> None:
        """
        Processes a request. Save the request client_id and request_id.
        Simulates processing time.
        """
        if not self.is_alive():
            raise Exception(f"Server {self.server_id} is not alive.")
        # Simulate processing
        time.sleep(request.processing_time)
        with self._lock:
            self._processed_requests.add(request.request_id)

    def is_alive(self) -> bool:
        """
        Checks if the server is alive.
        """
        with self._lock:
            return self._alive

    def crash(self) -> None:
        """
        Crashes the server.
        """
        with self._lock:
            self._alive = False

    def recover(self) -> None:
        """
        Recovers the server.
        """
        with self._lock:
            self._alive = True

    def is_processed(self, request_id: str) -> bool:
        """
        Checks if the request is processed. Emulates caching.
        """
        with self._lock:
            return request_id in self._processed_requests


class Router:
    """
    Represents a router that can route requests to servers using Smooth Weighted Round Robin.
    """
    def __init__(self, servers: List[Server], max_load: int):
        self.servers: List[Server] = list(servers)  # Public attribute for testing
        self.max_load = max_load
        self.lock = threading.RLock()  # Use reentrant lock for nested locking
        self.load_semaphore = threading.Semaphore(max_load)
        self.client_affinity: Dict[str, str] = dict()  # Maps client_id to server_id

        # Semaphore per server to limit concurrency (1 request per server)
        self.server_semaphores: Dict[str, threading.Semaphore] = {
            server.server_id: threading.Semaphore(1) for server in self.servers
        }

        # Initialize Smooth Weighted Round Robin variables
        self.current_weights: Dict[str, int] = {
            server.server_id: 0 for server in self.servers if server.is_alive()
        }
        self.total_weight = sum(server.performance_score for server in self.servers if server.is_alive())

        # Condition variable to wait for server availability
        self.server_available = threading.Condition(self.lock)

    def _rebuild_server_weights(self):
        """
        Rebuilds the current_weights and total_weight based on alive servers.
        Does NOT recreate server_semaphores to prevent inconsistencies.
        Notifies all waiting threads if servers are available.
        """
        with self.lock:
            # Reset current_weights for alive servers
            self.current_weights = {
                server.server_id: 0 for server in self.servers if server.is_alive()
            }
            # Recalculate total_weight
            self.total_weight = sum(server.performance_score for server in self.servers if server.is_alive())

            # Notify all waiting threads that servers are available
            if self.total_weight > 0:
                self.server_available.notify_all()

    def _get_next_server_id(self) -> Optional[str]:
        """
        Gets the next server_id based on Smooth Weighted Round Robin.
        """
        if not self.current_weights:
            return None

        # Increment current_weight by server's weight for all alive servers
        for server_id in self.current_weights:
            server = self._get_server_by_id(server_id)
            if server and server.is_alive():
                self.current_weights[server_id] += server.performance_score

        # Select server with the highest current_weight
        selected_server_id = max(self.current_weights, key=lambda s_id: self.current_weights[s_id])

        # Decrement selected_server's current_weight by total_weight
        self.current_weights[selected_server_id] -= self.total_weight

        return selected_server_id

    def _get_server_by_id(self, server_id: str) -> Optional[Server]:
        """
        Retrieves a server instance by its server_id.
        """
        for server in self.servers:
            if server.server_id == server_id:
                return server
        return None

    def route(self, request: Request) -> None:
        """
        Routes a request to an appropriate server using Smooth Weighted Round Robin.
        Waits if no servers are available.
        """
        self.load_semaphore.acquire()
        try:
            while True:
                with self.lock:
                    server_id_to_use = None

                    # Check client affinity
                    if request.client_id in self.client_affinity:
                        server_id = self.client_affinity[request.client_id]
                        server = self._get_server_by_id(server_id)

                        if server and server.is_alive():
                            semaphore = self.server_semaphores.get(server_id)
                            if semaphore and semaphore.acquire(blocking=False):
                                # Server is available and semaphore acquired
                                server_id_to_use = server_id
                                break  # Exit loop to process request
                            else:
                                # Unable to acquire semaphore, remove affinity
                                del self.client_affinity[request.client_id]
                        else:
                            # Server is not alive, remove affinity
                            del self.client_affinity[request.client_id]

                    # If no affinity or unable to use it, select next server via SWWR
                    if not server_id_to_use:
                        selected_server_id = self._get_next_server_id()
                        if selected_server_id:
                            semaphore = self.server_semaphores.get(selected_server_id)
                            if semaphore and semaphore.acquire(blocking=False):
                                # Assign client affinity to selected server
                                self.client_affinity[request.client_id] = selected_server_id
                                server_id_to_use = selected_server_id
                                break  # Exit loop to process request

                if server_id_to_use is None:
                    # If no server was selected, wait until a server becomes available
                    with self.server_available:
                        self.server_available.wait()

            # Now, server_id_to_use is assigned and semaphore is acquired
            server = self._get_server_by_id(server_id_to_use)
            if server and server.is_alive():
                try:
                    server.process_request(request)
                except Exception:
                    with self.lock:
                        # Remove client affinity on exception
                        if request.client_id in self.client_affinity:
                            del self.client_affinity[request.client_id]
                            # Rebuild server weights if server crashed during processing
                            self._rebuild_server_weights()
                    with self.server_available:
                        self.server_available.notify_all()
                    raise
                finally:
                    # Release server semaphore and notify waiting threads
                    self.server_semaphores[server_id_to_use].release()
                    with self.server_available:
                        self.server_available.notify()
            else:
                # Server is not alive, release semaphore and remove affinity
                if server_id_to_use in self.server_semaphores:
                    self.server_semaphores[server_id_to_use].release()
                with self.lock:
                    if request.client_id in self.client_affinity:
                        del self.client_affinity[request.client_id]
                with self.server_available:
                    self.server_available.notify_all()
        finally:
            # Release the load semaphore
            self.load_semaphore.release()

    def add_server(self, server: Server) -> None:
        """
        Adds a new server to the router.
        """
        with self.lock:
            if server.server_id not in [s.server_id for s in self.servers]:
                self.servers.append(server)
                # Initialize semaphore for the new server
                self.server_semaphores[server.server_id] = threading.Semaphore(1)
                if server.is_alive():
                    self.current_weights[server.server_id] = 0
                    self.total_weight += server.performance_score
                    # Notify all waiting threads about the new server
                    with self.server_available:
                        self.server_available.notify_all()

    def remove_server(self, server: Server) -> None:
        """
        Removes a server from the router.
        """
        with self.lock:
            if server.server_id in [s.server_id for s in self.servers]:
                self.servers = [s for s in self.servers if s.server_id != server.server_id]
                # Adjust weights and total_weight if server was alive
                if server.is_alive() and server.server_id in self.current_weights:
                    self.total_weight -= server.performance_score
                    del self.current_weights[server.server_id]
                # Remove client affinities tied to this server
                clients_to_remove = [
                    client for client, srv_id in self.client_affinity.items() if srv_id == server.server_id
                ]
                for client in clients_to_remove:
                    del self.client_affinity[client]
                # Remove semaphore of the server
                if server.server_id in self.server_semaphores:
                    del self.server_semaphores[server.server_id]
                # Notify all waiting threads about the removal
                with self.server_available:
                    self.server_available.notify_all()

    def crash_server(self, server_id: str) -> None:
        """
        Crashes a server by its ID.
        """
        with self.lock:
            server = self._get_server_by_id(server_id)
            if server:
                server.crash()
            # Remove client affinities tied to this server
            clients_to_remove = [
                client for client, srv_id in self.client_affinity.items() if srv_id == server_id
            ]
            for client in clients_to_remove:
                del self.client_affinity[client]
            # Rebuild server weights without touching server_semaphores
            self._rebuild_server_weights()

    def recover_server(self, server_id: str) -> None:
        """
        Recovers a server by its ID.
        """
        with self.lock:
            server = self._get_server_by_id(server_id)
            if server:
                server.recover()
                if server.server_id not in self.current_weights:
                    self.current_weights[server.server_id] = 0
                    self.total_weight += server.performance_score
            # Rebuild server weights without touching server_semaphores
            self._rebuild_server_weights()

    def get_servers(self) -> List[Server]:
        """
        Returns the list of current servers.
        """
        with self.lock:
            return list(self.servers)Ошибка на этом тесте:

test_public.py:98 (test_router_server_failure)
router = <routing.Router object at 0x106a12c30>

    def test_router_server_failure(router):
        router.servers[0].crash()
    
        requests = [Request(f"client{i}", f"request{i}", 0.1) for i in range(6)]
        for request in requests:
            router.route(request)
    
        assert not any(router.servers[0].is_processed(f"request{i}") for i in range(6))
>       assert all(any(server.is_processed(f"request{i}") for server in router.servers[1:]) for i in range(6))
E       assert False
E        +  where False = all(<generator object test_router_server_failure.<locals>.<genexpr> at 0x106916e30>)

test_public.py:107: AssertionError

Но вот в этом коде такой ошибки нет:

import threading  
import time  
from typing import List, Dict, Optional  
from concurrent.futures import ThreadPoolExecutor, as_completed  
  
  
class Request:  
    def __init__(self, client_id: str, request_id: str, processing_time: float):  
        self.client_id = client_id  
        self.request_id = request_id  
        self.processing_time = processing_time  
  
  
class Server:  
    """  
    Represents a server that can process requests.    """    def __init__(self, server_id: str, performance_score: int):  
        self.server_id = server_id  
        self.performance_score = performance_score  
        self._alive = True  
        self._processed_requests = set()  
        self._lock = threading.Lock()  
  
    def process_request(self, request: Request) -> None:  
        """  
        Processes a request. Save the request client_id and request_id.        Simulates processing time.        """        if not self.is_alive():  
            raise Exception(f"Server {self.server_id} is not alive.")  
        # Simulate processing  
        time.sleep(request.processing_time)  
        with self._lock:  
            self._processed_requests.add(request.request_id)  
  
    def is_alive(self) -> bool:  
        """  
        Checks if the server is alive.        """        with self._lock:  
            return self._alive  
  
    def crash(self) -> None:  
        """  
        Crashes the server.        """        with self._lock:  
            self._alive = False  
  
    def recover(self) -> None:  
        """  
        Recovers the server.        """        with self._lock:  
            self._alive = True  
  
    def is_processed(self, request_id: str) -> bool:  
        """  
        Checks if the request is processed. Emulates caching.        """        with self._lock:  
            return request_id in self._processed_requests  
  
  
class Router:  
    """  
    Represents a router that can route requests to servers using Smooth Weighted Round Robin.    """    def __init__(self, servers: List[Server], max_load: int):  
        self.servers: List[Server] = list(servers)  # Open attribute for testing  
        self.max_load = max_load  
        self.lock = threading.RLock()  # Use reentrant lock  
        self.load_semaphore = threading.Semaphore(max_load)  
        self.client_affinity: Dict[str, Server] = dict()  
  
        # Condition variable to wait for available servers  
        self.server_available = threading.Condition(self.lock)  
  
        # Initialize Smooth Weighted Round Robin variables  
        self.current_weights: Dict[Server, int] = {  
            server: 0 for server in self.servers if server.is_alive()  
        }  
        self.total_weight = sum(server.performance_score for server in self.servers if server.is_alive())  
  
    def _rebuild_server_weights(self):  
        """  
        Rebuilds the current_weights and total_weight based on alive servers.        Notifies all waiting threads if servers are available.        """        with self.lock:  
            # Reset current_weights  
            self.current_weights = {  
                server: 0 for server in self.servers if server.is_alive()  
            }  
            # Recalculate total_weight  
            self.total_weight = sum(server.performance_score for server in self.servers if server.is_alive())  
            # Notify all waiting threads that servers are available  
            if self.total_weight > 0:  
                with self.server_available:  
                    self.server_available.notify_all()  
  
    def _get_next_server(self) -> Optional[Server]:  
        """  
        Gets the next server based on Smooth Weighted Round Robin.        """        with self.lock:  
            alive_servers = [s for s in self.servers if s.is_alive()]  
            if not alive_servers:  
                return None  
  
            # Increment current_weight by server's weight  
            for server in alive_servers:  
                self.current_weights[server] += server.performance_score  
  
            # Select server with highest current_weight  
            selected_server = max(alive_servers, key=lambda s: self.current_weights[s])  
  
            # Decrement selected_server's current_weight by total_weight  
            self.current_weights[selected_server] -= self.total_weight  
  
            return selected_server  
  
    def route(self, request: Request) -> None:  
        """  
        Routes a request to an appropriate server using Smooth Weighted Round Robin.        Waits if no servers are available.        """        self.load_semaphore.acquire()  
        try:  
            while True:  
                with self.lock:  
                    # Check client affinity  
                    if request.client_id in self.client_affinity:  
                        server = self.client_affinity[request.client_id]  
                        if not server.is_alive():  
                            # Remove affinity if server is down  
                            del self.client_affinity[request.client_id]  
                            server = self._get_next_server()  
                            if server:  
                                self.client_affinity[request.client_id] = server  
                    else:  
                        server = self._get_next_server()  
                        if server:  
                            self.client_affinity[request.client_id] = server  
  
                    if server and server.is_alive():  
                        break  # Found an available server  
                    else:  
                        # Wait until a server becomes available  
                        with self.server_available:  
                            self.server_available.wait()  
  
            if server and server.is_alive():  
                server.process_request(request)  
        finally:  
            self.load_semaphore.release()  
  
    def add_server(self, server: Server) -> None:  
        """  
        Adds a new server to the router.        """        with self.lock:  
            if server not in self.servers:  
                self.servers.append(server)  
                if server.is_alive():  
                    self.current_weights[server] = 0  
                    self.total_weight += server.performance_score  
                    self.server_available.notify_all()  
  
    def remove_server(self, server: Server) -> None:  
        """  
        Removes a server from the router.        """        with self.lock:  
            if server in self.servers:  
                self.servers.remove(server)  
                # Remove from current_weights and adjust total_weight  
                if server in self.current_weights:  
                    self.total_weight -= server.performance_score  
                    del self.current_weights[server]  
                # Remove client affinities tied to this server  
                clients_to_remove = [  
                    client for client, srv in self.client_affinity.items() if srv == server  
                ]  
                for client in clients_to_remove:  
                    del self.client_affinity[client]  
                self.server_available.notify_all()  
  
    def crash_server(self, server_id: str) -> None:  
        """  
        Crashes a server by its ID.        """        with self.lock:  
            for server in self.servers:  
                if server.server_id == server_id:  
                    server.crash()  
                    break  
            # Remove client affinities tied to this server  
            clients_to_remove = [  
                client for client, srv in self.client_affinity.items() if srv.server_id == server_id  
            ]  
            for client in clients_to_remove:  
                del self.client_affinity[client]  
            # Rebuild server weights and notify  
            self._rebuild_server_weights()  
  
    def recover_server(self, server_id: str) -> None:  
        """  
        Recovers a server by its ID.        """        with self.lock:  
            for server in self.servers:  
                if server.server_id == server_id:  
                    server.recover()  
                    break  
            # Rebuild server weights and notify  
            self._rebuild_server_weights()  
  
    def get_servers(self) -> List[Server]:  
        """  
        Returns the list of current servers.        """        with self.lock:  
            return list(self.servers)