Protocol versioning (#297)

## What is the goal of this PR?

We use a new protocol API to perform a "connection open". This API does server-side protocol version compatibility checks, and replaces our previous need to get all databases to check that the connection is available.

This API is called transparently during the construction of both the Core and Cluster clients.

## What are the changes implemented in this PR?

* Refactor Stubs to call `connectionOpen` in their constructors
* Throw a useful message if the server returns an `unimplemented` error

Author: flyingsilverfin

dependencies/vaticle/artifacts.bzl

@@ -29,7 +29,7 @@ def vaticle_typedb_artifacts():
         artifact_name = "typedb-server-{platform}-{version}.{ext}",
         tag_source = deployment["artifact.release"],
         commit_source = deployment["artifact.snapshot"],
-        commit = "cb119c536c44275f58df7b54da033f8fe076cac5",
+        commit = "ad86a23fb78dda097a8b61b9d01ef68751cb14c1",
     )
 
 def vaticle_typedb_cluster_artifacts():
@@ -39,5 +39,5 @@ def vaticle_typedb_cluster_artifacts():
         artifact_name = "typedb-cluster-all-{platform}-{version}.{ext}",
         tag_source = deployment_private["artifact.release"],
         commit_source = deployment_private["artifact.snapshot"],
-        commit = "8044753056dd20b8e6bec6f62036eb0003d4ba06",
+        commit = "6e00e0bff33c8c9c90f6393620dd30a9526748c3",
     )

requirements.txt

@@ -35,7 +35,7 @@
 
 ## Dependencies
 
-typedb-protocol==2.18.0.dev2
+typedb-protocol==2.18.0.dev3
 grpcio>=1.43.0,<2
 protobuf>=3.15.6,<4
 parse==1.18.0

typedb/common/exception.py

@@ -18,7 +18,7 @@
 # specific language governing permissions and limitations
 # under the License.
 #
-from typing import Optional, Union, Any
+from typing import Union, Any
 
 from grpc import RpcError, Call, StatusCode
 
@@ -38,7 +38,9 @@ def __init__(self, msg: Union["ErrorMessage", str], cause: BaseException = None,
 
     @staticmethod
     def of_rpc(rpc_error: Union[RpcError, Call]) -> "TypeDBClientException":
-        if rpc_error.code() in [StatusCode.UNAVAILABLE, StatusCode.UNKNOWN] or "Received RST_STREAM" in str(rpc_error):
+        if rpc_error.code() is StatusCode.UNIMPLEMENTED:
+            return TypeDBClientException(msg=RPC_METHOD_UNAVAILABLE, cause=rpc_error.details())
+        elif rpc_error.code() in [StatusCode.UNAVAILABLE, StatusCode.UNKNOWN] or "Received RST_STREAM" in str(rpc_error):
             return TypeDBClientException(msg=UNABLE_TO_CONNECT, cause=rpc_error)
         elif rpc_error.code() is StatusCode.INTERNAL and "[RPL01]" in str(rpc_error):
             return TypeDBClientException(msg=CLUSTER_REPLICA_NOT_PRIMARY, cause=None)
@@ -75,24 +77,25 @@ def __init__(self, code: int, message: str):
         super(ClientErrorMessage, self).__init__(code_prefix="CLI", code_number=code, message_prefix="Client Error", message_body=message)
 
 
-CLIENT_CLOSED = ClientErrorMessage(1, "The client has been closed and no further operation is allowed.")
-SESSION_CLOSED = ClientErrorMessage(2, "The session has been closed and no further operation is allowed.")
-TRANSACTION_CLOSED = ClientErrorMessage(3, "The transaction has been closed and no further operation is allowed.")
-TRANSACTION_CLOSED_WITH_ERRORS = ClientErrorMessage(4, "The transaction has been closed with error(s):\n%s.")
-UNABLE_TO_CONNECT = ClientErrorMessage(5, "Unable to connect to TypeDB server.")
-NEGATIVE_VALUE_NOT_ALLOWED = ClientErrorMessage(6, "Value cannot be less than 1, was: '%d'.")
-MISSING_DB_NAME = ClientErrorMessage(7, "Database name cannot be empty.")
-DB_DOES_NOT_EXIST = ClientErrorMessage(8, "The database '%s' does not exist.")
-MISSING_RESPONSE = ClientErrorMessage(9, "Unexpected empty response for request ID '%s'.")
-UNKNOWN_REQUEST_ID = ClientErrorMessage(10, "Received a response with unknown request id '%s':\n%s")
-CLUSTER_NO_PRIMARY_REPLICA_YET = ClientErrorMessage(11, "No replica has been marked as the primary replica for latest known term '%d'.")
-CLUSTER_UNABLE_TO_CONNECT = ClientErrorMessage(12, "Unable to connect to TypeDB Cluster. Attempted connecting to the cluster members, but none are available: '%s'.")
-CLUSTER_REPLICA_NOT_PRIMARY = ClientErrorMessage(13, "The replica is not the primary replica.")
-CLUSTER_ALL_NODES_FAILED = ClientErrorMessage(14, "Attempted connecting to all cluster members, but the following errors occurred: \n%s")
-CLUSTER_USER_DOES_NOT_EXIST = ClientErrorMessage(15, "The user '%s' does not exist.")
-CLUSTER_TOKEN_CREDENTIAL_INVALID = ClientErrorMessage(16, "Invalid token credential.")
-CLUSTER_INVALID_ROOT_CA_PATH = ClientErrorMessage(17, "The provided Root CA path '%s' does not exist.")
-CLUSTER_CLIENT_CALLED_WITH_STRING = ClientErrorMessage(18, "The first argument of TypeDBClient.cluster() must be a List of server addresses to connect to. It was called with a string, not a List, which is not allowed.")
+RPC_METHOD_UNAVAILABLE = ClientErrorMessage(1, "The server does not support this method, please check the client-server compatibility:\n'%s'.")
+CLIENT_NOT_OPEN = ClientErrorMessage(2, "The client is not open.")
+SESSION_CLOSED = ClientErrorMessage(3, "The session has been closed and no further operation is allowed.")
+TRANSACTION_CLOSED = ClientErrorMessage(4, "The transaction has been closed and no further operation is allowed.")
+TRANSACTION_CLOSED_WITH_ERRORS = ClientErrorMessage(5, "The transaction has been closed with error(s):\n%s.")
+UNABLE_TO_CONNECT = ClientErrorMessage(6, "Unable to connect to TypeDB server.")
+NEGATIVE_VALUE_NOT_ALLOWED = ClientErrorMessage(7, "Value cannot be less than 1, was: '%d'.")
+MISSING_DB_NAME = ClientErrorMessage(8, "Database name cannot be empty.")
+DB_DOES_NOT_EXIST = ClientErrorMessage(9, "The database '%s' does not exist.")
+MISSING_RESPONSE = ClientErrorMessage(10, "Unexpected empty response for request ID '%s'.")
+UNKNOWN_REQUEST_ID = ClientErrorMessage(11, "Received a response with unknown request id '%s':\n%s")
+CLUSTER_NO_PRIMARY_REPLICA_YET = ClientErrorMessage(12, "No replica has been marked as the primary replica for latest known term '%d'.")
+CLUSTER_UNABLE_TO_CONNECT = ClientErrorMessage(13, "Unable to connect to TypeDB Cluster. Attempted connecting to the cluster members, but none are available: '%s'.")
+CLUSTER_REPLICA_NOT_PRIMARY = ClientErrorMessage(14, "The replica is not the primary replica.")
+CLUSTER_ALL_NODES_FAILED = ClientErrorMessage(15, "Attempted connecting to all cluster members, but the following errors occurred: \n%s")
+CLUSTER_USER_DOES_NOT_EXIST = ClientErrorMessage(16, "The user '%s' does not exist.")
+CLUSTER_TOKEN_CREDENTIAL_INVALID = ClientErrorMessage(17, "Invalid token credential.")
+CLUSTER_INVALID_ROOT_CA_PATH = ClientErrorMessage(18, "The provided Root CA path '%s' does not exist.")
+CLUSTER_CLIENT_CALLED_WITH_STRING = ClientErrorMessage(19, "The first argument of TypeDBClient.cluster() must be a List of server addresses to connect to. It was called with a string, not a List, which is not allowed.")
 
 
 class ConceptErrorMessage(ErrorMessage):

typedb/common/rpc/request_builder.py

@@ -25,6 +25,8 @@
 import typedb_protocol.cluster.cluster_server_pb2 as cluster_server_proto
 import typedb_protocol.cluster.cluster_user_pb2 as cluster_user_proto
 import typedb_protocol.common.concept_pb2 as concept_proto
+import typedb_protocol.common.connection_pb2 as connection_proto
+import typedb_protocol.common.version_pb2 as version_proto
 import typedb_protocol.common.logic_pb2 as logic_proto
 import typedb_protocol.common.options_pb2 as options_proto
 import typedb_protocol.common.query_pb2 as query_proto
@@ -35,6 +37,13 @@
 from typedb.common.exception import TypeDBClientException, GET_HAS_WITH_MULTIPLE_FILTERS
 from typedb.common.label import Label
 
+# Connection
+
+def connection_open_req():
+    req = connection_proto.Connection.Open.Req()
+    req.version = version_proto.Version.VERSION
+    return req
+
 
 # CoreDatabaseManager
 

typedb/common/rpc/stub.py

@@ -23,6 +23,7 @@
 from typing import Iterator
 from typing import TypeVar, Callable
 
+import typedb_protocol.common.connection_pb2 as connection_proto
 import typedb_protocol.common.session_pb2 as session_proto
 import typedb_protocol.common.transaction_pb2 as transaction_proto
 import typedb_protocol.core.core_database_pb2 as core_database_proto

typedb/connection/client.py

@@ -27,6 +27,7 @@
 from typedb.api.connection.options import TypeDBOptions
 from typedb.api.connection.session import SessionType
 from typedb.common.concurrent.scheduled_executor import ScheduledExecutor
+from typedb.common.exception import CLIENT_NOT_OPEN, TypeDBClientException
 from typedb.common.rpc.stub import TypeDBStub
 from typedb.connection.database_manager import _TypeDBDatabaseManagerImpl
 from typedb.connection.session import _TypeDBSessionImpl
@@ -42,11 +43,13 @@ def __init__(self, address: str, parallelisation: int = 2):
         self._transmitter = RequestTransmitter(parallelisation)
         self._sessions: Dict[bytes, _TypeDBSessionImpl] = {}
         self._sessions_lock = Lock()
-        self._is_open = True
         self._pulse_executor = ScheduledExecutor()
         self._pulse_executor.schedule_at_fixed_rate(interval=self._PULSE_INTERVAL_SECONDS, action=self._transmit_pulses)
 
     def session(self, database: str, session_type: SessionType, options=None) -> _TypeDBSessionImpl:
+        if not self.is_open():
+            raise TypeDBClientException.of(CLIENT_NOT_OPEN)
+
         if not options:
             options = TypeDBOptions.core()
         session = _TypeDBSessionImpl(self, database, session_type, options)

typedb/connection/cluster/client.py

@@ -32,7 +32,7 @@
 from typedb.connection.cluster.stub import _ClusterServerStub
 from typedb.connection.cluster.user_manager import _ClusterUserManager
 from typedb.common.rpc.request_builder import cluster_server_manager_all_req
-from typedb.common.exception import TypeDBClientException, UNABLE_TO_CONNECT, CLUSTER_UNABLE_TO_CONNECT
+from typedb.common.exception import TypeDBClientException, UNABLE_TO_CONNECT, CLUSTER_UNABLE_TO_CONNECT, CLIENT_NOT_OPEN
 
 
 class _ClusterClient(TypeDBClusterClient):
@@ -62,9 +62,12 @@ def _fetch_server_addresses(self, addresses: Iterable[str]) -> Set[str]:
         raise TypeDBClientException.of(CLUSTER_UNABLE_TO_CONNECT, ",".join(addresses))
 
     def session(self, database: str, session_type: SessionType, options=None) -> _ClusterSession:
+        if not self.is_open():
+            raise TypeDBClientException.of(CLIENT_NOT_OPEN)
         if not options:
             options = TypeDBOptions.cluster()
-        return self._session_any_replica(database, session_type, options) if getattr(options, "read_any_replica", False) else self._session_primary_replica(database, session_type, options)
+        return self._session_any_replica(database, session_type, options) if getattr(options, "read_any_replica", False) \
+            else self._session_primary_replica(database, session_type, options)
 
     def _session_primary_replica(self, database: str, session_type: SessionType, options=None) -> _ClusterSession:
         return _OpenSessionFailsafeTask(database, session_type, options, self).run_primary_replica()

typedb/connection/cluster/server_client.py

@@ -40,6 +40,7 @@ def __init__(self, address: str, credential: TypeDBCredential, parallelisation:
             self._channel_credentials = grpc.ssl_channel_credentials()
         self._channel, self._stub = self.new_channel_and_stub()
         self._databases = _TypeDBDatabaseManagerImpl(self.stub())
+        self._is_open = True
 
     def databases(self) -> _TypeDBDatabaseManagerImpl:
         return self._databases

typedb/connection/cluster/stub.py

@@ -33,7 +33,7 @@
 
 from typedb.api.connection.credential import TypeDBCredential
 from typedb.common.exception import CLUSTER_TOKEN_CREDENTIAL_INVALID, TypeDBClientException, UNABLE_TO_CONNECT
-from typedb.common.rpc.request_builder import cluster_user_token_req
+from typedb.common.rpc.request_builder import cluster_user_token_req, connection_open_req
 from typedb.common.rpc.stub import TypeDBStub
 
 T = TypeVar('T')
@@ -47,8 +47,10 @@ def __init__(self, channel: Channel, credential: TypeDBCredential):
         self._channel = channel
         self._stub = core_service_proto.TypeDBStub(channel)
         self._cluster_stub = cluster_service_proto.TypeDBClusterStub(channel)
+
         self._token = None
         try:
+            self._stub.connection_open(connection_open_req())
             res = self._cluster_stub.user_token(cluster_user_token_req(self._credential.username()))
             self._token = res.token
         except RpcError as e:

typedb/connection/core/client.py

@@ -21,6 +21,7 @@
 
 from grpc import Channel, insecure_channel
 
+from typedb.common.exception import TypeDBClientException, CLIENT_NOT_OPEN
 from typedb.common.rpc.stub import TypeDBStub
 from typedb.connection.client import _TypeDBClientImpl
 from typedb.connection.core.stub import _CoreStub
@@ -33,8 +34,11 @@ def __init__(self, address: str, parallelisation: int = 2):
         super(_CoreClient, self).__init__(address, parallelisation)
         self._channel, self._stub = self.new_channel_and_stub()
         self._databases = _TypeDBDatabaseManagerImpl(self.stub())
+        self._is_open = True
 
     def databases(self) -> _TypeDBDatabaseManagerImpl:
+        if not self.is_open():
+            raise TypeDBClientException.of(CLIENT_NOT_OPEN)
         return self._databases
 
     def channel(self) -> Channel: