RDBC-700 Final sync

Author: poissoncorp

ravendb/documents/session/event_args.py

@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional
 
 import requests
 
@@ -263,7 +263,12 @@ def attempt_number(self) -> int:
 
 class FailedRequestEventArgs(EventArgs):
     def __init__(
-        self, database: str, url: str, exception: Exception, request: requests.Request, response: requests.Response
+        self,
+        database: str,
+        url: str,
+        exception: Exception,
+        request: Optional[requests.Request],
+        response: Optional[requests.Response],
     ):
         self.__database = database
         self.__url = url

ravendb/documents/subscriptions/worker.py

@@ -33,12 +33,14 @@
     SubscriberErrorException,
     SubscriptionDoesNotExistException,
     AllTopologyNodesDownException,
+    InvalidNetworkTopologyException,
+    SubscriptionMessageTypeException,
 )
 from ravendb.exceptions.raven_exceptions import ClientVersionMismatchException
 from ravendb.extensions.json_extensions import JsonExtensions
 from ravendb.json.metadata_as_dictionary import MetadataAsDictionary
 from ravendb.documents.session.document_session import DocumentSession
-from ravendb.documents.subscriptions.options import SubscriptionWorkerOptions
+from ravendb.documents.subscriptions.options import SubscriptionWorkerOptions, SubscriptionOpeningStrategy
 from ravendb.primitives.exceptions import OperationCancelledException
 from ravendb.primitives.misc import CancellationTokenSource
 from ravendb.http.request_executor import RequestExecutor
@@ -405,6 +407,8 @@ def _read_server_response_and_get_version(self, url: str, sock: socket) -> int:
             # Kindly request the server to drop the connection
             self._send_drop_message(reply)
             raise RuntimeError(f"Can't connect to database {self._db_name} because: {reply.message}")
+        if reply.status == TcpConnectionStatus.INVALID_NETWORK_TOPOLOGY:
+            raise InvalidNetworkTopologyException(f"Failed to connect to url {url} because {reply.message}")
 
         return reply.version
 
@@ -427,11 +431,16 @@ def _assert_connection_state(self, connection_status: SubscriptionConnectionServ
                 raise DatabaseDoesNotExistException(f"{self._db_name} does not exists. {connection_status.message}")
 
         if connection_status.type_of_message != SubscriptionConnectionServerMessage.MessageType.CONNECTION_STATUS:
-            raise RuntimeError(
+            message = (
                 f"Server returned illegal type message when expecting connection status, "
                 f"was: {connection_status.type_of_message}"
             )
 
+            if connection_status.type_of_message == SubscriptionConnectionServerMessage.MessageType.ERROR:
+                message += f". Exception: {connection_status.exception}"
+
+            raise SubscriptionMessageTypeException(message)
+
         if connection_status.status == SubscriptionConnectionServerMessage.ConnectionStatus.ACCEPTED:
             pass
         elif connection_status.status == SubscriptionConnectionServerMessage.ConnectionStatus.IN_USE:
@@ -459,6 +468,18 @@ def _assert_connection_state(self, connection_status: SubscriptionConnectionServ
                 f"{connection_status.exception}"
             )
         elif connection_status.status == SubscriptionConnectionServerMessage.ConnectionStatus.REDIRECT:
+            if self._options.strategy == SubscriptionOpeningStrategy.WAIT_FOR_FREE:
+                if connection_status.data:
+                    register_connection_duration_in_ticks_object = connection_status.data.get(
+                        "RegisterConnectionDurationInTicks"
+                    )
+                    if (
+                        register_connection_duration_in_ticks_object / 10000000
+                        >= self._options.max_erroneous_period.seconds
+                    ):
+                        # this worker connection waited for free for more than max erroneous period
+                        self._last_connection_failure = None
+
             data = connection_status.data
             redirected_tag = data.get("RedirectedTag")
             appropriate_node = None if redirected_tag is None else redirected_tag
@@ -691,7 +712,9 @@ def __run_async() -> None:
                             self._forced_topology_update_attempts += 1
                             next_node_index = self._forced_topology_update_attempts % len(cur_topology)
                             try:
-                                self._redirect_node = cur_topology[next_node_index]
+                                self._redirect_node = req_ex.get_requested_node(
+                                    cur_topology[next_node_index].cluster_tag, True
+                                ).current_node
                                 self._logger.info(
                                     f"Subscription '{self._options.subscription_name}'. "
                                     f"Will modify redirect node from None to "

ravendb/exceptions/exceptions.py

@@ -107,10 +107,17 @@ class SubscriptionDoesNotExistException(SubscriptionException):
 
 
 class SubscriptionDoesNotBelongToNodeException(SubscriptionException):
-    def __init__(self, *args, appropriate_node: Optional[str] = None, reasons: Dict[str, str] = None):
+    def __init__(
+        self,
+        *args,
+        appropriate_node: Optional[str] = None,
+        reasons: Dict[str, str] = None,
+        register_connection_duration_in_ticks: bool = None,
+    ):
         super(SubscriptionDoesNotBelongToNodeException, self).__init__(*args)
         self.appropriate_node = appropriate_node
         self.reasons = reasons
+        self.register_connection_duration_in_ticks = register_connection_duration_in_ticks
 
 
 class SubscriptionChangeVectorUpdateConcurrencyException(SubscriptionException):
@@ -119,3 +126,15 @@ class SubscriptionChangeVectorUpdateConcurrencyException(SubscriptionException):
 
 class SubscriberErrorException(Exception):
     pass
+
+
+class InvalidNetworkTopologyException(Exception):
+    pass
+
+
+class SubscriptionMessageTypeException(Exception):
+    pass
+
+
+class RavenTimeoutException(Exception):
+    pass

ravendb/exceptions/raven_exceptions.py

@@ -31,3 +31,11 @@ def __init__(self, message):
 class ClientVersionMismatchException(RavenException):
     def __init__(self, message: Optional[str] = None, cause: Optional[Exception] = None):
         super().__init__(message, cause)
+
+
+class PortInUseException(RavenException):
+    pass
+
+
+class IndexCompactionInProgressException(RavenException):
+    pass

ravendb/http/request_executor.py

@@ -19,6 +19,7 @@
     UnsuccessfulRequestException,
     DatabaseDoesNotExistException,
     AuthorizationException,
+    RequestedNodeUnavailableException,
 )
 from ravendb.documents.operations.configuration import GetClientConfigurationOperation
 from ravendb.exceptions.exception_dispatcher import ExceptionDispatcher
@@ -199,9 +200,28 @@ def preferred_node(self) -> CurrentIndexAndNode:
         self.__ensure_node_selector()
         return self._node_selector.get_preferred_node()
 
+    def get_requested_node(self, node_tag: str, throw_if_contains_failures: bool = False) -> CurrentIndexAndNode:
+        self.__ensure_node_selector()
+        current_index_and_node = self._node_selector.get_requested_node(node_tag)
+
+        if throw_if_contains_failures and not self._node_selector.node_is_available(
+            current_index_and_node.current_index
+        ):
+            raise RequestedNodeUnavailableException(
+                f"Requested node {node_tag} currently unavailable, please try again later."
+            )
+
+        return current_index_and_node
+
     def __on_failed_request_invoke(self, url: str, e: Exception):
         for event in self.__on_failed_request:
-            event(FailedRequestEventArgs(self.__database_name, url, e))
+            event(FailedRequestEventArgs(self.__database_name, url, e, None, None))
+
+    def __on_failed_request_invoke_details(
+        self, url: str, e: Exception, request: Optional[requests.Request], response: Optional[requests.Response]
+    ) -> None:
+        for event in self.__on_failed_request:
+            event(FailedRequestEventArgs(self.__database_name, url, e, request, response))
 
     def __on_succeed_request_invoke(
         self, database: str, url: str, response: requests.Response, request: requests.Request, attempt_number: int
@@ -550,27 +570,22 @@ def __refresh_if_needed(self, chosen_node: ServerNode, response: requests.Respon
         refresh_topology = response.headers.get(constants.Headers.REFRESH_TOPOLOGY, False)
         refresh_client_configuration = response.headers.get(constants.Headers.REFRESH_CLIENT_CONFIGURATION, False)
 
-        if refresh_topology or refresh_client_configuration:
-            server_node = ServerNode(chosen_node.url, self.__database_name)
+        refresh_task = Future()
+        refresh_task.set_result(False)
+
+        refresh_client_configuration_task = Future()
+        refresh_client_configuration_task.set_result(None)
 
-            update_parameters = UpdateTopologyParameters(server_node)
+        if refresh_topology:
+            update_parameters = UpdateTopologyParameters(chosen_node)
             update_parameters.timeout_in_ms = 0
             update_parameters.debug_tag = "refresh-topology-header"
+            refresh_task = self.update_topology_async(update_parameters)
 
-            if refresh_topology:
-                topology_task = self.update_topology_async(update_parameters)
-            else:
-                topology_task = Future()
-                topology_task.set_result(False)
-
-            if refresh_client_configuration:
-                client_configuration = self._update_client_configuration_async(server_node)
-            else:
-                client_configuration = Future()
-                client_configuration.set_result(None)
+        if refresh_client_configuration:
+            refresh_client_configuration_task = self._update_client_configuration_async(chosen_node)
 
-            return [topology_task, client_configuration]
-        return []
+        return [refresh_task, refresh_client_configuration_task]
 
     def __send_request_to_server(
         self,
@@ -1154,7 +1169,7 @@ def __handle_server_down(
         if index_node_and_etag.current_node in command.failed_nodes:
             return False
 
-        self.__on_failed_request_invoke(url, e)
+        self.__on_failed_request_invoke(url, e, request, response)
 
         self.execute(
             index_node_and_etag.current_node, index_node_and_etag.current_index, command, should_retry, session_info

ravendb/http/server_node.py

@@ -1,5 +1,8 @@
 from enum import Enum
-from typing import Optional
+from typing import Optional, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from ravendb.http.topology import ClusterTopology
 
 
 class ServerNode:
@@ -44,6 +47,22 @@ def __hash__(self) -> int:
     def last_server_version(self) -> str:
         return self.__last_server_version
 
+    @classmethod
+    def create_from(cls, topology: "ClusterTopology"):
+        nodes = []
+        if topology is None:
+            return nodes
+
+        for key, value in topology.members.items():
+            server_node = cls(value, cluster_tag=key)
+            nodes.append(server_node)
+
+        for key, value in topology.watchers.items():
+            server_node = cls(value, cluster_tag=key)
+            nodes.append(server_node)
+
+        return nodes
+
     def should_update_server_version(self) -> bool:
         if self.last_server_version is None or self.__last_server_version_check > 100:
             return True

ravendb/http/topology.py

@@ -6,7 +6,7 @@
 import uuid
 from abc import abstractmethod
 from concurrent.futures import ThreadPoolExecutor
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional
 from typing import Union, List, Dict
 
 from ravendb.exceptions.exceptions import (
@@ -77,6 +77,16 @@ def __init__(self, topology: Topology):
             self.fastest_records = [0] * len(topology.nodes)
             self.fastest: Union[None, int] = None
             self.speed_test_mode = 0
+            self.unlikely_everyone_faulted_choice_index: Optional[int] = 0
+
+        @property
+        def node_when_everyone_marked_as_faulted(self) -> CurrentIndexAndNode:
+            index = self.unlikely_everyone_faulted_choice_index
+            self.unlikely_everyone_faulted_choice_index = (self.unlikely_everyone_faulted_choice_index + 1) % len(
+                self.nodes
+            )
+
+            return CurrentIndexAndNode(index, self.nodes[index])
 
     def __enter__(self):
         return self
@@ -94,6 +104,9 @@ def __init__(self, topology: Topology, thread_pool: ThreadPoolExecutor):
     def topology(self) -> Topology:
         return self.__state.topology
 
+    def node_is_available(self, index: int) -> bool:
+        return self.__state.failures[index] == 0
+
     def on_failed_request(self, node_index: int) -> None:
         state = self.__state
         if node_index < 0 or node_index >= len(state.failures):
@@ -117,16 +130,10 @@ def on_update_topology(self, topology: Topology, force_update: bool = False) ->
 
     def get_requested_node(self, node_tag: str) -> CurrentIndexAndNode:
         state = self.__state
-        state_failures = state.failures
         server_nodes = state.nodes
-        length = min(len(server_nodes), len(state_failures))
-        for i in range(length):
+        for i in range(len(server_nodes)):
             if server_nodes[i].cluster_tag == node_tag:
-                if state_failures[i] == 0 and server_nodes[i].url:
-                    return CurrentIndexAndNode(i, server_nodes[i])
-                raise RequestedNodeUnavailableException(
-                    f"Requested node {node_tag} is currently unavailable, please try again later."
-                )
+                return CurrentIndexAndNode(i, server_nodes[i])
 
         if len(state.nodes) == 0:
             raise DatabaseDoesNotExistException("There are no nodes in the topology at all")
@@ -142,7 +149,7 @@ def get_preferred_node_internal(cls, state: NodeSelector.__NodeSelectorState) ->
         server_nodes = state.nodes
         length = min(len(server_nodes), len(state_failures))
         for i in range(length):
-            if state_failures[0] == 0 and server_nodes[i].url:
+            if state_failures[0] == 0:
                 return CurrentIndexAndNode(i, server_nodes[i])
         return cls.unlikely_everyone_faulted_choice(state)
 
@@ -154,11 +161,11 @@ def get_preferred_node_with_topology(self) -> CurrentIndexAndNodeAndEtag:
 
     @staticmethod
     def unlikely_everyone_faulted_choice(state: NodeSelector.__NodeSelectorState) -> CurrentIndexAndNode:
-        # if there are all marked as failed, we'll chose the first
+        # if there are all marked as failed, we'll chose the next (the one in CurrentNodeIndex)
         # one so the user will get an error (or recover :-) )
         if len(state.nodes) == 0:
             raise DatabaseDoesNotExistException("There are no nodes in the topology at all")
-        return CurrentIndexAndNode(0, state.nodes[0])
+        return state.node_when_everyone_marked_as_faulted
 
     def get_node_by_session_id(self, session_id: int) -> CurrentIndexAndNode:
         state = self.__state
@@ -289,20 +296,34 @@ def __init__(self, current_index: int, current_node: ServerNode, etag: int):
 class NodeStatus:
     def __init__(
         self,
+        name: str,
         connected: bool,
         error_details: str,
         last_send: datetime.datetime,
         last_reply: datetime.datetime,
         last_sent_message: str,
         last_matching_index: int,
     ):
+        self.name = name
         self.connected = connected
         self.error_details = error_details
         self.last_send = last_send
         self.last_reply = last_reply
         self.last_sent_message = last_sent_message
         self.last_matching_index = last_matching_index
 
+    def __str__(self):
+        return (
+            "NodeStatus{"
+            f"name='{self.name}'"
+            f", connected={self.connected}"
+            f", errorDetails={self.error_details}"
+            f", lastSend={self.last_send}"
+            f", lastReply={self.last_reply}"
+            f", lastSentMessage={self.last_sent_message}"
+            "}"
+        )
+
 
 class RaftCommand:
     @abstractmethod