PYCBC-1928: Hold the close paths to one connection lifecycle model

Changes
--------
* Gate the lazily constructed Transactions property on the same two
  guards every other cluster operation uses, in both trees, so a closed
  or never-connected cluster no longer hands back a live Transactions
  owning cleanup threads and a client-record registration
* Stop close waiting on a connection it has to start, in the async
  cluster and bucket paths both, and stop a failed connect or open
  coming back out of close(): wait only for one already in flight, so a
  connection that comes up late is still torn down, and treat its
  failure as nothing to tear down
* Make bucket close idempotent in both trees, matching cluster close,
  which already returns rather than raising on a closed or
  never-connected cluster
* Add lifecycle suites for both trees, needing no cluster, and rework
  the transactions race suite for the new guard without losing its
  PYCBC-1821 coverage: removing the double-checked lock still fails it

Change-Id: I64bb70abb0fd56b859b3c98113e1ff129a51c2ac
Reviewed-on: https://review.couchbase.org/c/couchbase-python-client/+/252028
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Sergey Avseyev <sergey.avseyev@gmail.com>

Author: thejcfactor

acouchbase/logic/bucket_impl.py

@@ -110,7 +110,12 @@ def observability_instruments(self) -> ObservabilityInstruments:
 
     async def close_bucket(self) -> None:
         """**INTERNAL**"""
-        await self.wait_until_bucket_connected()
+        try:
+            await self.wait_until_bucket_connected()
+        except Exception:
+            # The bucket never opened, so there is nothing to tear down, and the caller cannot
+            # act on an open error raised out of close.
+            return
         await self._client_adapter.execute_close_bucket_request(self._bucket_name)
 
     async def ping(self, req: PingRequest) -> PingResult:

acouchbase/logic/client_adapter.py

@@ -130,8 +130,12 @@ def _errback(exc) -> None:
         return ft
 
     def execute_close_bucket_request(self, bucket_name: str) -> Future[None]:
-        req = CloseBucketRequest(bucket_name)
-        return self.execute_bucket_request(req)
+        if self._closed or not self.connected:
+            # execute_bucket_request guards for operations, which close is not
+            ft = self.loop.create_future()
+            ft.set_result(None)
+            return ft
+        return self.execute_bucket_request(CloseBucketRequest(bucket_name))
 
     def execute_cluster_request(self, req: ClusterRequest) -> Future[Any]:
         self._ensure_not_closed()
@@ -299,15 +303,27 @@ def _errback(ret: Any) -> None:
 
         req_dict = req.req_to_dict(callback=_callback, errback=_errback)
         if not self.connected:
-            # If we're closed, don't try to reconnect just to close again
-            if self._closed:
+            if self._connect_ft is None:
+                # Never attempted, so there is nothing to tear down.  Starting a connection in
+                # order to close it is the one thing close must not do.
+                self._reset_connection()
                 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:
-            self._execute_req(ft, req.op_name, req_dict)
+            # A connect is already in flight.  Wait for it, so a connection that comes up after
+            # this call is still torn down, but never let its failure become a close failure:
+            # nothing came up, and the caller cannot act on a connect error raised out of close.
+            def _close_once_connected(chained_ft: Future[None]) -> None:
+                if chained_ft.cancelled() or chained_ft.exception() is not None:
+                    self._reset_connection()
+                    ft.set_result(None)
+                    return
+                self._execute_req(ft, req.op_name, req_dict)
+
+            self._connect_ft.add_done_callback(_close_once_connected)
+            return ft
+
+        self._execute_req(ft, req.op_name, req_dict)
         return ft
 
     def _execute_connect_request(self) -> Future[None]:

acouchbase/logic/cluster_impl.py

@@ -160,6 +160,8 @@ def transactions(self) -> Transactions:
         # Transactions.__init__ goes straight into the synchronous create_transactions (no executor
         # anywhere in the async path), which releases the GIL for a full cluster round-trip.
         # Use threading.Lock, not asyncio.Lock, for the same reason.
+        self._client_adapter._ensure_not_closed()
+        self._client_adapter._ensure_connected()
         if self._transactions is None:
             with self._transactions_lock:
                 if self._transactions is None:

acouchbase/tests/connection_lifecycle_t.py

@@ -0,0 +1,119 @@
+#  Copyright 2016-2026. Couchbase, Inc.
+#  All Rights Reserved.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License")
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+"""
+The async close paths, held to the same lifecycle model as the blocking API: an operation
+raises on a cluster that never connected and on one that is closed, and a close is
+idempotent and never starts a connection in order to tear one down.
+
+The async tree makes the last of those easy to get wrong, because a connect is normally
+already in flight by the time close is called, so close has a future to chain onto.
+Chaining is right only while that connect may still succeed; a connect that failed leaves
+nothing to close, and its error must not come back out of close().
+
+No live cluster is needed.  Every cluster here is built with
+skip_connect='TEST_SKIP_CONNECT', so no connect is ever started and the cluster sits in
+the never-connected state.
+"""
+
+import pytest
+
+from acouchbase.logic.cluster_impl import AsyncClusterImpl
+from couchbase.auth import PasswordAuthenticator
+from couchbase.options import ClusterOptions
+
+BUCKET_NAME = 'default'
+
+
+class AsyncConnectionLifecycleTestSuite:
+    TEST_MANIFEST = [
+        'test_close_bucket_after_cluster_close_is_a_noop',
+        'test_close_bucket_when_never_connected_is_a_noop',
+        'test_close_connection_is_idempotent',
+        'test_close_connection_when_never_connected_does_not_connect',
+        'test_close_does_not_surface_a_failed_connect',
+        'test_transactions_raises_when_never_connected',
+    ]
+
+    @pytest.fixture(name='unconnected_cluster')
+    def cluster_without_core_connection(self, couchbase_config):
+        conn_string = couchbase_config.get_connection_string()
+        username, pw = couchbase_config.get_username_and_pw()
+        return AsyncClusterImpl(conn_string,
+                                ClusterOptions(PasswordAuthenticator(username, pw)),
+                                skip_connect='TEST_SKIP_CONNECT')
+
+    @pytest.mark.asyncio
+    async def test_close_connection_when_never_connected_does_not_connect(self, unconnected_cluster):
+        adapter = unconnected_cluster.client_adapter
+        adapter._execute_connect_request = _fail_if_called
+
+        await adapter.close_connection()
+
+        assert adapter._closed is True
+
+    @pytest.mark.asyncio
+    async def test_close_connection_is_idempotent(self, unconnected_cluster):
+        adapter = unconnected_cluster.client_adapter
+        adapter._execute_connect_request = _fail_if_called
+
+        await adapter.close_connection()
+        await adapter.close_connection()
+
+        assert adapter._closed is True
+
+    @pytest.mark.asyncio
+    async def test_close_does_not_surface_a_failed_connect(self, unconnected_cluster):
+        adapter = unconnected_cluster.client_adapter
+        failed = adapter.loop.create_future()
+        failed.set_exception(RuntimeError('connect failed'))
+        adapter._connect_ft = failed
+
+        await adapter.close_connection()
+
+        assert adapter._closed is True
+
+    @pytest.mark.asyncio
+    async def test_close_bucket_when_never_connected_is_a_noop(self, unconnected_cluster):
+        await unconnected_cluster.client_adapter.execute_close_bucket_request(BUCKET_NAME)
+
+    @pytest.mark.asyncio
+    async def test_close_bucket_after_cluster_close_is_a_noop(self, unconnected_cluster):
+        adapter = unconnected_cluster.client_adapter
+        adapter._execute_connect_request = _fail_if_called
+        await adapter.close_connection()
+
+        await adapter.execute_close_bucket_request(BUCKET_NAME)
+
+    @pytest.mark.asyncio
+    async def test_transactions_raises_when_never_connected(self, unconnected_cluster):
+        with pytest.raises(RuntimeError):
+            unconnected_cluster.transactions
+
+
+def _fail_if_called(*args, **kwargs):
+    raise AssertionError('close started a connection in order to tear one down')
+
+
+class ClassicAsyncConnectionLifecycleTests(AsyncConnectionLifecycleTestSuite):
+    @pytest.fixture(scope='class', autouse=True)
+    def validate_test_manifest(self):
+        def valid_test_method(meth):
+            attr = getattr(ClassicAsyncConnectionLifecycleTests, meth)
+            return callable(attr) and not meth.startswith('__') and meth.startswith('test')
+        method_list = [meth for meth in dir(ClassicAsyncConnectionLifecycleTests) if valid_test_method(meth)]
+        manifest_invalid = set(AsyncConnectionLifecycleTestSuite.TEST_MANIFEST).symmetric_difference(method_list)
+        if manifest_invalid:
+            pytest.fail(f'Test manifest not validated.  Missing/extra tests: {manifest_invalid}.')

acouchbase/tests/transactions_property_race_t.py

@@ -25,20 +25,20 @@
 cluster round-trip, so the racers here are OS threads, not tasks -- the guard
 is a threading.Lock for the same reason.
 
-No live cluster is needed.  The property never touches the network before
-constructing a Transactions object, so the cluster is built with
+No live cluster is needed.  The cluster is built with
 skip_connect='TEST_SKIP_CONNECT' (same mechanism couchbase/tests/connection_t.py
 uses) and exercised purely at the Python layer, with Transactions swapped for a
 fake that holds the construction window open long enough to hit the race
-reliably.
+reliably.  The property gates on the connection before it constructs anything,
+so the race fixtures stub that guard out; the guard itself is covered separately
+below.
 """
 
 import threading
 
 import pytest
 
 from acouchbase.cluster import AsyncCluster
-from acouchbase.logic.client_adapter import AsyncClientAdapter
 from couchbase.auth import PasswordAuthenticator
 from couchbase.options import ClusterOptions
 
@@ -53,6 +53,8 @@ class TransactionsPropertyRaceTestSuite:
     TEST_MANIFEST = [
         'test_close_during_first_access_closes_instance',
         'test_concurrent_first_access_returns_single_instance',
+        'test_transactions_raises_when_closed',
+        'test_transactions_raises_when_not_connected',
     ]
 
     @pytest.fixture(name='tracking_transactions')
@@ -93,52 +95,57 @@ def cluster_without_core_connection(self, couchbase_config):
                             ClusterOptions(PasswordAuthenticator(username, pw)),
                             skip_connect='TEST_SKIP_CONNECT')
 
-    @pytest.fixture(name='stub_adapter_close')
-    def stub_async_adapter_close(self, monkeypatch):
-        """Stub out the core half of close_connection().
+    @pytest.fixture(name='race_cluster')
+    def cluster_with_the_connection_guard_stubbed(self, unconnected_cluster, monkeypatch):
+        """The race is in the property's check-then-set, not in its guard."""
+        monkeypatch.setattr(unconnected_cluster._impl._client_adapter, '_ensure_connected', lambda: None)
+        return unconnected_cluster
 
-        When not connected, AsyncClientAdapter chains a connect request ahead of the
-        close (client_adapter.py:301), so under skip_connect the close would establish
-        the very connection this test skipped.  The sync adapter early-outs instead and
-        needs no stub, so this keeps both tests network-free and asserting only the
-        lock ordering.
-        """
+    @pytest.mark.asyncio
+    async def test_transactions_raises_when_not_connected(self, tracking_transactions, unconnected_cluster):
+        with pytest.raises(RuntimeError):
+            unconnected_cluster.transactions
+
+        assert tracking_transactions.instances_created == 0
+
+    @pytest.mark.asyncio
+    async def test_transactions_raises_when_closed(self, tracking_transactions, race_cluster):
+        await race_cluster.close()
 
-        async def _noop(self):
-            self._closed = True
+        with pytest.raises(RuntimeError):
+            race_cluster.transactions
 
-        monkeypatch.setattr(AsyncClientAdapter, 'close_connection', _noop)
+        assert tracking_transactions.instances_created == 0
 
     @pytest.mark.asyncio
-    @pytest.mark.usefixtures('stub_adapter_close')
-    async def test_close_during_first_access_closes_instance(self, tracking_transactions, unconnected_cluster):
+    async def test_close_during_first_access_closes_instance(self, tracking_transactions, race_cluster):
         result = {}
 
         def touch():
             # Off-loop: how the property stays reachable while close() runs on the loop.
-            result['txns'] = unconnected_cluster.transactions
+            result['txns'] = race_cluster.transactions
 
         toucher = threading.Thread(target=touch)
         toucher.start()
         # Proceed only once the property is inside Transactions(), so close() races it.
         assert tracking_transactions.construction_started.wait(timeout=10), 'construction never started'
-        await unconnected_cluster.close()
+        await race_cluster.close()
         toucher.join(timeout=10)
         assert not toucher.is_alive(), 'thread failed to join, possible deadlock'
 
         txns = result.get('txns')
         assert txns is not None
         assert tracking_transactions.instances_created == 1
         assert txns.close_count == 1, 'the instance published by the racing first access was never closed'
-        assert unconnected_cluster._impl._transactions is None, 'a live Transactions outlived cluster.close()'
+        assert race_cluster._impl._transactions is None, 'a live Transactions outlived cluster.close()'
 
-    def test_concurrent_first_access_returns_single_instance(self, tracking_transactions, unconnected_cluster):
+    def test_concurrent_first_access_returns_single_instance(self, tracking_transactions, race_cluster):
         barrier = threading.Barrier(NUM_THREADS)
         results = [None] * NUM_THREADS
 
         def touch(idx):
             barrier.wait()
-            results[idx] = unconnected_cluster.transactions
+            results[idx] = race_cluster.transactions
 
         threads = [threading.Thread(target=touch, args=(i,)) for i in range(NUM_THREADS)]
         for t in threads:

couchbase/logic/client_adapter.py

@@ -78,8 +78,8 @@ def _ensure_connected(self) -> None:
 
     def close_bucket(self, bucket_name: str) -> None:
         """**INTERNAL**"""
-        self._ensure_not_closed()
-        self._ensure_connected()
+        if self._closed or not self.connected:
+            return  # nothing was opened, so there is nothing to tear down
         self.execute_bucket_request(CloseBucketRequest(bucket_name))
 
     def close_connection(self) -> None:

couchbase/logic/cluster_impl.py

@@ -162,6 +162,8 @@ def observability_instruments(self) -> ObservabilityInstruments:
     @property
     def transactions(self) -> Transactions:
         """**INTERNAL**"""
+        self._client_adapter._ensure_not_closed()
+        self._client_adapter._ensure_connected()
         if self._transactions is None:
             with self._transactions_lock:
                 if self._transactions is None: