diff --git a/products/warehouse_sources/backend/temporal/data_imports/sources/postgres/source.py b/products/warehouse_sources/backend/temporal/data_imports/sources/postgres/source.py index 2e823eebf5df..0572351b6423 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/sources/postgres/source.py +++ b/products/warehouse_sources/backend/temporal/data_imports/sources/postgres/source.py @@ -519,7 +519,20 @@ def get_non_retryable_errors(self) -> dict[str, str | None]: "dashboard for this branch's connection settings, then re-enable the sync." ), "FATAL: no such database": None, - "does not exist": None, + # A relation or column the sync reads was dropped or renamed on the source, so the + # streaming query fails with SQLSTATE 42P01 ("relation ... does not exist") or 42703 + # ("column ... does not exist"). The stored schema/query is fixed until the customer + # changes it, so every retry replays the same statement. Already non-retryable through + # this bucket; the actionable message replaces the raw psycopg text, which echoes the + # relation name and a SQL fragment back into `latest_error`. This key is a broad + # substring match (case-insensitive `does not exist` anywhere in the driver text), so it + # can also catch other dropped Postgres objects (e.g. a type or role); the message is + # worded to not overclaim it's always a table or column. + "does not exist": ( + "Something this sync depends on (a table, column, or other object) no longer exists in " + "your source database. Remove it from the source's selected tables, or reset and re-sync " + "this table, then re-enable the sync." + ), "timestamp too small": None, "QueryTimeoutException": None, # Activity-layer twin of the `QueryTimeoutException` key above. That key only matches once diff --git a/products/warehouse_sources/backend/temporal/data_imports/sources/postgres/test_postgres.py b/products/warehouse_sources/backend/temporal/data_imports/sources/postgres/test_postgres.py index e21f55681cf8..94e4e3b078f8 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/sources/postgres/test_postgres.py +++ b/products/warehouse_sources/backend/temporal/data_imports/sources/postgres/test_postgres.py @@ -453,6 +453,24 @@ def test_unrecognized_session_parameter_surfaces_actionable_message(self, source assert matches[0] is not None, "unrecognized session parameter must surface an actionable message" assert "session setting" in matches[0].lower() + def test_missing_relation_surfaces_actionable_message(self, source): + # A dropped/renamed table or column stays non-retryable, but must surface an actionable + # message rather than the raw psycopg text (which echoes the relation name and SQL fragment). + # Mirror the finalizer's first-match selection so a reorder that shadows it with an earlier + # None-valued key, or a revert of this bucket back to None, is caught. The relation name is + # invented, not a real customer value. + error_msg = ( + 'relation "public.orders" does not exist LINE 1: DECLARE _cur CURSOR FOR SELECT * FROM "public"."orders"' + ) + matches = [ + friendly + for pattern, friendly in source.get_non_retryable_errors().items() + if error_message_matches(error_msg, [pattern]) + ] + assert matches, "a dropped relation must be classified non-retryable" + assert matches[0] is not None, "a dropped relation must surface an actionable message, not raw driver text" + assert "no longer exists" in matches[0].lower() + def test_connect_timeout_surfaces_actionable_message(self, source): # A persistently timing-out connect stays non-retryable, but must surface firewall/reachability # guidance rather than the bare "connection timeout expired" driver text. Mirror the finalizer's diff --git a/products/warehouse_sources/backend/temporal/data_imports/sources/supabase/test_supabase_source.py b/products/warehouse_sources/backend/temporal/data_imports/sources/supabase/test_supabase_source.py index 0c763149e1ce..8cac2e8554a8 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/sources/supabase/test_supabase_source.py +++ b/products/warehouse_sources/backend/temporal/data_imports/sources/supabase/test_supabase_source.py @@ -231,21 +231,24 @@ def _resolve_friendly_error(source: SupabaseSource, raw_error: str) -> str | Non @pytest.mark.parametrize( - "raw_error,expect_message", + "raw_error,expect_realtime_message", [ - # Retention dropped the dated realtime.messages partition — actionable message, not the - # inherited generic "does not exist" (which resolves to None / the raw driver string). + # Retention dropped the dated realtime.messages partition, so its specific realtime message + # must win over the inherited generic "does not exist" bucket (first matching key wins). ('relation "realtime.messages_2020_01_01" does not exist', True), - # A regular missing table must still fall through to the generic (None) mapping, so the + # A regular missing table must fall through to the generic missing-relation message, so the # realtime key stays specific and doesn't swallow every "does not exist". ('relation "public.orders" does not exist', False), ], ) -def test_expired_realtime_partition_gets_actionable_message(raw_error, expect_message): +def test_expired_realtime_partition_gets_actionable_message(raw_error, expect_realtime_message): friendly = _resolve_friendly_error(SupabaseSource(), raw_error) - if expect_message: - assert friendly is not None + # Both cases are non-retryable with an actionable message now; only the realtime partition gets + # the realtime-specific copy. + assert friendly is not None + if expect_realtime_message: assert "realtime.messages" in friendly else: - assert friendly is None + assert "realtime.messages" not in friendly + assert "no longer exists" in friendly.lower()