PYCBC-1744: Do not allow cluster to reconnect after close

Motivation
==========
We want all APIs in the SDK (acouchbase, couchbase & txcouchbase) to be
as consistent as possible.  Currently how the SDK handles attempted
operations after a close() has been called is not consistent across
operations and/or APIs.  This change is to enforce not allowing any
operations once a close() has been called.   The only way to reconnect
is to create another cluster instance.

The previous behavior would not have been good user experience b/c once
close() has completed the underlying C++ core stops its connection and
does not allow further operations.

Modification
============
* Consistenlyt raise a RuntimeError w/ a consistent message if an
  operation is attempted after close() has completed.
* close() is idempotent
* Add tests to confirm functionality (across all operations)

Results
=======
All tests pass.

Change-Id: I63df727f9c08c5c952402229729e7720de48e867
Reviewed-on: https://review.couchbase.org/c/couchbase-python-client/+/240282
Reviewed-by: Dimitris Christodoulou <dimitris.christodoulou@couchbase.com>
Tested-by: Build Bot <build@couchbase.com>

Author: thejcfactor

acouchbase/logic/bucket_impl.py

@@ -117,8 +117,9 @@ async def ping(self, req: PingRequest) -> PingResult:
 
     def view_query(self, req: ViewQueryRequest) -> ViewResult:
         """**INTERNAL**"""
+        self._client_adapter._ensure_not_closed()
         if not self.connected or not self.bucket_connected:
-            raise RuntimeError('Cannot attempt to execute a view query to establishing a connection.')
+            raise RuntimeError('Cannot perform operations without first establishing a connection.')
         # If the view_query was provided a timeout we will use that value for the streaming timeout
         # when the streaming object is created in the bindings.  If the view_query does not specify a
         # timeout, the streaming_timeout defaults to cluster's view_timeout (set here). If the cluster

acouchbase/logic/client_adapter.py

@@ -60,6 +60,7 @@ def __init__(self,
         self._binding_map = BindingMap()
         self._close_ft: Optional[Future[None]] = None
         self._connect_ft: Optional[Future[None]] = None
+        self._closed = False
         self._create_connection()
 
     @property
@@ -78,13 +79,26 @@ def connect_ft(self) -> Optional[Future[None]]:
     def loop(self) -> AbstractEventLoop:
         return self._loop
 
+    def _ensure_not_closed(self) -> None:
+        if self._closed:
+            raise RuntimeError(
+                'Cannot perform operations on a closed cluster. Create a new cluster instance to reconnect.')
+
+    def _ensure_connected(self) -> None:
+        if not self.connected:
+            raise RuntimeError('Cannot perform operations without first establishing a connection.')
+
     async def close_connection(self) -> None:
+        if self._closed:
+            return  # Already closed, idempotent behavior
+
         self._close_ft = self._execute_close_connection_request()
         await self._close_ft
+        self._closed = True
 
     def execute_bucket_request(self, req: BucketRequest) -> Future[Any]:
-        if not self.connected:
-            raise RuntimeError('Cannot attempt bucket request without first establishing a connection.')
+        self._ensure_not_closed()
+        self._ensure_connected()
 
         ft = self._loop.create_future()
 
@@ -104,6 +118,8 @@ def execute_close_bucket_request(self, bucket_name: str) -> Future[None]:
         return self.execute_bucket_request(req)
 
     def execute_cluster_request(self, req: ClusterRequest) -> Future[Any]:
+        self._ensure_not_closed()
+
         ft = self._loop.create_future()
 
         def _callback(result) -> None:
@@ -122,15 +138,16 @@ def _errback(exc) -> None:
         return ft
 
     def execute_cluster_request_sync(self, req: ClusterRequest) -> Any:
+        self._ensure_not_closed()
         req_dict = req.req_to_dict(self._connection)
         ret = self._execute_req_sync(req.op_name, req_dict)
         if isinstance(ret, BaseCouchbaseException):
             raise ErrorMapper.build_exception(ret)
         return ret
 
     def execute_collection_request(self, req: CollectionRequest) -> Future[Any]:
-        if not self.connected:
-            raise RuntimeError('Cannot attempt collection request without first establishing a connection.')
+        self._ensure_not_closed()
+        self._ensure_connected()
 
         ft = self._loop.create_future()
 
@@ -146,6 +163,8 @@ def _errback(exc) -> None:
         return ft
 
     def execute_connect_bucket_request(self, bucket_name: str) -> Future[None]:
+        self._ensure_not_closed()
+
         req = OpenBucketRequest(bucket_name, OpenOrCloseBucket.OPEN)
         ft = self._loop.create_future()
 
@@ -167,6 +186,8 @@ def _errback(ret: Any) -> None:
         return ft
 
     def execute_mgmt_request(self, req: MgmtRequest) -> Future[Any]:
+        self._ensure_not_closed()
+
         ft = self._loop.create_future()
 
         def _callback(ret: Any) -> None:
@@ -187,13 +208,15 @@ def _errback(ret: Any) -> None:
         return ft
 
     async def wait_until_connected(self) -> None:
+        self._ensure_not_closed()
         if self.connected:
             return
         if self._connect_ft is None:
             self._create_connection()
         await self._connect_ft
 
     def _create_connection(self) -> None:
+        self._ensure_not_closed()
         if self._close_ft is not None and self._close_ft.done() is not True:
             raise RuntimeError('Cannot attempt to connect when close attempt is pending.')
 
@@ -235,6 +258,11 @@ def _errback(ret: Any) -> None:
 
         req_dict = req.req_to_dict(self._connection, callback=_callback, errback=_errback)
         if not self.connected:
+            # If we're closed, don't try to reconnect just to close again
+            if self._closed:
+                ft.set_result(None)
+                return ft
+
             chained_ft = self._execute_connect_request() if self._connect_ft is None else self._connect_ft
             chained_ft.add_done_callback(partial(self._execute_chained_req, ft, req.op_name, req_dict))
         else:

acouchbase/logic/cluster_impl.py

@@ -145,8 +145,8 @@ def transactions(self) -> Transactions:
 
     def analytics_query(self, req: AnalyticsQueryRequest) -> AnalyticsResult:
         """**INTERNAL**"""
-        if not self.connected:
-            raise RuntimeError('Cannot attempt to execute an analytics query prior to establishing a connection.')
+        self._client_adapter._ensure_not_closed()
+        self._client_adapter._ensure_connected()
         # If the analytics_query was provided a timeout we will use that value for the streaming timeout
         # when the streaming object is created in the bindings.  If the analytics_query does not specify a
         # timeout, the streaming_timeout defaults to cluster's analytics_timeout (set here). If the cluster
@@ -196,8 +196,8 @@ async def ping(self, req: PingRequest) -> PingResult:
 
     def query(self, req: QueryRequest) -> QueryResult:
         """**INTERNAL**"""
-        if not self.connected:
-            raise RuntimeError('Cannot attempt to execute a query prior to establishing a connection.')
+        self._client_adapter._ensure_not_closed()
+        self._client_adapter._ensure_connected()
         # If the n1ql_query was provided a timeout we will use that value for the streaming timeout
         # when the streaming object is created in the bindings.  If the n1ql_query does not specify a
         # timeout, the streaming_timeout defaults to cluster's query_timeout (set here). If the cluster
@@ -213,8 +213,8 @@ def query(self, req: QueryRequest) -> QueryResult:
 
     def search(self, req: SearchQueryRequest) -> SearchResult:
         """**INTERNAL**"""
-        if not self.connected:
-            raise RuntimeError('Cannot attempt to execute a search prior to establishing a connection.')
+        self._client_adapter._ensure_not_closed()
+        self._client_adapter._ensure_connected()
         # If the search_query was provided a timeout we will use that value for the streaming timeout
         # when the streaming object is created in the bindings.  If the search_query does not specify a
         # timeout, the streaming_timeout defaults to cluster's search_timeout (set here). If the cluster

acouchbase/logic/scope_impl.py

@@ -101,8 +101,8 @@ def name(self) -> str:
         return self._scope_name
 
     def analytics_query(self, req: AnalyticsQueryRequest) -> AnalyticsResult:
-        if not self.connected:
-            raise RuntimeError('Cannot attempt to execute an analytics query prior to establishing a connection.')
+        self._client_adapter._ensure_not_closed()
+        self._client_adapter._ensure_connected()
         # If the analytics_query was provided a timeout we will use that value for the streaming timeout
         # when the streaming object is created in the bindings.  If the analytics_query does not specify a
         # timeout, the streaming_timeout defaults to cluster's analytics_timeout (set here). If the cluster
@@ -117,8 +117,8 @@ def analytics_query(self, req: AnalyticsQueryRequest) -> AnalyticsResult:
                                                                                 num_workers=req.num_workers))
 
     def query(self, req: QueryRequest) -> QueryResult:
-        if not self.connected:
-            raise RuntimeError('Cannot attempt to execute a query prior to establishing a connection.')
+        self._client_adapter._ensure_not_closed()
+        self._client_adapter._ensure_connected()
         # If the n1ql_query was provided a timeout we will use that value for the streaming timeout
         # when the streaming object is created in the bindings.  If the n1ql_query does not specify a
         # timeout, the streaming_timeout defaults to cluster's query_timeout (set here). If the cluster
@@ -133,8 +133,8 @@ def query(self, req: QueryRequest) -> QueryResult:
                                                                   num_workers=req.num_workers))
 
     def search(self, req: SearchQueryRequest) -> SearchResult:
-        if not self.connected:
-            raise RuntimeError('Cannot attempt to execute a search prior to establishing a connection.')
+        self._client_adapter._ensure_not_closed()
+        self._client_adapter._ensure_connected()
         # If the search_query was provided a timeout we will use that value for the streaming timeout
         # when the streaming object is created in the bindings.  If the search_query does not specify a
         # timeout, the streaming_timeout defaults to cluster's search_timeout (set here). If the cluster

acouchbase/tests/cluster_t.py

@@ -20,7 +20,8 @@
 import pytest
 import pytest_asyncio
 
-from acouchbase.cluster import get_event_loop
+from acouchbase.cluster import Cluster, get_event_loop
+from couchbase.auth import PasswordAuthenticator
 from couchbase.diagnostics import (ClusterState,
                                    EndpointPingReport,
                                    EndpointState,
@@ -30,7 +31,9 @@
                                   InvalidArgumentException,
                                   ParsingFailedException,
                                   QueryIndexNotFoundException)
-from couchbase.options import DiagnosticsOptions, PingOptions
+from couchbase.options import (ClusterOptions,
+                               DiagnosticsOptions,
+                               PingOptions)
 from couchbase.result import DiagnosticsResult, PingResult
 
 from ._test_utils import TestEnvironment
@@ -218,3 +221,139 @@ async def test_diagnostics_as_json(self, cb_env):
                 assert data[0]['remote'] is not None
                 assert data[0]['local'] is not None
                 assert data[0]['state'] is not None
+
+    @pytest.mark.asyncio
+    async def test_await_on_close_cluster(self, cb_env):
+        conn_string = cb_env.config.get_connection_string()
+        username, pw = cb_env.config.get_username_and_pw()
+        auth = PasswordAuthenticator(username, pw)
+        opts = ClusterOptions(auth)
+        cluster = Cluster(conn_string, opts)
+        assert cluster._impl.connected is False
+        await cluster.close()
+        assert cluster._impl.connected is False
+
+    @pytest.mark.asyncio
+    async def test_multiple_close_cluster(self, cb_env):
+        conn_string = cb_env.config.get_connection_string()
+        username, pw = cb_env.config.get_username_and_pw()
+        auth = PasswordAuthenticator(username, pw)
+        opts = ClusterOptions(auth)
+        cluster = await Cluster.connect(conn_string, opts)
+        assert cluster._impl.connected is True
+        for _ in range(10):
+            await cluster.close()
+        assert cluster._impl.connected is False
+
+    @pytest.mark.asyncio
+    async def test_operations_after_close(self, cb_env):
+        conn_string = cb_env.config.get_connection_string()
+        username, pw = cb_env.config.get_username_and_pw()
+        auth = PasswordAuthenticator(username, pw)
+        opts = ClusterOptions(auth)
+        cluster = await Cluster.connect(conn_string, opts)
+        assert cluster._impl.connected is True
+
+        # Close the cluster
+        await cluster.close()
+        assert cluster._impl.connected is False
+
+        # Verify operations after close raise RuntimeError
+        with pytest.raises(RuntimeError, match="Cannot perform operations on a closed cluster"):
+            await cluster.ping()
+
+        with pytest.raises(RuntimeError, match="Cannot perform operations on a closed cluster"):
+            await cluster.diagnostics()
+
+        with pytest.raises(RuntimeError, match="Cannot perform operations on a closed cluster"):
+            cluster.bucket(cb_env.bucket.name)
+
+    @pytest.mark.asyncio
+    async def test_connected_property(self, cb_env):
+        conn_string = cb_env.config.get_connection_string()
+        username, pw = cb_env.config.get_username_and_pw()
+        auth = PasswordAuthenticator(username, pw)
+        opts = ClusterOptions(auth)
+        cluster = await Cluster.connect(conn_string, opts)
+        assert cluster._impl.connected is True
+
+        # Close the cluster
+        await cluster.close()
+
+        assert cluster._impl.connected is False
+
+    @pytest.mark.asyncio
+    async def test_streaming_operations_after_close(self, cb_env):
+        """Verify all streaming operations raise RuntimeError after cluster close"""
+        from couchbase.search import MatchQuery
+
+        conn_string = cb_env.config.get_connection_string()
+        username, pw = cb_env.config.get_username_and_pw()
+        auth = PasswordAuthenticator(username, pw)
+        opts = ClusterOptions(auth)
+        cluster = await Cluster.connect(conn_string, opts)
+
+        # Close the cluster
+        await cluster.close()
+
+        # Test cluster.query
+        with pytest.raises(RuntimeError, match="Cannot perform operations on a closed cluster"):
+            cluster.query("SELECT 1=1")
+
+        # Test cluster.analytics_query
+        with pytest.raises(RuntimeError, match="Cannot perform operations on a closed cluster"):
+            cluster.analytics_query("SELECT 1=1")
+
+        # Test cluster.search
+        with pytest.raises(RuntimeError, match="Cannot perform operations on a closed cluster"):
+            cluster.search_query("dummy-index", MatchQuery("test"))
+
+        # Test bucket.view_query
+        with pytest.raises(RuntimeError, match="Cannot perform operations on a closed cluster"):
+            bucket = cluster.bucket(cb_env.bucket.name)
+            bucket.view_query("dummy-design", "dummy-view")
+
+    @pytest.mark.flaky(reruns=5, reruns_delay=1)
+    @pytest.mark.asyncio
+    async def test_management_operations_after_close(self, cb_env):
+        from couchbase.management.views import DesignDocumentNamespace
+
+        conn_string = cb_env.config.get_connection_string()
+        username, pw = cb_env.config.get_username_and_pw()
+        auth = PasswordAuthenticator(username, pw)
+        opts = ClusterOptions(auth)
+        cluster = await Cluster.connect(conn_string, opts)
+        bucket_name = cb_env.bucket.name
+
+        # Get bucket instance before closing (for bucket-level manager tests)
+        bucket = cluster.bucket(bucket_name)
+
+        # Close the cluster
+        await cluster.close()
+        assert cluster._impl.connected is False
+
+        # Test cluster-level management APIs
+        with pytest.raises(RuntimeError, match="Cannot perform operations on a closed cluster"):
+            await cluster.buckets().get_all_buckets()
+
+        with pytest.raises(RuntimeError, match="Cannot perform operations on a closed cluster"):
+            await cluster.users().get_all_users()
+
+        with pytest.raises(RuntimeError, match="Cannot perform operations on a closed cluster"):
+            await cluster.query_indexes().get_all_indexes(bucket_name)
+
+        with pytest.raises(RuntimeError, match="Cannot perform operations on a closed cluster"):
+            await cluster.analytics_indexes().get_all_datasets()
+
+        with pytest.raises(RuntimeError, match="Cannot perform operations on a closed cluster"):
+            await cluster.search_indexes().get_all_indexes()
+
+        with pytest.raises(RuntimeError, match="Cannot perform operations on a closed cluster"):
+            await cluster.eventing_functions().get_all_functions()
+
+        # Test bucket-level management APIs
+        with pytest.raises(RuntimeError, match="Cannot perform operations on a closed cluster"):
+            await bucket.collections().get_all_scopes()
+
+        with pytest.raises(RuntimeError, match="Cannot perform operations on a closed cluster"):
+            await bucket.view_indexes().get_all_design_documents(DesignDocumentNamespace.DEVELOPMENT)

couchbase/logic/bucket_impl.py

@@ -92,6 +92,8 @@ def ping(self, req: PingRequest) -> PingResult:
 
     def view_query(self, req: ViewQueryRequest) -> ViewResult:
         """**INTERNAL**"""
+        self._client_adapter._ensure_not_closed()
+        self._client_adapter._ensure_connected()
         # If the view_query was provided a timeout we will use that value for the streaming timeout
         # when the streaming object is created in the bindings.  If the view_query does not specify a
         # timeout, the streaming_timeout defaults to cluster's view_timeout (set here). If the cluster