Let the test suite run against the Rust Postgres backend

Add SYNAPSE_TEST_RUST_DRIVER to run the Postgres test suite against the native
Rust driver (sets use_rust_driver on the test database config, off by default).
Teach the harness's make_fake_db_pool to build a RustConnectionPool for the Rust
engine, with its worker thread pool swapped for the threadless test pool so
queries run deterministically on the reactor during `pump` (the shim's block_on
waits there for the tokio workers). Test-database create/drop is admin work, so
it now goes directly over psycopg2 (always available in tests) rather than the
engine module's `connect`.

Running the suite this way surfaces a few real psycopg2-compatibility gaps:

  - `fixup_config_defaults` (was `fixup_default_host`) now also fills libpq's
    PGUSER / PGPASSWORD environment defaults, not just the host — tokio-postgres
    reads none of them from the environment, but Synapse's test setup relies on
    them (as psycopg2 does).
  - `build_dsn` skips `None`-valued args, matching psycopg2's handling of unset
    kwargs.

With these, a homeserver boots on the Rust backend under the test harness and
starts running real transactions and background updates. The remaining psycopg2
type-coercion / behaviour differences are addressed in the commits that follow;
psycopg2 and sqlite runs are unaffected.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d

rust/src/database/postgres/mod.rs

@@ -3,7 +3,7 @@
 //! The driver itself is async; we drive it from sync Python methods via a
 //! shared multi-thread tokio runtime (see `super::runtime`).
 
-use anyhow::Error;
+use anyhow::{Context, Error};
 use pyo3::prelude::*;
 use pyo3::types::PyModule;
 
@@ -39,27 +39,69 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()>
 }
 
 /// Fix up a DSN to ensure it has a host, using libpq's default host if
-/// necessary.
+/// necessary, and fill in libpq's environment defaults for the fields
+/// [`tokio_postgres`] doesn't read from the environment.
 ///
-/// [`tokio_postgres`] has a different default host than libpq, which is what
-/// Synapse previously used (and is what e.g. `psql` uses). libpq's default host
-/// is configurable, so when the DSN omits a host we ask libpq what its default
-/// would be and use that instead (see [`libpq::default_host`]).
-pub(crate) fn fixup_default_host(dsn: &str) -> Result<tokio_postgres::Config, Error> {
+/// [`tokio_postgres`] parses only the DSN string: unlike libpq (which is what
+/// Synapse previously used, and what e.g. `psql` uses) it does not consult
+/// `PGHOST` / `PGUSER` / `PGPASSWORD` or the compiled-in defaults. So when the
+/// DSN omits one of these we fill it the way libpq would, so a config that
+/// relied on those environment variables (as Synapse's test setup does)
+/// connects the same way it did under psycopg2.
+pub(crate) fn fixup_config_defaults(dsn: &str) -> Result<tokio_postgres::Config, Error> {
     let mut config = dsn.parse::<tokio_postgres::Config>()?;
 
-    // `tokio_postgres` parses only the DSN string (it does not consult `PGHOST`
-    // or the compiled-in default), so an empty host list means the DSN really
-    // omitted the host. A DSN that gives a `hostaddr` instead of a `host` is
-    // still connectable as-is, so leave it alone too — injecting a default host
-    // there would just confuse TLS/SNI.
-    if !config.get_hosts().is_empty() || !config.get_hostaddrs().is_empty() {
-        return Ok(config);
+    // Host. A DSN that gives a `hostaddr` instead of a `host` is still
+    // connectable as-is, so leave it alone too — injecting a default host there
+    // would just confuse TLS/SNI. Otherwise resolve libpq's default host
+    // without connecting (see `libpq::default_host`).
+    if config.get_hosts().is_empty() && config.get_hostaddrs().is_empty() {
+        config.host(&libpq::default_host()?);
     }
 
-    // Resolve libpq's default host without connecting (see `libpq::default_host`).
-    let host = libpq::default_host()?;
-    config.host(&host);
+    // Port: libpq falls back to `PGPORT` (tokio_postgres itself defaults to
+    // 5432 when none is set, matching libpq's compiled-in default). libpq
+    // also accepts a comma-separated list, one port per host.
+    if config.get_ports().is_empty() {
+        if let Some(port) = env_nonempty("PGPORT") {
+            for part in port.split(',') {
+                let parsed = part.trim().parse::<u16>().with_context(|| {
+                    format!("invalid port {part:?} in the PGPORT environment variable")
+                })?;
+                config.port(parsed);
+            }
+        }
+    }
+
+    // User: libpq falls back to `PGUSER`, then the OS user.
+    if config.get_user().is_none() {
+        if let Some(user) = env_nonempty("PGUSER").or_else(|| env_nonempty("USER")) {
+            config.user(&user);
+        }
+    }
+
+    // Password: libpq falls back to `PGPASSWORD`.
+    if config.get_password().is_none() {
+        if let Some(password) = env_nonempty("PGPASSWORD") {
+            config.password(&password);
+        }
+    }
+
+    // Database: libpq falls back to `PGDATABASE` (then the user name, which
+    // the *server* also does when the startup packet names no database, so
+    // that final fallback needs no help here).
+    if config.get_dbname().is_none() {
+        if let Some(dbname) = env_nonempty("PGDATABASE") {
+            config.dbname(&dbname);
+        }
+    }
 
     Ok(config)
 }
+
+/// Read an environment variable, treating a set-but-empty value as unset —
+/// libpq's behaviour, and what a blanked-out `PGPORT=` line in a compose file
+/// or systemd unit means.
+fn env_nonempty(var: &str) -> Option<String> {
+    std::env::var(var).ok().filter(|value| !value.is_empty())
+}

rust/src/database/postgres/pool.rs

@@ -4,7 +4,7 @@
 //! We use the generic `deadpool::managed` pool with our own [`ConnectionManager`]
 //! rather than the `deadpool-postgres` crate, so that creating a connection
 //! reuses our own DSN handling (libpq's default host, see
-//! [`super::fixup_default_host`]) and drives the connection task on the shared
+//! [`super::fixup_config_defaults`]) and drives the connection task on the shared
 //! runtime.
 //!
 //! The pooled item is a plain [`tokio_postgres::Client`]. Rust-native code takes
@@ -21,7 +21,7 @@ use tokio_postgres::{Client, Config, NoTls};
 
 use crate::database::postgres::connection::Connection;
 use crate::database::postgres::errors::{pg_err_to_py, OperationalError};
-use crate::database::postgres::fixup_default_host;
+use crate::database::postgres::fixup_config_defaults;
 use crate::database::postgres::helpers::BlockingPostgres;
 use crate::database::runtime::runtime;
 
@@ -85,7 +85,7 @@ impl ConnectionManager {
     /// Build a manager from a libpq-style DSN and session settings.
     pub fn from_dsn(dsn: &str, session: SessionConfig) -> Result<Self, anyhow::Error> {
         Ok(Self {
-            config: fixup_default_host(dsn)?,
+            config: fixup_config_defaults(dsn)?,
             session,
         })
     }

tests/server.py

@@ -113,6 +113,7 @@ from tests.utils import (
     POSTGRES_USER,
     SQLITE_PERSIST_DB,
     USE_POSTGRES_FOR_TESTS,
+    USE_RUST_DRIVER_FOR_TESTS,
     default_config,
 )
 
@@ -837,6 +838,11 @@ def make_fake_db_pool(
     is a drop-in replacement for the normal `make_pool` which builds such a connection
     pool.
     """
+    from synapse.storage.engines.postgres_rust import RustPostgresEngine
+
+    if isinstance(engine, RustPostgresEngine):
+        return _make_fake_rust_db_pool(reactor, clock, db_config, engine, server_name)
+
     pool = make_pool(
         reactor=reactor,
         clock=clock,
@@ -878,6 +884,35 @@ def make_fake_db_pool(
     return pool
 
 
+def _make_fake_rust_db_pool(
+    reactor: ISynapseReactor,
+    clock: "Clock",
+    db_config: DatabaseConnectionConfig,
+    engine: BaseDatabaseEngine,
+    server_name: str,
+) -> Any:
+    """`make_fake_db_pool` for the native Rust backend.
+
+    Builds a real `RustConnectionPool`, then swaps its worker thread pool for the
+    threadless test `ThreadPool` so `runWithConnection` runs on the reactor's
+    main thread during `pump` (the shim's `block_on` waits there for the tokio
+    workers that actually drive the connection). This keeps db queries
+    deterministic, matching the adbapi fake.
+    """
+    pool = make_pool(
+        reactor=reactor,
+        clock=clock,
+        db_config=db_config,
+        engine=engine,
+        server_name=server_name,
+    )
+    # `make_pool` started a real worker thread pool; stop it and swap in the
+    # threadless one so nothing actually runs on background threads.
+    pool.threadpool.stop()
+    pool.threadpool = ThreadPool(reactor)
+    return pool
+
+
 class ThreadPool:
     """
     Threadless thread pool.
@@ -1232,6 +1267,8 @@ def setup_test_homeserver(
                 "cp_max": 5,
             },
         }
+        if USE_RUST_DRIVER_FOR_TESTS:
+            database_config["use_rust_driver"] = True
     else:
         if SQLITE_PERSIST_DB:
             # The current working directory is in _trial_temp, so this gets created within that directory.
@@ -1277,19 +1314,23 @@ def setup_test_homeserver(
     database = DatabaseConnectionConfig("master", database_config)
     config.database.databases = [database]
 
-    db_engine = create_engine(database.config)
-
     # Create the database before we actually try and connect to it, based off
     # the template database we generate in setupdb()
     if USE_POSTGRES_FOR_TESTS:
-        db_conn = db_engine.module.connect(
+        # Creating and dropping databases is admin work, done directly over
+        # psycopg2 (always available in tests) regardless of the driver under
+        # test — CREATE DATABASE can't run inside a transaction, so it needs an
+        # autocommit connection.
+        import psycopg2
+
+        db_conn = psycopg2.connect(
             dbname=POSTGRES_BASE_DB,
             user=POSTGRES_USER,
             host=POSTGRES_HOST,
             port=POSTGRES_PORT,
             password=POSTGRES_PASSWORD,
         )
-        db_engine.attempt_to_set_autocommit(db_conn, True)
+        db_conn.autocommit = True
         cur = db_conn.cursor()
         cur.execute("DROP DATABASE IF EXISTS %s;" % (test_db,))
         cur.execute(
@@ -1303,15 +1344,15 @@ def setup_test_homeserver(
 
             dropped = False
 
-            # Drop the test database
-            db_conn = db_engine.module.connect(
+            # Drop the test database (admin work over psycopg2, as above).
+            db_conn = psycopg2.connect(
                 dbname=POSTGRES_BASE_DB,
                 user=POSTGRES_USER,
                 host=POSTGRES_HOST,
                 port=POSTGRES_PORT,
                 password=POSTGRES_PASSWORD,
             )
-            db_engine.attempt_to_set_autocommit(db_conn, True)
+            db_conn.autocommit = True
             cur = db_conn.cursor()
 
             # Try a few times to drop the DB. Some things may hold on to the

tests/storage/test_rust_dbapi.py

@@ -58,6 +58,22 @@ class BuildDsnTestCase(unittest.TestCase):
             "dbname=synapse user=u host=db port=5432",
         )
 
+    def test_maps_database_alias_to_dbname(self) -> None:
+        # psycopg2 accepts `database` as an alias for libpq's `dbname` (Synapse's
+        # sample config and most deployments use it); the strict libpq DSN the
+        # Rust pool parses only knows `dbname`, so it must be translated.
+        self.assertEqual(
+            rust_dbapi.build_dsn({"database": "synapse", "user": "u"}),
+            "dbname=synapse user=u",
+        )
+
+    def test_skips_none_values(self) -> None:
+        # `None` kwargs (unset config) are omitted, as psycopg2 treats them.
+        self.assertEqual(
+            rust_dbapi.build_dsn({"dbname": "d", "host": None, "port": None}),
+            "dbname=d",
+        )
+
     def test_quotes_values_needing_it(self) -> None:
         # Spaces / quotes / backslashes get single-quoted and escaped; empty → ''.
         self.assertEqual(

tests/utils.py

@@ -51,6 +51,9 @@ except ImportError:
 # POSTGRES_BASE_DB and update it to the current schema. Then, for each test case, we
 # create another unique database, using the base database as a template.
 USE_POSTGRES_FOR_TESTS = os.environ.get("SYNAPSE_POSTGRES", False)
+# Run the Postgres tests against the native Rust driver instead of psycopg2.
+# Only meaningful together with SYNAPSE_POSTGRES.
+USE_RUST_DRIVER_FOR_TESTS = os.environ.get("SYNAPSE_TEST_RUST_DRIVER", False)
 LEAVE_DB = os.environ.get("SYNAPSE_LEAVE_DB", False)
 POSTGRES_USER = os.environ.get("SYNAPSE_POSTGRES_USER", None)
 POSTGRES_HOST = os.environ.get("SYNAPSE_POSTGRES_HOST", None)