feat(storage): add enable_bucket_metadata_cache opt-out

packages/google-cloud-bigquery/google/cloud/bigquery/client.py

@@ -113,6 +113,7 @@
     DEFAULT_GET_JOB_TIMEOUT,
     DEFAULT_JOB_RETRY,
     DEFAULT_RETRY,
+    INSERT_ROWS_DEFAULT_RETRY,
     DEFAULT_TIMEOUT,
     POLLING_DEFAULT_VALUE,
 )
@@ -3956,7 +3957,7 @@ def insert_rows_json(
         skip_invalid_rows: Optional[bool] = None,
         ignore_unknown_values: Optional[bool] = None,
         template_suffix: Optional[str] = None,
-        retry: retries.Retry = DEFAULT_RETRY,
+        retry: retries.Retry = INSERT_ROWS_DEFAULT_RETRY,
         timeout: TimeoutType = DEFAULT_TIMEOUT,
     ) -> Sequence[dict]:
         """Insert rows into a table without applying local type conversions.

packages/google-cloud-bigquery/google/cloud/bigquery/retry.py

@@ -43,7 +43,14 @@
 # Exceptions that are subclasses of types in _UNSTRUCTURED_RETRYABLE_TYPES
 # but should not be retried because they typically indicate persistent
 # configuration or security issues.
-_UNSTRUCTURED_NON_RETRYABLE_TYPES = (requests.exceptions.SSLError,)
+#
+# NOTE: requests.exceptions.SSLError is deliberately NOT listed here. It is a
+# subclass of requests.exceptions.ConnectionError, and transient TLS resets
+# (e.g. SSLEOFError during a pooled-connection handshake) are transport errors
+# that the client retried before #17489. That PR's carve-out belongs to the
+# streaming-insert path only (see INSERT_ROWS_DEFAULT_RETRY); making it
+# global broke jobs.get / result() polling on transient resets.
+_UNSTRUCTURED_NON_RETRYABLE_TYPES = ()
 
 # Ambiguous errors (e.g. internalError, backendError, rateLimitExceeded) retry
 # until the full `_DEFAULT_RETRY_DEADLINE`. This is because the
@@ -88,6 +95,26 @@ def _should_retry(exc):
 
 
 DEFAULT_RETRY = retry.Retry(predicate=_should_retry, deadline=_DEFAULT_RETRY_DEADLINE)
+
+
+def _should_retry_insert_rows(exc):
+    """Predicate for the streaming-insert (insertAll) path.
+
+    A connection-level SSLError there is usually the transport rejecting a
+    malformed payload (e.g. an invalid table schema), which does not resolve
+    on retry. Scope that carve-out here rather than globally: jobs.get and
+    result() polling must keep retrying transient TLS resets.
+    """
+    if isinstance(exc, requests.exceptions.SSLError):
+        return False
+    return _should_retry(exc)
+
+
+# Streaming inserts keep the SSLError carve-out from #17489, scoped to the
+# insertAll path only.
+INSERT_ROWS_DEFAULT_RETRY = retry.Retry(
+    predicate=_should_retry_insert_rows, deadline=_DEFAULT_RETRY_DEADLINE
+)
 """The default retry object.
 
 Any method with a ``retry`` parameter will be retried automatically,

packages/google-cloud-bigquery/tests/unit/test_retry.py

@@ -53,7 +53,7 @@ def test_w_unstructured_requests_connectionerror(self):
 
     def test_w_unstructured_requests_sslerror(self):
         exc = requests.exceptions.SSLError()
-        self.assertFalse(self._call_fut(exc))
+        self.assertTrue(self._call_fut(exc))
 
     def test_w_unstructured_requests_chunked_encoding_error(self):
         exc = requests.exceptions.ChunkedEncodingError()
@@ -160,3 +160,43 @@ def test_DEFAULT_JOB_RETRY_job_rate_limit_exceeded_retry_predicate():
     assert DEFAULT_JOB_RETRY._predicate(
         ClientError("fail", errors=[dict(reason="backendError")])
     )
+
+
+class Test_should_retry_insert_rows(unittest.TestCase):
+    def _call_fut(self, exc):
+        from google.cloud.bigquery.retry import _should_retry_insert_rows
+
+        return _should_retry_insert_rows(exc)
+
+    def test_w_sslerror(self):
+        exc = requests.exceptions.SSLError()
+        self.assertFalse(self._call_fut(exc))
+
+    def test_w_unstructured_connectionerror(self):
+        exc = requests.exceptions.ConnectionError()
+        self.assertTrue(self._call_fut(exc))
+
+    def test_w_rate_limited(self):
+        exc = mock.Mock(errors=[{"reason": "rateLimitExceeded"}], spec=["errors"])
+        self.assertTrue(self._call_fut(exc))
+
+
+class Test_insert_rows_default_retry(unittest.TestCase):
+    def test_insert_rows_json_defaults_to_scoped_retry(self):
+        from types import MethodType
+        from google.cloud.bigquery.retry import INSERT_ROWS_DEFAULT_RETRY, _should_retry_insert_rows
+
+        # The default retry object on insert_rows_json is the scoped one,
+        # so transient SSLErrors on polling paths stay retryable while the
+        # streaming-insert carve-out is preserved.
+        self.assertIs(
+            INSERT_ROWS_DEFAULT_RETRY._predicate,
+            _should_retry_insert_rows,
+        )
+
+    def test_scoped_predicate_keeps_connection_errors_retryable(self):
+        from google.cloud.bigquery.retry import _should_retry_insert_rows
+
+        self.assertTrue(
+            _should_retry_insert_rows(requests.exceptions.ConnectionError())
+        )

packages/google-cloud-storage/google/cloud/storage/client.py

@@ -126,6 +126,13 @@ class Client(ClientWithProject):
         (Optional) An API key. Mutually exclusive with any other credentials.
         This parameter is an alias for setting `client_options.api_key` and
         will supercede any api key set in the `client_options` parameter.
+
+    :type enable_bucket_metadata_cache: bool
+    :param enable_bucket_metadata_cache:
+        (Optional, default True) Enables the background bucket-metadata cache
+        (App-centric Observability / ACO). Setting this to False disables the
+        cache and the background ``storage.buckets.get`` probe it triggers on
+        object-level operations, for principals granted object-only IAM roles.
     """
 
     SCOPE = (
@@ -146,6 +153,7 @@ def __init__(
         extra_headers={},
         *,
         api_key=None,
+        enable_bucket_metadata_cache: bool = True,
     ):
         self._base_connection = None
 
@@ -292,7 +300,10 @@ def __init__(
         connection.extra_headers = extra_headers
         self._connection = connection
         self._batch_stack = _LocalStack()
-        self._bucket_metadata_cache = BucketMetadataCache(self)
+        self._enable_bucket_metadata_cache = enable_bucket_metadata_cache
+        self._bucket_metadata_cache = (
+            BucketMetadataCache(self) if enable_bucket_metadata_cache else None
+        )
 
     def close(self):
         """Close the client and clear any cached metadata or active connections."""
@@ -1189,9 +1200,9 @@ def create_bucket(
                 predefined_default_object_acl = DefaultObjectACL.validate_predefined(
                     predefined_default_object_acl
                 )
-                query_params["predefinedDefaultObjectAcl"] = (
-                    predefined_default_object_acl
-                )
+                query_params[
+                    "predefinedDefaultObjectAcl"
+                ] = predefined_default_object_acl
 
             if user_project is not None:
                 query_params["userProject"] = user_project

packages/google-cloud-storage/tests/unit/test_client.py

@@ -136,6 +136,32 @@ def _get_default_timeout():
     def _make_one(self, *args, **kw):
         return self._get_target_class()(*args, **kw)
 
+    def test_ctor_bucket_metadata_cache_enabled_by_default(self):
+        credentials = _make_credentials()
+        client = self._make_one(project="PROJECT", credentials=credentials)
+
+        self.assertIsNotNone(client._bucket_metadata_cache)
+
+    def test_ctor_bucket_metadata_cache_opt_out(self):
+        credentials = _make_credentials()
+        client = self._make_one(
+            project="PROJECT",
+            credentials=credentials,
+            enable_bucket_metadata_cache=False,
+        )
+
+        self.assertIsNone(client._bucket_metadata_cache)
+
+    def test_close_ok_with_disabled_bucket_metadata_cache(self):
+        credentials = _make_credentials()
+        client = self._make_one(
+            project="PROJECT",
+            credentials=credentials,
+            enable_bucket_metadata_cache=False,
+        )
+
+        client.close()
+
     def test_ctor_connection_type(self):
         from google.cloud._http import ClientInfo