fix(spanner): stop deleting multiplexed sessions on rotation and close

Multiplexed sessions in Cloud Spanner cannot and should not be deleted
by client applications. Calling DeleteSession on a multiplexed session
causes Cloud Spanner to return an INVALID_ARGUMENT error. In addition,
calling delete() upon session rotation risks aborting in-flight queries
that are still actively using the previous session reference.

- Remove `old_session.delete()` from `_rotate_multiplexed_session()` in
  both `_async/database_sessions_manager.py` and
  `database_sessions_manager.py`.
- Remove `session_to_delete.delete()` from `close()` in both managers,
  simply clearing the session reference (`self._multiplexed_session = None`)
  and relying on Spanner's server-side session eviction once idle.
- Update docstrings to reflect that multiplexed sessions are rotated
  rather than deleted and recreated.

Author: olavloite

packages/google-cloud-spanner/google/cloud/spanner_v1/_async/database_sessions_manager.py

@@ -123,8 +123,8 @@ async def _get_multiplexed_session(self) -> Session:
         """Returns a multiplexed session from the database session manager.
 
         If the multiplexed session is not defined, creates a new multiplexed
-        session and starts a maintenance thread to periodically delete and
-        recreate it so that it remains valid. Otherwise, simply returns the
+        session and starts a maintenance thread to periodically rotate
+        it so that it remains valid. Otherwise, simply returns the
         current multiplexed session.
 
         :rtype: :class:`~google.cloud.spanner_v1.session.Session`
@@ -167,8 +167,8 @@ def _build_maintenance_thread(
         self, session: Optional[Session] = None
     ) -> CrossSync.Task:
         """Builds and returns a multiplexed session maintenance thread for
-        the database session manager. This thread will periodically delete
-        and recreate the multiplexed session to ensure that it is always valid.
+        the database session manager. This thread will periodically rotate
+        the multiplexed session to ensure that it is always valid.
 
         :type session: :class:`~google.cloud.spanner_v1.session.Session`
         :param session: (Optional) The multiplexed session to maintain.
@@ -209,25 +209,18 @@ async def _rotate_multiplexed_session(self) -> bool:
             return False
 
         async with self._multiplexed_session_lock:
-            old_session = self._multiplexed_session
             self._multiplexed_session = new_session
 
-        if old_session is not None:
-            try:
-                await CrossSync.run_if_async(old_session.delete)
-            except Exception:
-                pass
-
         return True
 
     @staticmethod
     @CrossSync.convert
     async def _maintain_multiplexed_session(session_manager_ref) -> None:
         """Maintains the multiplexed session for the database session manager.
 
-        This method will delete and recreate the referenced database session manager's
+        This method will periodically rotate the referenced database session manager's
         multiplexed session to ensure that it is always valid. The method will run until
-        the database session manager is deleted or the multiplexed session is deleted.
+        the database session manager is garbage collected or the session manager is closed.
 
         :type session_manager_ref: :class:`_weakref.ReferenceType`
         :param session_manager_ref: A weak reference to the database session manager."""
@@ -292,7 +285,4 @@ async def close(self) -> None:
                     pass
             else:
                 self._multiplexed_session_thread.join()
-        if self._multiplexed_session is not None:
-            session_to_delete = self._multiplexed_session
-            self._multiplexed_session = None
-            await session_to_delete.delete()
+        self._multiplexed_session = None

packages/google-cloud-spanner/google/cloud/spanner_v1/database_sessions_manager.py

@@ -121,8 +121,8 @@ def _get_multiplexed_session(self) -> Session:
         """Returns a multiplexed session from the database session manager.
 
         If the multiplexed session is not defined, creates a new multiplexed
-        session and starts a maintenance thread to periodically delete and
-        recreate it so that it remains valid. Otherwise, simply returns the
+        session and starts a maintenance thread to periodically rotate
+        it so that it remains valid. Otherwise, simply returns the
         current multiplexed session.
 
         :rtype: :class:`~google.cloud.spanner_v1.session.Session`
@@ -162,8 +162,8 @@ def _build_maintenance_thread(
         self, session: Optional[Session] = None
     ) -> CrossSync._Sync_Impl.Task:
         """Builds and returns a multiplexed session maintenance thread for
-        the database session manager. This thread will periodically delete
-        and recreate the multiplexed session to ensure that it is always valid.
+        the database session manager. This thread will periodically rotate
+        the multiplexed session to ensure that it is always valid.
 
         :type session: :class:`~google.cloud.spanner_v1.session.Session`
         :param session: (Optional) The multiplexed session to maintain.
@@ -196,24 +196,17 @@ def _rotate_multiplexed_session(self) -> bool:
             return False
 
         with self._multiplexed_session_lock:
-            old_session = self._multiplexed_session
             self._multiplexed_session = new_session
 
-        if old_session is not None:
-            try:
-                CrossSync._Sync_Impl.run_if_async(old_session.delete)
-            except Exception:
-                pass
-
         return True
 
     @staticmethod
     def _maintain_multiplexed_session(session_manager_ref) -> None:
         """Maintains the multiplexed session for the database session manager.
 
-        This method will delete and recreate the referenced database session manager's
+        This method will periodically rotate the referenced database session manager's
         multiplexed session to ensure that it is always valid. The method will run until
-        the database session manager is deleted or the multiplexed session is deleted.
+        the database session manager is garbage collected or the session manager is closed.
 
         :type session_manager_ref: :class:`_weakref.ReferenceType`
         :param session_manager_ref: A weak reference to the database session manager."""
@@ -269,7 +262,4 @@ def close(self) -> None:
             self._multiplexed_session_terminate_event.set()
         if self._multiplexed_session_thread is not None:
             self._multiplexed_session_thread.join()
-        if self._multiplexed_session is not None:
-            session_to_delete = self._multiplexed_session
-            self._multiplexed_session = None
-            session_to_delete.delete()
+        self._multiplexed_session = None

packages/google-cloud-spanner/tests/unit/_async/test_database.py

@@ -189,7 +189,7 @@ def __await__(self):
 
         manager._multiplexed_session_terminate_event.set.assert_called_once()
         manager._multiplexed_session_thread.cancel.assert_called_once()
-        mock_session.delete.assert_called_once()
+        mock_session.delete.assert_not_called()
         self.assertIsNone(manager._multiplexed_session)
 
     @CrossSync.pytest

packages/google-cloud-spanner/tests/unit/_async/test_sessions_manager_extra.py

@@ -105,7 +105,8 @@ async def fake_coro():
 
         task = asyncio.create_task(fake_coro())
         manager._multiplexed_session_thread = task
-        manager._multiplexed_session = mock.AsyncMock()
+        mock_session = mock.AsyncMock()
+        manager._multiplexed_session = mock_session
         manager._multiplexed_session_terminate_event = mock.Mock()
 
         with mock.patch(
@@ -116,10 +117,13 @@ async def fake_coro():
             # task is cancelled and awaited in close()
             self.assertTrue(task.done())
             manager._multiplexed_session_terminate_event.set.assert_called_once()
+            self.assertIsNone(manager._multiplexed_session)
+            mock_session.delete.assert_not_called()
 
         # Sync branch of close
         manager._multiplexed_session_thread = mock.Mock()
-        manager._multiplexed_session = mock.AsyncMock()
+        mock_session_sync = mock.AsyncMock()
+        manager._multiplexed_session = mock_session_sync
         manager._multiplexed_session_terminate_event = mock.Mock()
         with mock.patch(
             "google.cloud.spanner_v1._async.database_sessions_manager.CrossSync.is_async",
@@ -128,6 +132,8 @@ async def fake_coro():
             await manager.close()
             self.assertTrue(manager._multiplexed_session_thread.join.called)
             manager._multiplexed_session_terminate_event.set.assert_called_once()
+            self.assertIsNone(manager._multiplexed_session)
+            mock_session_sync.delete.assert_not_called()
 
     async def test_maintain_multiplexed_session_refresh(self):
         # coverage for line 196-202
@@ -250,25 +256,21 @@ async def test_get_multiplexed_session_fast_path_lock_already_created(self):
         mock_lock.acquire.assert_not_called()
         mock_lock.__aenter__.assert_not_called()
 
-    async def test_maintain_multiplexed_session_swaps_before_deleting_old_session(
-        self,
-    ):
+    async def test_maintain_multiplexed_session_rotates_session(self):
         from weakref import ref
 
         manager = DatabaseSessionsManager(self.database, self.pool)
         manager._multiplexed_session_lock = asyncio.Lock()
-        manager._multiplexed_session_terminate_event = asyncio.Event()
+        manager._multiplexed_session_terminate_event = mock.Mock()
+        manager._multiplexed_session_terminate_event.is_set.side_effect = [
+            False,
+            True,
+        ]
 
         old_session = mock.AsyncMock()
         new_session = mock.AsyncMock()
         manager._multiplexed_session = old_session
 
-        async def verify_swap_on_delete():
-            self.assertIs(manager._multiplexed_session, new_session)
-            manager._multiplexed_session_terminate_event.set()
-
-        old_session.delete.side_effect = verify_swap_on_delete
-
         refresh_interval = manager._MAINTENANCE_THREAD_REFRESH_INTERVAL.total_seconds()
         call_count = 0
 
@@ -290,7 +292,8 @@ def mock_time():
                     ref(manager)
                 )
                 mock_build.assert_called_once()
-                old_session.delete.assert_called_once()
+                old_session.delete.assert_not_called()
+                new_session.delete.assert_not_called()
                 self.assertIs(manager._multiplexed_session, new_session)
 
     async def test_maintain_multiplexed_session_handles_build_failure(self):
@@ -337,48 +340,6 @@ async def mock_event_wait(event, timeout=None):
                     current_session.delete.assert_not_called()
                     self.assertIs(manager._multiplexed_session, current_session)
 
-    async def test_maintain_multiplexed_session_handles_delete_failure(self):
-        from weakref import ref
-
-        manager = DatabaseSessionsManager(self.database, self.pool)
-        manager._multiplexed_session_lock = asyncio.Lock()
-        manager._multiplexed_session_terminate_event = asyncio.Event()
-
-        old_session = mock.AsyncMock()
-        old_session.delete.side_effect = Exception("delete failed")
-        new_session = mock.AsyncMock()
-        manager._multiplexed_session = old_session
-
-        async def verify_swap_on_delete():
-            self.assertIs(manager._multiplexed_session, new_session)
-            manager._multiplexed_session_terminate_event.set()
-            raise Exception("delete failed")
-
-        old_session.delete.side_effect = verify_swap_on_delete
-
-        refresh_interval = manager._MAINTENANCE_THREAD_REFRESH_INTERVAL.total_seconds()
-        call_count = 0
-
-        def mock_time():
-            nonlocal call_count
-            call_count += 1
-            if call_count == 1:
-                return 0
-            return refresh_interval + 100
-
-        with mock.patch(
-            "google.cloud.spanner_v1._async.database_sessions_manager.time.monotonic",
-            side_effect=mock_time,
-        ):
-            with mock.patch.object(
-                manager, "_build_multiplexed_session", return_value=new_session
-            ):
-                await DatabaseSessionsManager._maintain_multiplexed_session(
-                    ref(manager)
-                )
-                old_session.delete.assert_called_once()
-                self.assertIs(manager._multiplexed_session, new_session)
-
     async def test_maintain_multiplexed_session_old_session_none(self):
         from weakref import ref
 
@@ -580,7 +541,8 @@ async def test_rotate_multiplexed_session_success(self):
             result = await manager._rotate_multiplexed_session()
             self.assertTrue(result)
             self.assertIs(manager._multiplexed_session, new_session)
-            old_session.delete.assert_called_once()
+            old_session.delete.assert_not_called()
+            new_session.delete.assert_not_called()
 
     async def test_rotate_multiplexed_session_build_failure(self):
         manager = DatabaseSessionsManager(self.database, self.pool)
@@ -597,19 +559,3 @@ async def test_rotate_multiplexed_session_build_failure(self):
             self.assertFalse(result)
             self.assertIs(manager._multiplexed_session, current_session)
             current_session.delete.assert_not_called()
-
-    async def test_rotate_multiplexed_session_delete_failure(self):
-        manager = DatabaseSessionsManager(self.database, self.pool)
-        manager._multiplexed_session_lock = asyncio.Lock()
-        old_session = mock.AsyncMock()
-        old_session.delete.side_effect = Exception("delete failed")
-        new_session = mock.AsyncMock()
-        manager._multiplexed_session = old_session
-
-        with mock.patch.object(
-            manager, "_build_multiplexed_session", return_value=new_session
-        ):
-            result = await manager._rotate_multiplexed_session()
-            self.assertTrue(result)
-            self.assertIs(manager._multiplexed_session, new_session)
-            old_session.delete.assert_called_once()

packages/google-cloud-spanner/tests/unit/test_database_session_manager.py

@@ -274,7 +274,7 @@ def test_get_multiplexed_session_fast_path_lock_already_created(self):
         mock_lock.acquire.assert_not_called()
         mock_lock.__enter__.assert_not_called()
 
-    def test_maintain_multiplexed_session_swaps_before_deleting_old_session(self):
+    def test_maintain_multiplexed_session_rotates_session(self):
         import threading
         from weakref import ref
 
@@ -287,11 +287,6 @@ def test_maintain_multiplexed_session_swaps_before_deleting_old_session(self):
         new_session = Mock()
         manager._multiplexed_session = old_session
 
-        def verify_swap_on_delete():
-            self.assertIs(manager._multiplexed_session, new_session)
-
-        old_session.delete.side_effect = verify_swap_on_delete
-
         call_count = 0
 
         def mock_time():
@@ -310,7 +305,8 @@ def mock_time():
             ) as mock_build:
                 DatabaseSessionsManager._maintain_multiplexed_session(ref(manager))
                 mock_build.assert_called_once()
-                old_session.delete.assert_called_once()
+                old_session.delete.assert_not_called()
+                new_session.delete.assert_not_called()
                 self.assertIs(manager._multiplexed_session, new_session)
 
     def test_close_branches(self):
@@ -330,7 +326,7 @@ def test_close_branches(self):
         manager._multiplexed_session_terminate_event.set.assert_called_once()
         mock_thread.join.assert_called_once()
         self.assertIsNone(manager._multiplexed_session)
-        mock_session.delete.assert_called_once()
+        mock_session.delete.assert_not_called()
 
     def test_maintain_multiplexed_session_handles_build_failure(self):
         import threading
@@ -374,40 +370,6 @@ def mock_time():
                     current_session.delete.assert_not_called()
                     self.assertIs(manager._multiplexed_session, current_session)
 
-    def test_maintain_multiplexed_session_handles_delete_failure(self):
-        import threading
-        from weakref import ref
-
-        manager = DatabaseSessionsManager(self._manager._database, self._manager._pool)
-        manager._multiplexed_session_lock = threading.Lock()
-        manager._multiplexed_session_terminate_event = Mock()
-        manager._multiplexed_session_terminate_event.is_set.side_effect = [False, True]
-
-        old_session = Mock()
-        old_session.delete.side_effect = Exception("delete failed")
-        new_session = Mock()
-        manager._multiplexed_session = old_session
-
-        call_count = 0
-
-        def mock_time():
-            nonlocal call_count
-            call_count += 1
-            if call_count == 1:
-                return 0
-            return 1000000
-
-        with patch(
-            "google.cloud.spanner_v1.database_sessions_manager.time.monotonic",
-            side_effect=mock_time,
-        ):
-            with patch.object(
-                manager, "_build_multiplexed_session", return_value=new_session
-            ):
-                DatabaseSessionsManager._maintain_multiplexed_session(ref(manager))
-                old_session.delete.assert_called_once()
-                self.assertIs(manager._multiplexed_session, new_session)
-
     def test_maintain_multiplexed_session_old_session_none(self):
         import threading
         from weakref import ref
@@ -662,7 +624,8 @@ def test_rotate_multiplexed_session_success(self):
             result = manager._rotate_multiplexed_session()
             self.assertTrue(result)
             self.assertIs(manager._multiplexed_session, new_session)
-            old_session.delete.assert_called_once()
+            old_session.delete.assert_not_called()
+            new_session.delete.assert_not_called()
 
     def test_rotate_multiplexed_session_build_failure(self):
         import threading
@@ -682,24 +645,6 @@ def test_rotate_multiplexed_session_build_failure(self):
             self.assertIs(manager._multiplexed_session, current_session)
             current_session.delete.assert_not_called()
 
-    def test_rotate_multiplexed_session_delete_failure(self):
-        import threading
-
-        manager = DatabaseSessionsManager(self._manager._database, self._manager._pool)
-        manager._multiplexed_session_lock = threading.Lock()
-        old_session = Mock()
-        old_session.delete.side_effect = Exception("delete failed")
-        new_session = Mock()
-        manager._multiplexed_session = old_session
-
-        with patch.object(
-            manager, "_build_multiplexed_session", return_value=new_session
-        ):
-            result = manager._rotate_multiplexed_session()
-            self.assertTrue(result)
-            self.assertIs(manager._multiplexed_session, new_session)
-            old_session.delete.assert_called_once()
-
     def _assert_true_with_timeout(self, condition: Callable) -> None:
         """Asserts that the given condition is met within a timeout period.