Every `execute` did a fresh `prepare` + `query_raw` — two server round-trips
per query — and threw the `Statement` away; even BEGIN/COMMIT/ROLLBACK went
through a prepare. psycopg2 pays neither cost, which showed up as timing
margins under load (e.g. the worker-lock contention stress test timing out
under a parallel test run).
The pooled item is now a `PooledClient`: the `Client` plus a per-connection
statement cache keyed by SQL (`prepare_cached`). The cache lives with the
pooled connection — named prepared statements are per-session server state, so
they survive checkouts and a repeated query on any later checkout skips the
prepare round-trip. `PooledClient` derefs to `Client`, so existing call sites
are unchanged. The map is cleared at a size cap so one-off SQL (e.g.
execute_values' literal-spliced statements) can't grow it without bound; the
hot set re-warms in one round.
Concurrent DDL can invalidate a cached plan (SQLSTATE 0A000, "cached plan must
not change result type"). `execute` handles it: outside a transaction the
statement is re-prepared fresh and retried once; inside one (already aborted)
the stale cache entry is shed without I/O and the error propagates, so the
caller's next transaction gets a valid plan.
Transaction control (BEGIN/COMMIT/ROLLBACK) now uses `batch_execute` — the
simple-query protocol, one round-trip, nothing to prepare.
Verified: rust tests 158/158; storage/lock/worker-lock suites pass on the Rust
backend, including the worker-lock contention test under -j4 that previously
exceeded its wall-clock budget.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
The driver-agnostic base declares it and the executescript path reads it, but neither concrete engine set it, so booting on the Rust engine failed with an AttributeError in schema preparation.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ULdieJyQoHNzeHq42H2M1p
Small robustness fixes to `RustConnectionPool`, independent of any query path:
guard `connectionFactory` against a closed pool (a clear error rather than an
`AttributeError` on `None`), snapshot the per-thread connections before closing
them so a concurrent `connect()` can't mutate the dict mid-iteration, drop the
now-unused `defer_to_threadpool` import, and correct the class docstring to
describe the raw-`Deferred` contract.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
Two unit tests still asserted that `json` is rejected by the value mapping,
which became stale when json/jsonb decode and encode were added:
`accepts_lists_match_supported_types` now checks json/jsonb are accepted by
both `ToSql` and the Python decoder (while remaining outside the shared scalar
list), and `from_sql_rejects_unsupported_column_type` uses `uuid` as its
genuinely-unsupported type.
Full `cargo test` is green again (157 passed); caught by a full-suite run after
filtered runs had missed it.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
Synapse binds some binary values (e.g. slices of event data) as a `memoryview`
rather than `bytes`/`bytearray`. The shim's `from_py` handled the latter two but
not `memoryview`, so those parameters raised `TypeError: unsupported parameter
type for postgres: memoryview` — e.g. persisting an event over federation
(seen via `test_third_party_rules.test_on_new_event`).
Accept a `memoryview` as a BYTEA parameter, copying its bytes out with
`tobytes()` (the buffer protocol isn't available under the limited ABI the crate
builds against).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
A custom profile field value is a JSON document passed as text and cast
with `?::jsonb` in the query; reading one back binds a `jsonpath` text
parameter. `json`'s wire format is the raw text, and `jsonb` / `jsonpath`
share a one-byte version header followed by the text, so accept a `Text`
value for all three.
The matching query-side change (pass the value as a str instead of
`psycopg2.extras.Json`) is driver-independent and lives in its own PR.
Reading a `json`/`jsonb` column failed with "error deserializing column" — the
value mapping had no decoder for them. Synapse reads genuine `jsonb` values, most
visibly for custom profile fields (`get_profile_field` selects
`JSONB_EXTRACT_PATH(fields, ?)`), so those requests returned HTTP 500 on the Rust
backend.
Decode `json`/`jsonb` into the matching Python object (dict/list/scalar), as
psycopg2 does by default: `json` is the raw JSON text, `jsonb` a one-byte version
header (currently 1) then the text; parse with serde_json and convert via
pythonize. `PythonPgFromSql::accepts` now also accepts `json`/`jsonb`; the
Rust-native `DbValue` decoder stays scalar-only (nothing reads json through it).
Fixes the profile custom-field 500s (e.g. test_non_string,
test_can_lookup_own_profile) on the Rust backend.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
The value mapping bound arrays as parameters (a `list` → `Array`) but couldn't
decode an array *column* back out — the decode side assumed "Synapse only binds
arrays as parameters". That assumption was wrong: the caches replication stream
reads `keys` (a `text[]`) via `get_all_updated_caches`, so decoding failed with
"error deserializing column", the stream read errored, and a worker never
received bulk cache invalidations (the `wait_for_stream_position` in the test
hung and never fired).
Decode an array column into a Python `list`, each element decoded by the array's
element type (via `array_from_sql`); `PythonPgFromSql::accepts` now also accepts
arrays of a supported scalar element type, mirroring `PgValue`'s `ToSql`. The
Rust-native `DbValue` decoder stays scalar-only — nothing reads array columns
through it. Adds `fallible-iterator` (the version postgres-protocol already
uses) to iterate the decoded array's elements.
Fixes tests.storage.databases.main.test_cache.CacheInvalidationOverReplication's
test_bulk_invalidation_replicates on the Rust backend.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
`adbapi.ConnectionPool` obtains connections via a `connectionFactory` attribute,
which the database-outage tests swap out to make every checkout fail (and
restore afterwards). `RustConnectionPool` checked out connections inline, so it
had no such hook and the outage tests errored with `AttributeError`.
Give it a `connectionFactory` attribute (defaulting to a checkout from the
native pool, wrapped in the DBAPI2 adapter) and route both `runWithConnection`
and `connect` through it, so a test's replacement takes effect. Also make the
native pool reopenable: `close` drops it and `start` opens a fresh one, so the
tests' `close()`/`start()` outage cycle recovers rather than leaving the pool
permanently shut.
Fixes the tests.storage.databases.main.test_events_worker DatabaseOutage
failures on the Rust backend (12/12 pass); psycopg2 and sqlite are unaffected.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
Two value-mapping gaps surfaced by the storage suite on the Rust backend:
- `bytearray` parameters. Synapse deliberately passes binary data as
`bytearray` (the psycopg2 engine disables the `bytes` adapter to catch
accidental text-as-bytes bugs), but the shim only accepted `bytes`, so
binary background-update parameters raised `TypeError`. Accept `bytearray`
as a BYTEA parameter too.
- `tid` columns. The receipts-dedup background update selects a row's `ctid`
and then compares against it (`WHERE ctid != ?`). psycopg2 renders `tid` as
a `(block,offset)` string and accepts that string back; the shim had no
`tid` support at all. Decode `tid` to that same string (three big-endian
u16s: the block number's halves, then the offset — matching the server's
`tidsend`) and encode the string back to the wire form when it is bound as a
parameter.
Both directions are covered for the Python (`PythonPgFromSql`) and the
Rust-native (`DbValueFromSql`) decoders and the `ToSql` encoder, and `tid` is
added to the shared `accepts_column_type` set.
Fixes tests.storage.test_user_filters.test_bg_migration and
tests.storage.databases.main.test_receipts's linearized-unique-index background
update on the Rust backend.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
`new_transaction` sets a transaction's isolation level (for the receipts,
push-actions and purge paths, which ask for a specific level) by calling
`engine.attempt_to_set_isolation_level` before the transaction and resetting it
after. The Rust engine raised `NotImplementedError` there, so those transactions
errored out on the Rust backend.
The shim has no psycopg2-style `conn.set_isolation_level`, so implement it in
SQL: run `SET SESSION CHARACTERISTICS AS TRANSACTION ISOLATION LEVEL <level>` on
the connection. That sets the session's default for *subsequent* transactions,
so it's committed before the caller's transaction begins (SET is transactional);
`new_transaction` resets it to the default (REPEATABLE READ, which the pool
already applies at connection setup) afterwards, scoping the override to the one
transaction — matching psycopg2's behaviour. The level name comes from a fixed
`IsolationLevel` map, not caller input.
Fixes the isolation-level failures in tests.storage.test_event_push_actions,
tests.storage.test_purge and tests.storage.test_receipts on the Rust backend
(20/20 pass); psycopg2 and sqlite are unaffected.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
`RustConnectionPool` stands in for `DatabasePool._db_pool`, a
`twisted.enterprise.adbapi.ConnectionPool`. It presented the slice Synapse's
`DatabasePool` uses (`runWithConnection`, `threadID`), but some tests reach for
`_db_pool` directly and call adbapi's `runQuery` / `runOperation` / `connect`,
which weren't implemented — so those tests errored with `AttributeError` on the
Rust backend.
Add them, mirroring adbapi's semantics:
- `runQuery` / `runOperation` run one statement in its own transaction on a
worker thread (commit on success, roll back on error), returning the rows
(or nothing) as a `Deferred`.
- `connect` returns a connection for direct synchronous use on the caller's
thread, cached per thread as adbapi does and closed together with the pool.
Fixes the tests.storage.test_appservice and tests.storage.test_rollback_worker
failures on the Rust backend (26/26 pass); psycopg2 and sqlite are unaffected
(they use the real adbapi pool).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
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
`DatabasePool`'s `inner_func` calls `conn.reconnect()` when a connection is
found closed, or to recycle one that has exceeded the per-connection
transaction limit (`txn_limit`). The Rust DBAPI2 adapter had no `reconnect`, so
those paths (off by default, but real) would have raised.
Add `reconnect` to the adapter: it returns the current connection to the pool
(or discards it if unusable) and checks out a fresh one. The adapter now holds
the pool it was checked out of, with an `owns_pool` flag distinguishing a shared
pool (RustConnectionPool — reconnect from it, don't close it) from a bootstrap
pool-of-one (`rust_dbapi.connect` — closed together with the connection).
Tested: reconnect swaps in a working connection; runWithConnection and the
bootstrap/adapter paths are unchanged.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
A running homeserver exercises three shim capabilities beyond what the unit
tests reached; add them so it boots on the Rust backend end to end:
- `prepare_database` reads `db_conn.autocommit`; add a readable `autocommit`
getter to the shim `Connection` (and the adapter), mirroring psycopg2's
property.
- schema preparation runs multi-statement scripts via
`engine.execute_script_file` -> `cursor.executescript`; route
`LoggingTransaction.executescript` to the shim's multi-statement primitive
for the Rust backend (as it already does for sqlite), and expose
`executescript` on the DBAPI2 adapter cursor.
- store loading binds Python lists as Postgres arrays for `column = ANY($1)` /
`!= ALL($1)` (Synapse's `make_in_list_sql_clause`); add a `PgValue::Array`
variant that classifies a `list` and encodes it as an array of the element
column type, and accept array column types in `ToSql`.
With these, a homeserver boots on the Rust backend: it prepares the full schema
(176 tables), listens, `/health` returns OK, and DB-backed client endpoints
respond correctly.
Tested: array `to_sql` / classification and array-column `accepts` (Rust unit
tests); and, against a live Postgres, `= ANY($1)` list binding, adapter
`executescript`, and the `autocommit` getter. sqlite and psycopg2 homeserver
boots remain green.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
The Rust `ConnectionPool` takes the homeserver's `RustRuntime`, whose
tokio runtime drives the pool. `make_pool`, `rust_dbapi.connect` and
`make_conn` now take it too, and `DatabasePool` and `Databases` pass
`hs.get_rust_runtime()`. `make_conn` falls back to a standalone runtime
on the global reactor for the scripts and tests that have no homeserver
to take one from.
The adapter and engine tests build their pools through the shared
`_make_pool` helper for the same reason.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ULdieJyQoHNzeHq42H2M1p
Turn the Rust backend on with `database.use_rust_driver: true` (alongside
`name: psycopg2`). Keeping `name: psycopg2` means the engine is still selected as
Postgres and all the `isinstance(engine, PostgresEngine)` dialect checks hold;
the flag only swaps the driver implementation.
- `create_engine` returns `RustPostgresEngine` instead of `Psycopg2Engine` when
the flag is set.
- `make_pool` gains a `clock` argument and, for the Rust engine, builds a
`RustConnectionPool` (via `_make_rust_pool`) instead of an adbapi pool:
it derives the libpq DSN from the config `args`, sizes threads/connections
to `cp_max`, passes the engine's synchronous_commit / statement_timeout,
starts the pool, and registers a shutdown hook via the clock. Its declared
return type stays `adbapi.ConnectionPool` — `RustConnectionPool` provides the
`_db_pool` subset `DatabasePool` uses (runWithConnection / threadID /
running / threadpool) — so callers that lean on adbapi-specific methods keep
type-checking.
- `RustConnectionPool` takes synchronous_commit / statement_timeout_ms and
threads them into its pooled connections' session setup.
- `DatabasePool` passes its clock to `make_pool`; the test harness's
`make_fake_db_pool` accepts the new argument.
Tested: create_engine honours the flag (and the Rust engine is still a
PostgresEngine, not a Psycopg2Engine); make_pool builds a started
RustConnectionPool that serves a query as `_db_pool`. psycopg2 and sqlite
homeserver boots remain green. A full homeserver boot on the Rust backend isn't
covered here because the test harness patches `make_pool` with a synchronous
adbapi fake.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
Startup opens a bootstrap connection via `make_conn` (outside the runtime pool)
and validates it with `engine.check_database` before `prepare_database` runs.
Neither worked for the Rust backend: the Rust module has no `module.connect`,
and check_database/server_version were NotImplementedError stubs. Fill them in:
- `rust_dbapi.build_dsn` turns the libpq-style `args` into a DSN string, and
`rust_dbapi.connect` opens a standalone connection (a pool of one, kept
alive by the returned Connection) for one-off bootstrap use.
- `make_conn` routes the Rust engine through `rust_dbapi.connect` (with the
engine's synchronous_commit / statement_timeout) instead of
`engine.module.connect`; psycopg2 and sqlite are unchanged.
- `RustPostgresEngine.check_database` reads the server version over a cursor
(`SHOW server_version_num`) rather than psycopg2's `conn.server_version`,
and applies the same version / encoding / collation / ctype checks;
`server_version` is derived from the cached value.
Tested: build_dsn quoting; and, against a live Postgres, make_conn +
check_database + server_version + a query. psycopg2 and sqlite homeserver boots
(make_conn → check_database) remain green.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
Make RustConnectionPool a drop-in for DatabasePool._db_pool so real
runInteraction traffic can flow through it (once make_pool is switched):
- it now hands each function a DBAPI2-adapter connection (rust_dbapi.Connection)
wrapping the pooled shim, so LoggingDatabaseConnection.cursor() yields a
working LoggingTransaction and the engine's in_transaction / is_closed /
set_autocommit operate on it;
- the entry point is named `runWithConnection` (matching
twisted.enterprise.adbapi.ConnectionPool, which database.py calls by that
name), alongside `threadID` for the transaction-limit path. With
`threadpool` and `running` already present, the pool covers the slice of the
adbapi interface DatabasePool uses.
A new test drives a full transaction through the pool the way
DatabasePool.runWithConnection's inner_func does — engine.in_transaction check,
LoggingDatabaseConnection + LoggingTransaction, `?`→`$n` conversion, commit —
off the reactor thread, and gets the result back via the Deferred.
Still outstanding before make_pool can return it: `reconnect` on the connection
(transaction-limit / closed-connection paths) and the startup path
(make_conn / check_database).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
Previously RustPostgresEngine subclassed the psycopg2 PostgresEngine, which
dragged psycopg2 into the Rust path and conflated "is Postgres" with "is
psycopg2". Split the hierarchy so both drivers are siblings:
- `PostgresEngine` (new `postgres_base` module) is now the driver-agnostic
base holding the shared SQL-dialect and config logic (single_threaded,
supports_using_any_list, row_id_name, get_db_locale, check_new_database,
lock_table, synchronous_commit / statement_timeout). It has no driver
dependency, so it always imports. Everything touching a live connection,
the DBAPI2 exception module, or the placeholder style is left abstract.
- `Psycopg2Engine(PostgresEngine)` holds the psycopg2 specifics (register_type
/ register_adapter, isolation-level map, conn.status/closed/server_version,
`%s` placeholders, psycopg2 execute path, `uses_psycopg2_extras = True`).
- `RustPostgresEngine(PostgresEngine)` is re-parented onto the base (no longer
inherits psycopg2). It passes the Rust DBAPI2 module to the base and gets
NotImplementedError stubs for the still-psycopg2-shaped check_database /
server_version (part of the deferred startup wiring).
The base keeps the name `PostgresEngine`, so all ~91
`isinstance(engine, PostgresEngine)` checks across the storage layer (which mean
"emit Postgres SQL") hold for both drivers unchanged. `create_engine` now
returns `Psycopg2Engine` for `name == "psycopg2"`.
Verified: full lint clean; Rust engine/adapter tests pass; a psycopg2 homeserver
boots and runs (test_room_search under Postgres) and the sqlite path is
unaffected (test_room_search / test_event_federation under sqlite).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
Synapse's LoggingTransaction drives a cursor through the DBAPI2 spelling
(fetchone/fetchmany/fetchall, iteration, rowcount/description as properties),
but the Rust shim cursor exposes fetch_one/fetch_all/fetch_next_batch and
rowcount()/description() as methods. Rather than reshape the Rust API, add thin
Python `Connection`/`Cursor` wrappers that present the DBAPI2 shape and delegate
to the shim:
- `Cursor` maps fetchone/fetchmany/fetchall/__iter__ and rowcount/description
onto the shim, and tracks exhaustion so fetching past the end keeps
returning "no more rows" (the shim raises instead);
- `Connection.cursor()` returns the adapter cursor, and the transaction-control
and engine-facing methods (commit/rollback/close, set_autocommit, is_closed,
in_transaction) delegate straight through.
An end-to-end test drives a real LoggingTransaction backed by the adapter and
RustPostgresEngine: `?` placeholders are converted to `$n`, a query runs, and
rows come back via fetchone/iteration/description.
Not handled yet: execute_batch / execute_values (psycopg2 extras that
LoggingTransaction calls directly for PostgresEngine) still need a routing
change to reach a shim-backed implementation.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
Move the equivalent of the engine's `on_new_connection` into the pool's
`ConnectionManager::create`, so every pooled connection is configured once when
it is opened rather than on each checkout. A new `SessionConfig` (carried by the
manager) drives it:
- the default isolation level is set to REPEATABLE READ, matching the engine's
`default_isolation_level`, so a plain BEGIN behaves as it did under psycopg2;
- `synchronous_commit` is turned off when configured;
- `statement_timeout` is set when configured.
`bytea_output` is deliberately not set: tokio-postgres uses the binary protocol
for prepared statements, so that text-format GUC is irrelevant to how we decode
bytea.
`create_pool` keeps its signature (default session); a new
`create_pool_with_session` and two new keyword-only `ConnectionPool(...)`
arguments (`synchronous_commit`, `statement_timeout_ms`) expose the settings.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
A database engine that drives the Rust Connection/Cursor shim instead of
psycopg2. It subclasses PostgresEngine to reuse the pure SQL-generation and
configuration behaviour, and overrides only the parts that touch a live
connection or are wired to psycopg2 internals:
- `module` points at the Rust backend's DBAPI2 exception hierarchy, which the
transaction driver catches on;
- `convert_param_style` rewrites `?` to `$1, $2, ...` (the shim binds libpq
positional placeholders, not psycopg2's `%s`);
- `in_transaction` / `is_connection_closed` / `attempt_to_set_autocommit` call
the shim's own methods;
- `is_deadlock` matches the Rust `DatabaseError` and its `pgcode`;
- `executescript` uses the shim's multi-statement primitive;
- `on_new_connection` is a no-op — session setup belongs in the Rust pool.
Per-transaction isolation-level overrides raise NotImplementedError for now, and
`check_database` / `server_version` still read psycopg2 attributes; the engine
is not yet selectable via `create_engine`, so neither is reached. Wiring it into
`make_pool` (which also feeds session config to the pool) is the next step.
Tested directly: param-style rewriting, deadlock/pgcode matching, the module
pointer, and — against a live shim connection — in_transaction, is_closed and
autocommit.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
Match adbapi's `_runWithConnection` contract in `RustConnectionPool._run`:
a transaction the function leaves open is committed when it returns and
rolled back when it raises, before the connection goes back to the pool.
The commit makes callers that rely on the pool's implicit commit durable,
and is a no-op after functions that commit themselves. The rollback
matters more: without it every raising transaction function hands a
mid-transaction connection back to the shim, which discards it (tearing
down the TCP session) rather than returning it to the pool for reuse.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ti5yoaVfjHYZ3LmzXFxSwq
The database engine's `is_connection_closed` and `in_transaction` checks (used
by the transaction driver before reusing a pooled connection) need to read
these off the raw connection, as they do from psycopg2's `conn.closed` /
`conn.status`. Add the two accessors to the shim: `is_closed()` is true once the
connection is closed/returned/discarded or the socket is torn down, and
`in_transaction()` reports the tracked transaction flag.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
`RustConnectionPool` runs synchronous Synapse transaction functions off the
reactor thread against the native Rust connection pool. Each
`run_with_connection(func, *args)` call runs, on a dedicated Twisted thread
pool, `func(conn, *args)` with a `Connection` checked out of the Rust pool, and
returns a Deferred that fires on the reactor with the result (or an errback if
it raised). It builds on `defer_to_threadpool`, so log contexts are preserved
across the hop, and returns the connection to the pool afterwards (clean →
reused, mid-transaction/broken → discarded, per the shim's disposal rules).
The Rust pool sizes 1:1 with the thread count, and `PyConnectionPool.close()`
(new) closes all idle connections so the owner can drop the server connections
deterministically. The bridge is lifecycle-owner-managed (start/close) and
takes no clock, keeping it trivially testable; slotting it into
`DatabasePool.make_pool` — which also needs engine-level support for the shim
connection (in_transaction, is_connection_closed, autocommit/isolation,
reconnect) — is left to a follow-up.
Tested against a live Postgres over the real reactor (Synapse's in-memory test
reactor mocks the DB thread pool out, so a plain Twisted trial TestCase is used):
result/return-value, arg forwarding, exception-to-errback, connection reuse
across calls, and concurrent checkouts.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
The Python-facing `Connection`/`Cursor` shim now always wraps a connection
checked out of the `deadpool` pool. `ConnInner` holds a `PooledConnection` and
disposes of it correctly when the last reference (the `Connection` and all its
cursors) goes away, via a `Drop` impl plus `release`/`discard` helpers:
- a clean connection is returned to the pool for reuse;
- a connection is *discarded* (detached with `Object::take`, which also
shrinks the pool) whenever reuse would be unsafe — a failed COMMIT/ROLLBACK,
a poisoned mutex, or being dropped with a transaction still open (which
can't be rolled back synchronously from `Drop`, so the socket close makes
the server do it).
A plain query error still does *not* throw the connection away — it stays
open+aborted for the driver to `rollback()`, exactly as psycopg2 behaves.
The pool is the only way to obtain a connection: the standalone `connect(dsn)`
free function is replaced by a Python-facing `ConnectionPool` class (Rust
`PyConnectionPool`, exposed as `postgres.ConnectionPool`). Build it once from a
DSN, then check connections out with `pool.connect()`. Checkout failures map
onto the DBAPI2 hierarchy — a backend connect error reuses the query-error
mapping (and its `pgcode`), while a timeout / closed pool becomes
`OperationalError` — so `connect()` behaves like psycopg2's for callers.
Adds live-Postgres tests (gated on SYNAPSE_TEST_POSTGRES_DSN) asserting which
connections end up back in the pool, and drives the Python test suite through a
pool in `setUp`.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
Introduce the native Rust connection pool that both the Rust-native
query path and the Python-facing Connection/Cursor shim will draw from,
so we run a single pool rather than two that could exhaust the server's
connection limit between them.
Uses the generic deadpool::managed pool with our own ConnectionManager
(rather than deadpool-postgres) so connection creation reuses the
existing connect/default-host logic and connection-task spawning. The
pooled item is a plain tokio_postgres::Client, used with the standard
async query functions; recycle() drops connections the driver reports
closed. Live tests are gated on SYNAPSE_TEST_POSTGRES_DSN.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Implement DBAPI2's `executemany`, which Synapse uses for batched writes.
The statement is prepared once and run for each parameter set inside the
connection's (lazily opened) transaction, so a failure part-way aborts
the whole batch. The per-set executions are pipelined — their futures
are driven concurrently so tokio_postgres streams the batch onto the
connection in one round-trip rather than one per statement.
Afterwards `rowcount` reports the total rows affected across all
executions (as psycopg2 does) via a new fetchless "command complete"
cursor state; there is no result set to fetch or describe. An empty
parameter sequence is a no-op that leaves rowcount at -1.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The column names for a result set were already carried through the
cursor state machine but kept write-only, awaiting a reader. Add the
PEP-249 `Cursor.description` accessor on top of that plumbing: it
returns one 7-tuple per column (only the name populated, which is all
Synapse reads), or `None` when there is no row-returning result set —
before any query, after an error, or for a column-less statement such
as a bare INSERT, matching psycopg2.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Synapse's schema setup (`prepare_database.py`) runs `;`-separated SQL
scripts via the engine's `executescript`. `Cursor.execute` can't serve
those: it `prepare`s the query, and Postgres rejects multiple commands in
a prepared statement.
Add `Cursor.executescript`, which runs the whole script on the simple-query
protocol (`batch_execute`), which does allow multiple statements. It takes
no parameters and produces no fetchable rows.
The script runs inside the connection's *current* transaction (opened
lazily like `execute`) and is left open for the caller to commit. It
deliberately does NOT reproduce the commit-any-pending-transaction-first
behaviour of `sqlite3.executescript` (which psycopg2's engine mirrors with a
leading `COMMIT`). That forced commit actually undercuts the atomicity
`prepare_database` sets out to get — it opens a transaction so "upgrades are
either applied completely, or not at all", but the first script's implicit
commit ends it. Running the script within the ongoing transaction instead
lets successive scripts accumulate and be committed once, which is both
simpler and more correct. The only engine-level piece left to layer on top
is the auto-increment placeholder substitution.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Previously every `tokio_postgres` error became a bare `RuntimeError`. Synapse's
transaction driver, though, branches on the *type* of a database error and on
its `pgcode`: `new_transaction` retries `OperationalError`, retries deadlocks it
recognises via `is_deadlock` (which reads `pgcode`) on a `DatabaseError`, and
`simple_upsert` retries `IntegrityError`. With everything collapsed to
`RuntimeError` none of that fired.
Add just the distinctions Synapse acts on, rather than psycopg2's full PEP-249
hierarchy: `Error` -> `DatabaseError` -> {`OperationalError`, `IntegrityError`},
exposed on the `postgres` submodule, each instance tagged with `pgcode` (the
SQLSTATE string, or `None`). A small classifier maps the SQLSTATE class:
constraint violations (`23`) to `IntegrityError`, connection/resource classes
(`08`/`53`/`57`/`58`) to `OperationalError`, everything else (incl. `40*`
deadlocks, which retry via `pgcode`) to `DatabaseError`. Codeless errors are
split with `is_closed()`: a lost connection is operational, any other (a bad
parameter, a failed connect) is a plain `DatabaseError` so it isn't retried.
Errors surfacing while a result stream is drained (the usual case for an
`INSERT` constraint violation) now route through the same mapping, so they carry
the right class and `pgcode` too.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Add the end-to-end test suite for the Rust DBAPI2 backend, driving the real
tokio-postgres client against a live Postgres server. It is skipped unless
the suite is configured for Postgres (SYNAPSE_POSTGRES), the same switch the
rest of the suite uses.
These cover the behaviours that need a real server rather than the
in-memory fakes the Rust unit tests use: connect (good and bad DSN),
run_interaction (return value, arg/kwarg forwarding, commit on success,
rollback on a Python exception, and recovery after both a Python-raised and
a server-rejected statement, including a constraint violation), cursor
reuse across queries, the fetch_one/fetch_all/fetch_next_batch/rowcount
surface (including batching across a 1000-row result set and the
exhausted-vs-no-active-query error distinction), and value round-trips for
every supported type (NULL, bytea with NUL/high bytes, float4 lossiness,
and an out-of-range int bound to a real int4 column).
A separately-guarded case exercises the libpq default-host fixup by
connecting with a DSN that omits the host.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
tokio_postgres has a different default host than libpq, which is what
Synapse used via psycopg2 (and what psql uses). libpq's default host is
configurable at build time, so when the DSN omits a host ask libpq what
its default would be, through the `crate::libpq` wrapper that loads the
system libpq at runtime, and use that.
Add the Python-facing `Connection` / `Cursor` pair and the `connect`
factory, implementing enough of the PEP-249 (DBAPI2) shape for Synapse's
needs.
A `Connection` owns a single `tokio_postgres::Client`. The client *moves*
between the connection and an in-flight cursor rather than being shared:
it lives in the connection between interactions, is taken out for the
duration of a cursor, and is handed back when the cursor finishes. That
single-owner baton (an `Option<Client>` slot on each side) makes it
structurally impossible to use the connection mid-transaction or to run
two overlapping transactions on one socket — both become a clean "already
closed" error.
The transaction lifecycle (`cursor` opens with `BEGIN`; `finish`
COMMIT/ROLLBACKs and hands the client back; `CursorGuard` rolls back an
abandoned transaction on drop) is included here. On any
transaction-control error the client is dropped rather than returned,
closing the socket — safer than handing a possibly-broken connection back
to what will become a connection pool.
`connect()` parses a libpq-style DSN, blocks until connected, and spawns
the long-lived connection task onto the shared runtime (the libpq
default-host fixup is a follow-up). The cursor query methods
(`execute` / `fetch_one` / `fetch_all` / `fetch_next_batch` / `rowcount`)
delegate to the `CursorQueryState` machine.
`run_interaction` — the high-level glue that opens a cursor, runs the
callback, and commits/rolls back — follows in the next change, so
`cursor`/`finish`/`CursorGuard` carry a transitional `#[allow(dead_code)]`
until then. Now that the value/helpers/cursor_state modules are consumed
internally, their visibility is tightened from `pub` back to private.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Comments only, no functional change. Describe the current code without
the staging history, drop comments that restate the next line, and fix
the test module doc, which said the blocking path was never exercised
here even though the SteppedStream tests do exercise it.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EvkFDjwNGBQcaxtQQ1YoUD
Every path that drains the stream (`fetch_one`, `fetch_next_batch`,
`fetch_all`, `rowcount`) moved the cursor to `Closed` by hand, each
re-reading the command tag and moving the column names across. Pull that
into one `close_drained` helper. This also removes the re-borrow dance in
`fetch_next_batch`'s exhausted arm and the free `rows_affected` function,
which shared its name with the trait method and the `RowStream` inherent
method it wrapped.
Map stream errors through the module's existing `pg_err_to_py` rather
than a second ad hoc message format, taking the error by value since
every caller owns it.
Drop the `#[allow(dead_code)]` on `Closed::description`: the enum is
public, so the lint never fired.
Add the licence header the other files in the crate carry.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EvkFDjwNGBQcaxtQQ1YoUD
Add a batched fetch alongside `fetch_one`/`fetch_all`: it blocks for the
first row, then scoops up any further rows that are already buffered
without blocking again, returning them as one batch. This lets Python
iterate large result sets without the per-row overhead of `fetch_one`,
while still not blocking on the whole result set like `fetch_all`.
Exhaustion is reported by an empty batch, and deliberately deferred: a
batch that runs into the end of the stream still returns the rows it has
and leaves the empty-batch report (and the move to `Closed`) to the next
call. The fused stream makes that re-poll safe.
Unit tests use the in-memory fakes from the previous change, plus a new
`SteppedStream` that can report "not ready yet" mid-stream, so the partial
-batch boundary, the deferred empty report, the capacity-is-a-hint
behaviour, interleaving with `fetch_one`, and the mid-batch error path are
all covered without a live database.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Model the lifecycle of a cursor's most recent query as an explicit state
machine (Idle / Active / Closed) so that illegal field combinations are
unrepresentable and fetching past the end of a result set is a clean,
specific error rather than the spurious "connection closed" you get from
re-polling a finished stream. The row stream is fused so an
exhausted-but-not-yet-reported stream can sit safely in `Active`.
This change adds `fetch_one`, `fetch_all` and `rowcount` (the batched
`fetch_next_batch` follows separately). On a stream error a cursor resets
to `Idle`; on normal exhaustion it moves to `Closed`, retaining the
PEP-249 rowcount from the command tag.
The state machine is generic over the stream type, defaulting to
`RowStream`, via a small `CursorRowStream` trait that abstracts the three
things the logic needs (the affected-row count, row->PyTuple conversion,
and error rendering). This is what lets the state transitions, exhaustion
handling and error recovery be unit-tested against an in-memory fake
stream, with no live Postgres server.
`cursor_state` is `pub` for now so its not-yet-consumed items don't trip
clippy's dead_code lint; the connection code wires it up and tightens that
later.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Rather than resolving the runtime from the thread-local context via
Handle::current(), which requires every calling thread to have entered
the shared runtime first (and panics when it hasn't), take the handle
as an argument.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NoZXfF47gtdcGLpfCUcXXU
Add the `block_on`/`block_on_result`/`block_on_next` helpers the Postgres
backend uses to drive its async `tokio-postgres` futures to completion from
sync, GIL-holding Python methods, releasing the GIL for the wait. They take
a `tokio::runtime::Handle` and block on it from the calling (Python) thread.
Rather than give the DB backend a runtime of its own, they use the
extension's existing shared runtime (`tokio_runtime::PyTokioRuntime`, stored
on the reactor). `start` is made idempotent and a `runtime_handle` accessor
starts it on demand, so a caller that needs a connection before the reactor
is running still gets a handle; once the reactor runs, its
`callWhenRunning(start)` hook is a no-op.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RPEeXx2fAG67o6u4CnmC8W
