Retry transactions whose cached plan went stale on the Rust backend

The Rust backend prepares and caches statements per connection. DDL that
changes a cached statement's result type, such as a schema delta applied
during a rolling restart, makes Postgres reject the stale plan with
SQLSTATE 0A000. Inside a transaction that error used to propagate as a
plain DatabaseError, which `new_transaction` does not retry, so other
workers saw transient failures until their caches turned over.

The cursor already drops the cached entry before the error propagates,
so treating 0A000 like a deadlock means the retried transaction prepares
a fresh plan.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ULdieJyQoHNzeHq42H2M1p

synapse/storage/engines/postgres_rust.py

@@ -50,8 +50,16 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
-# Deadlock / serialization-failure SQLSTATEs that Synapse retries.
-_RETRYABLE_PGCODES = ("40001", "40P01")
+# SQLSTATEs for which `new_transaction` retries the whole transaction.
+#
+# `40001` (serialization failure) and `40P01` (deadlock detected) are what the
+# psycopg2 engine retries. `0A000` is specific to this backend: it prepares and
+# caches statements per connection, and DDL that changes a cached statement's
+# result type (a rolling restart applying a schema delta, say) makes Postgres
+# reject the stale plan with "cached plan must not change result type". The
+# cursor drops the cached entry before the error propagates, so the retried
+# transaction prepares a fresh plan.
+_RETRYABLE_PGCODES = ("40001", "40P01", "0A000")
 
 
 class RustPostgresEngine(PostgresEngine[Connection, Cursor]):
@@ -115,6 +123,8 @@ class RustPostgresEngine(PostgresEngine[Connection, Cursor]):
         pass
 
     def is_deadlock(self, error: Exception) -> bool:
+        # Despite the name, this is the transaction driver's "retry the whole
+        # transaction" predicate; see `_RETRYABLE_PGCODES`.
         if isinstance(error, postgres.DatabaseError):
             return getattr(error, "pgcode", None) in _RETRYABLE_PGCODES
         return False

tests/storage/test_rust_engine.py

@@ -70,7 +70,9 @@ class RustPostgresEngineTestCase(unittest.TestCase):
         )
 
     def test_is_deadlock_matches_rust_database_error_pgcodes(self) -> None:
-        for pgcode in ("40001", "40P01"):
+        # Deadlock, serialization failure, and a stale cached plan after DDL:
+        # all retried as a whole transaction.
+        for pgcode in ("40001", "40P01", "0A000"):
             err = postgres.DatabaseError("boom")
             err.pgcode = pgcode
             self.assertTrue(self.engine.is_deadlock(err))