fix(data-imports): keep source-classified retryable errors out of error tracking

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

@@ -14,6 +14,7 @@
 from posthog.models.integration import UndecryptedIntegrationSecretError
 from posthog.sync import database_sync_to_async_pool
 from posthog.temporal.common.activity_context import current_activity_attempt
+from posthog.temporal.common.errors import NonReportableError
 from posthog.temporal.common.heartbeat import LivenessHeartbeater as Heartbeater
 from posthog.temporal.common.logger import get_logger
 from posthog.temporal.common.shutdown import ShutdownMonitor
@@ -326,9 +327,12 @@ async def _handle_import_error(
 
     Errors the source classifies as retryable (rate limits, transient 5xx) reach us only after
     the source's own retries are exhausted. Temporal retries the whole activity and the error is
-    transient and self-recovering, so we log at ``warning`` rather than ``exception`` to keep
-    this benign, recoverable failure out of error tracking. ``RESTClientRetryableError`` gets the
-    same treatment by type, since every REST-based source hits that condition already.
+    transient and self-recovering, so we log at ``warning`` rather than ``exception`` and re-raise
+    as ``NonReportableError`` — log level alone doesn't stop the activity interceptor
+    (``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.
 
     Everything else is logged as an exception and re-raised so Temporal retries it as usual.
     """
@@ -384,7 +388,7 @@ async def _handle_import_error(
     if any(match in error_msg for match in retryable_errors):
         await logger.awarning(error_msg)
         await logger.adebug("Source-classified retryable error - re-raising for Temporal retry")
-        raise error
+        raise NonReportableError(error_msg) from error
 
     await logger.aexception(error_msg)
     await logger.adebug("Error encountered during import_data_activity - re-raising")

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

@@ -5,6 +5,8 @@
 import pytest
 from unittest import mock
 
+from posthog.temporal.common.errors import NonReportableError
+
 from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.arrow_utils import (
     SchemaColumnTypeChangedException,
 )
@@ -141,7 +143,11 @@ async def test_unparseable_config_routes_through_handler():
 async def test_source_classified_retryable_error_logged_as_warning_not_exception():
     # A rate-limit / transient error the source retries internally reaches _handle_import_error only
     # once those retries exhaust. Temporal retries the whole activity, so it must be logged at
-    # warning (not aexception, which mints error-tracking noise) while still being re-raised.
+    # warning (not aexception, which mints error-tracking noise) while still being re-raised. Logging
+    # alone doesn't keep it out of error tracking though: the Temporal activity interceptor
+    # (posthog_client.py) captures whatever exception type escapes the activity regardless of log
+    # level, unless it's a NonReportableError — so the re-raise must wrap it as one, the same way
+    # RESTClientRetryableError already does for REST sources.
     error = Exception("Mixpanel API error (retryable): status=429, url=https://data.mixpanel.com/api/2.0/export")
     source = mock.MagicMock(spec=SimpleSource)
     source.get_non_retryable_errors.return_value = {}
@@ -153,9 +159,10 @@ async def test_source_classified_retryable_error_logged_as_warning_not_exception
     logger.adebug = mock.AsyncMock()
 
     with mock.patch.object(module.SourceRegistry, "get_source", return_value=source):
-        with pytest.raises(Exception, match="retryable"):
+        with pytest.raises(NonReportableError, match="retryable") 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()