fix(data-imports): keep transient S3 credential blips out of error tracking

products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py

@@ -10,6 +10,7 @@
 import deltalake as deltalake
 import pyarrow.compute as pc
 import posthoganalytics
+import botocore.exceptions
 import deltalake.exceptions
 from dlt.common.libs.deltalake import ensure_delta_compatible_arrow_schema
 from structlog.types import FilteringBoundLogger
@@ -68,13 +69,20 @@
 
 
 def is_transient_object_store_error(error: BaseException) -> bool:
-    """True for a transient object-store error, however delta-rs happened to surface it.
+    """True for a transient object-store error, however it happened to surface.
 
     `DeltaTable.is_deltatable()` raises these as a plain `OSError`, but table-level operations
     (e.g. `vacuum()`, `optimize.compact()`) wrap the identical underlying object-store error text in
     `deltalake.exceptions.DeltaError` instead — same blip, different exception type depending on
-    which delta-rs entry point hit it.
+    which delta-rs entry point hit it. `_purge_s3_prefix`'s s3fs/aiobotocore calls can also raise a
+    bare `botocore.exceptions.NoCredentialsError` unwrapped — the same IMDS/STS credential-provider
+    blip, just surfaced by aiobotocore's own credential resolution instead of delta-rs's Rust
+    `object_store` crate. `NoCredentialsError`'s message is a fixed, generic string (no needle to
+    match), but hitting our own instance-role-authenticated bucket always means the same transient
+    resolution hiccup, so it's recognized by type rather than by message.
     """
+    if isinstance(error, botocore.exceptions.NoCredentialsError):
+        return True
     return isinstance(error, OSError | deltalake.exceptions.DeltaError) and any(
         needle in str(error) for needle in TRANSIENT_OBJECT_STORE_ERRORS
     )

products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py

@@ -12,6 +12,7 @@
 import pyarrow as pa
 import deltalake
 import pyarrow.compute as pc
+import botocore.exceptions
 from parameterized import parameterized
 
 from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.arrow_utils import (
@@ -26,6 +27,7 @@
     _first_per_pk_table,
     _realign_decimal_buffers,
     is_transient_delta_maintenance_error,
+    is_transient_object_store_error,
 )
 
 
@@ -1267,3 +1269,30 @@ class TestIsTransientDeltaMaintenanceError:
     )
     def test_matches_only_the_racy_optimize_scan_signature(self, _name: str, error: Exception, expected: bool):
         assert is_transient_delta_maintenance_error(error) is expected
+
+
+class TestIsTransientObjectStoreError:
+    @parameterized.expand(
+        [
+            (
+                "credential_provider_not_enabled_os_error",
+                OSError(
+                    "Operation not supported: the credential provider was not enabled: "
+                    "no providers in chain provided credentials"
+                ),
+                True,
+            ),
+            ("unrelated_os_error", OSError("Permission denied: bucket policy forbids this operation"), False),
+            (
+                # s3fs/aiobotocore's own credential resolution (distinct from delta-rs's Rust
+                # object_store crate) can raise this bare, unwrapped — same IMDS/STS blip
+                # hitting our own instance-role-authenticated bucket, different client library.
+                "bare_no_credentials_error",
+                botocore.exceptions.NoCredentialsError(),
+                True,
+            ),
+            ("unrelated_exception_type", ValueError("some other unrelated failure"), False),
+        ]
+    )
+    def test_classifies_transient_errors(self, _name: str, error: Exception, expected: bool):
+        assert is_transient_object_store_error(error) is expected

products/warehouse_sources/backend/temporal/data_imports/workflow_activities/import_data_sync.py

@@ -36,6 +36,9 @@
 from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.arrow_utils import (
     SchemaColumnTypeChangedException,
 )
+from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.delta_table_helper import (
+    is_transient_object_store_error,
+)
 from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.typings import PipelineResult
 from products.warehouse_sources.backend.temporal.data_imports.pipelines.pipeline_sync import PipelineInputs
 from products.warehouse_sources.backend.temporal.data_imports.pipelines.pipeline_v2.pipeline import PipelineNonDLT
@@ -332,7 +335,8 @@ async def _handle_import_error(
     (``posthog/temporal/common/posthog_client.py``) from reporting whatever exception type escapes
     the activity; only that marker type does. ``RESTClientRetryableError`` gets the same treatment
     by type, since it's already a ``NonReportableError`` subclass and every REST-based source hits
-    that condition already.
+    that condition already. A transient object-store hiccup talking to our own data-warehouse
+    bucket is re-raised as ``NonReportableError`` the same way.
 
     Everything else is logged as an exception and re-raised so Temporal retries it as usual.
     """
@@ -378,6 +382,15 @@ async def _handle_import_error(
         await logger.adebug("REST client exhausted its retries - re-raising for Temporal retry")
         raise error
 
+    # A transient S3/object-store hiccup talking to our own data-warehouse bucket (IMDS/STS
+    # blip, SlowDown throttling) that surfaced during this run — e.g. resetting or opening the
+    # Delta table. Not a PostHog defect and not a customer credential problem (see
+    # TRANSIENT_OBJECT_STORE_ERRORS), and retrying resolves it, so it shouldn't page anyone.
+    if is_transient_object_store_error(error):
+        await logger.awarning(error_msg)
+        await logger.adebug("Transient object-store error - re-raising for Temporal retry")
+        raise NonReportableError(error_msg) from error
+
     # Cross-source non-retryable errors (missing primary key on an incremental table, bad SSH tunnel
     # auth, a widened column type) are raised from shared pipeline code, not any one source. The
     # finalization activity already consults this shared dict; this in-activity handler decides whether

products/warehouse_sources/backend/temporal/data_imports/workflow_activities/tests/test_import_data_sync.py

@@ -222,6 +222,33 @@ async def test_rest_client_retryable_error_logged_as_warning_without_source_opt_
     logger.aexception.assert_not_awaited()
 
 
+@pytest.mark.asyncio
+async def test_transient_object_store_error_reraised_as_non_reportable():
+    # A transient S3 credential-provider blip (IMDS/STS) talking to our own data-warehouse bucket,
+    # e.g. while resetting the Delta table. It's retryable (Temporal retries the activity as usual),
+    # but re-raising the bare OSError would still be captured by the activity interceptor, which
+    # only skips reporting for NonReportableError — so it must be wrapped, not just logged at warning.
+    error = OSError(
+        "Operation not supported: the credential provider was not enabled: no providers in chain provided credentials"
+    )
+    source = mock.MagicMock(spec=SimpleSource)
+    source.get_non_retryable_errors.return_value = {}
+    source.get_retryable_errors.return_value = set()
+
+    logger = mock.MagicMock()
+    logger.awarning = mock.AsyncMock()
+    logger.aexception = mock.AsyncMock()
+    logger.adebug = mock.AsyncMock()
+
+    with mock.patch.object(module.SourceRegistry, "get_source", return_value=source):
+        with pytest.raises(NonReportableError) as exc_info:
+            await module._handle_import_error(mock.MagicMock(), logger, error)
+
+    assert exc_info.value.__cause__ is error
+    logger.awarning.assert_awaited_once()
+    logger.aexception.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_schema_column_type_changed_routes_through_handler_without_source_opt_in():
     # SchemaColumnTypeChangedException is raised in shared pipeline code when incoming data can't be