fix: align stream retries and closure with canonical API behavior

Author: abbrowne126

google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py

@@ -62,14 +62,23 @@
 _REGULAR_SHUTDOWN_THREAD_NAME = "Thread-RegularStreamShutdown"
 _RPC_ERROR_THREAD_NAME = "Thread-OnRpcTerminated"
 _RETRYABLE_STREAM_ERRORS = (
+    exceptions.Aborted,
+    exceptions.Cancelled,
     exceptions.DeadlineExceeded,
-    exceptions.ServiceUnavailable,
+    exceptions.GatewayTimeout,
     exceptions.InternalServerError,
+    exceptions.ResourceExhausted,
+    exceptions.ServiceUnavailable,
     exceptions.Unknown,
-    exceptions.GatewayTimeout,
-    exceptions.Aborted,
 )
-_TERMINATING_STREAM_ERRORS = (exceptions.Cancelled,)
+_TERMINATING_STREAM_ERRORS = (
+    exceptions.InvalidArgument,
+    exceptions.NotFound,
+    exceptions.PermissionDenied,
+    exceptions.PermissionDenied,
+    exceptions.Unauthenticated,
+    exceptions.Unauthorized,
+)
 _MAX_LOAD = 1.0
 """The load threshold above which to pause the incoming message stream."""
 
@@ -1283,8 +1292,10 @@ def _should_terminate(self, exception: BaseException) -> bool:
             in a list of terminating exceptions.
         """
         exception = _wrap_as_exception(exception)
-        if isinstance(exception, _TERMINATING_STREAM_ERRORS):
-            _LOGGER.debug("Observed terminating stream error %s", exception)
+        is_api_error = isinstance(exception, exceptions.GoogleAPICallError)
+        # Terminate any non-API errors, or non-retryable errors (permission denied, unauthorized, etc.)
+        if not is_api_error or isinstance(exception, _TERMINATING_STREAM_ERRORS):
+            _LOGGER.error("Observed terminating stream error %s", exception)
             return True
         _LOGGER.debug("Observed non-terminating stream error %s", exception)
         return False

tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py

@@ -2270,18 +2270,15 @@ def test__should_recover_false():
 def test__should_terminate_true():
     manager = make_manager()
 
-    details = "Cancelled. Go away, before I taunt you a second time."
-    exc = exceptions.Cancelled(details)
-
-    assert manager._should_terminate(exc) is True
+    for exc in [exceptions.PermissionDenied(""), TypeError(), ValueError()]:
+        assert manager._should_terminate(exc)
 
 
 def test__should_terminate_false():
     manager = make_manager()
 
-    exc = TypeError("wahhhhhh")
-
-    assert manager._should_terminate(exc) is False
+    for exc in [exceptions.ResourceExhausted(""), exceptions.ServiceUnavailable(""), exceptions.Cancelled("")]:
+        assert not manager._should_terminate(exc)
 
 
 @mock.patch("threading.Thread", autospec=True)