diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py index 6027d575b141..66a1412ba746 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py +++ b/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 ) diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py index a1f58e4af4bd..1104500088a4 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py +++ b/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 diff --git a/products/warehouse_sources/backend/temporal/data_imports/workflow_activities/import_data_sync.py b/products/warehouse_sources/backend/temporal/data_imports/workflow_activities/import_data_sync.py index ba2c29905923..7867d8bcd7ac 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/workflow_activities/import_data_sync.py +++ b/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 diff --git a/products/warehouse_sources/backend/temporal/data_imports/workflow_activities/tests/test_import_data_sync.py b/products/warehouse_sources/backend/temporal/data_imports/workflow_activities/tests/test_import_data_sync.py index a04c2a182f4e..0da8c1ecfbd3 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/workflow_activities/tests/test_import_data_sync.py +++ b/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