The `PG*` fallbacks for the DSN lived in `mod.rs` and the `PGSSL*` ones
in `tls.rs`, each with its own env helper. Put both behind one
`resolve_connection_defaults`, which the pool calls once, so the two
cannot drift apart and `tls::build` stays a pure function of its params.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ULdieJyQoHNzeHq42H2M1p
`DatabasePool` enforces `txn_limit` by counting transactions per worker
thread and reconnecting the thread's connection past the limit. The
Rust pool does not pin connections to threads, so that count recycled
whichever connection the thread happened to check out.
Give the native pool a `max_uses` instead: deadpool's recycle hook
retires a connection once it has served that many checkouts, which is
what `txn_limit` means to bound. `DatabasePool` skips its own counting
when the pool reports that it handles the limit.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ULdieJyQoHNzeHq42H2M1p
Pool exhaustion was invisible until checkouts timed out. Expose
deadpool's occupancy from the Rust pool and publish it as gauges per
pool (max connections, open connections, idle connections, waiting
checkouts), plus a histogram of the time a checkout waits for a
connection.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ULdieJyQoHNzeHq42H2M1p
develop gained an `idle_in_transaction_session_timeout` database option,
applied by the psycopg2 engine's `on_new_connection`. The Rust backend
does its session setup in the pool instead, so carry the setting through
`PoolConfig` and set it alongside `statement_timeout` when a connection
is opened.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ULdieJyQoHNzeHq42H2M1p
`create_pool`, `create_pool_with_session` and the eleven-argument
`ConnectionPool.__new__` grew one parameter at a time. One `PoolConfig`
with a `Default` replaces the wrapper, so a caller sets only the fields
it cares about and the next pool option is a field rather than a new
positional argument.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ULdieJyQoHNzeHq42H2M1p
`numeric` results had no decoding arm, so `SUM(bigint)`, `AVG(...)` and
the `::NUMERIC[]` cast in the room statistics query all failed with
"failed to decode column". Decode the wire format (base-10000 digits
with a weight and display scale) to its decimal text and hand that to
`decimal.Decimal`, which is what psycopg2 returns for these columns.
Arrays of numeric decode element-wise through the existing array path.
Binding a `Decimal` parameter is still unsupported; Synapse never does.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ULdieJyQoHNzeHq42H2M1p
A checkout from the Rust connection pool (`ConnectionPool.connect`) previously
blocked indefinitely while waiting for a free slot or establishing a new
connection, so an unreachable or overloaded server could hang the worker
threads handling requests forever.
Bound the whole checkout with a configurable timeout: deadpool's `wait` and
`create` timeouts (enabled via its `rt_tokio_1` runtime feature, driven on our
own tokio runtime). Hitting it raises an OperationalError, which Synapse
already treats as a retryable operational error.
Configurable via the `pool_checkout_timeout` database option (milliseconds;
default 30s, `0` disables), read by the Rust engine and plumbed to both the
connection pool and the bootstrap (pool-of-one) connection.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
The Rust backend connected with `NoTls`, so it couldn't talk to a Postgres that
requires (or, as libpq defaults, prefers) TLS — a blocker for most managed /
networked databases. Honour the same libpq keys psycopg2 already forwards, so an
existing `database.args` keeps working: `sslmode`, `sslrootcert`, `sslcert`,
`sslkey`, `sslpassword`.
Implemented on rustls (already in the tree via reqwest; pinned to the `ring`
provider since `aws-lc-rs` needs a C toolchain we don't have). `tokio-postgres`
only knows `sslmode` disable/prefer/require and does no verification itself, so a
new `postgres/tls.rs` drives both the handshake mode and the verifier:
disable -> no TLS
allow/prefer -> encrypt opportunistically, no verification (accept-any)
require -> encrypt, no verification (libpq's `require` does NOT verify)
verify-ca -> verify the chain, not the hostname (WebPKI, name bypassed)
verify-full -> verify chain + hostname (WebPKI)
`sslrootcert` (or the system trust store) supplies the roots; `sslcert`/`sslkey`
add a client certificate for mutual TLS. The cert-path keys and the `verify-*`
modes are stripped from the DSN and passed to the pool as explicit params
(`rust_dbapi.split_ssl_params`), since `tokio-postgres` can't parse them.
Behaviour change: the default (no `sslmode`) is now libpq's `prefer` — attempt
TLS, fall back to plaintext — where it was previously no TLS. This matches
psycopg2; plaintext servers still connect via the fallback.
Caveats vs libpq/openssl: rustls is stricter (requires SANs, rejects a
non-CA self-signed root), and an encrypted client key (`sslpassword`) isn't
supported and is rejected with a clear error.
Validated end-to-end against an SSL Postgres: require / verify-ca / verify-full
connect and use TLS, verify-full with only the system roots correctly rejects an
unknown CA, and hostname is checked (verify-full) / ignored (verify-ca). Unit
tests cover the sslmode->verifier mapping; the Rust-backend storage suites still
pass with the new default (prefer falls back against the plaintext test server).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d
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
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
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
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
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
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>
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
Introduce the Python<->Postgres value-mapping layer that the forthcoming
Rust DBAPI2 backend will build on, plus the module scaffolding to hang it
off (`synapse_rust.database.postgres`).
`PgValue` is an owned representation of a bound parameter that implements
`tokio_postgres::types::ToSql`, encoding into Postgres' binary wire format
according to the column type from the prepared statement (so a single
Python `int` becomes INT2/INT4/INT8 as appropriate, with range checks).
`PythonPgFromSql` is the decode counterpart, turning a column's wire bytes
back into the natural Python object (and SQL NULL into `None`). The two
`accepts` lists are kept in sync via tests.
The `value` submodule is exposed as `pub` for now so its
not-yet-consumed public items don't trip clippy's `dead_code` lint; later
changes that wire it into the cursor/connection code tighten that back up.
Covers int / float / bool / str / bytes / None; lists and richer types
(json, decimal, timestamps) are left to a follow-up.
The unit tests exercise the whole mapping without a live Postgres server,
including the float4 lossy narrowing, integer width boundaries, the
WrongType and out-of-range paths, and the unsupported/non-UTF-8 decode
errors.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Synapse has always connected to Postgres through libpq (via psycopg2),
so a connection string that omits `host=` gets libpq's default: the
compiled-in socket directory (which distributions such as Debian patch)
and environment variables like `PGHOST`. A Rust Postgres client such as
`tokio-postgres` applies its own default (`localhost`) instead. To keep
the existing connection behaviour, the Rust driver will ask the real
libpq what it would do and copy that across.
This adds the asking: `libpq::default_host` calls `PQconnectStart` on an
empty conninfo, which applies libpq's defaults and environment, and
reads the host libpq settled on back with `PQhost`. The connection
attempt that `PQconnectStart` begins is never polled and its outcome is
ignored: `PQhost` reports the resolved default whether or not a server
is listening there.
libpq is an optional runtime dependency, not a build-time one. The
shared library is loaded with `dlopen` (via `libloading`) the first time
`default_host` is called, and the five entry points we use are declared
by hand. Synapse therefore builds and imports without libpq installed;
only a deployment that needs the default host has to have `libpq5`, and
`default_host` returns a clear error if it is missing.
Split out of the Rust Postgres driver branch so the libpq dependency
can be reviewed on its own. Nothing calls `default_host` yet; the
driver's DSN fixup is the consumer.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Part of #19414 (Support Matrix 1.16).
⚠️ Before merging, wait for the following PRs to be merged first:
- https://github.com/element-hq/synapse/pull/20081
- https://github.com/element-hq/synapse/pull/20169
- https://github.com/element-hq/synapse/pull/20171
---
Since Matrix 1.16, the
[spec](https://spec.matrix.org/v1.19/client-server-api/#get_matrixclientv3sync)
defines the stable identifiers for
[MSC4222](https://github.com/matrix-org/matrix-spec-proposals/pull/4222):
the `use_state_after` query parameter on `GET /_matrix/client/v3/sync`,
and the `state_after` room response field:
> **`use_state_after`** — `boolean` — Controls whether to receive state
changes between the previous sync and the start of the timeline, or
between the previous sync and the end of the timeline. If this is set to
`true`, servers MUST respond with the state between the previous sync
and the end of the timeline in `state_after` and MUST omit `state`. […]
>
> **`state_after`** — […] Required if `use_state_after` was set to true
in the request, even if it is empty. Added in `v1.16`
>
> — [Matrix
Spec](https://spec.matrix.org/v1.19/client-server-api/#get_matrixclientv3sync)
Synapse so far only implemented the unstable identifiers
(`org.matrix.msc4222.use_state_after` /
`org.matrix.msc4222.state_after`), gated behind the `msc4222_enabled`
experimental flag and the per-user experimental features admin API.
This PR:
- accepts the stable `use_state_after` query parameter, unconditionally,
and responds with the stable `state_after` field. The unstable query
parameter is kept for a transition period; clients opting in with it get
the unstable response field name back, so unstable-era clients see an
unchanged wire format. If both parameters are set, the stable one wins.
- removes the `msc4222_enabled` experimental flag and the `msc4222`
per-user feature from the experimental features admin API, since the
behaviour is no longer gated.
### Pull Request Checklist
<!-- Please read
https://element-hq.github.io/synapse/latest/development/contributing_guide.html
before submitting your pull request -->
* [x] Pull request is based on the develop branch
* [x] Pull request includes a [changelog
file](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#changelog).
The entry should:
- Be a short description of your change which makes sense to users.
"Fixed a bug that prevented receiving messages from other servers."
instead of "Moved X method from `EventStore` to `EventWorkerStore`.".
- Use markdown where necessary, mostly for `code blocks`.
- End with either a period (.) or an exclamation mark (!).
- Start with a capital letter.
- Feel free to credit yourself, by adding a sentence "Contributed by
@github_username." or "Contributed by [Your Name]." to the end of the
entry.
* [x] [Code
style](https://element-hq.github.io/synapse/latest/code_style.html) is
correct (run the
[linters](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#run-the-linters))
---------
Co-authored-by: Eric Eastwood <madlittlemods@gmail.com>
TLDR: `PUT` on a server-default push rule now applies the same existence
check that `GET` already does. A rule that is not served cannot be
modified.
Synapse ships some server-default push rules behind experimental feature
flags, so the set of server-default rules a user has depends on the
server's configuration. Most flags add rules that only exist while the
flag is on.
[MSC4210](https://github.com/matrix-org/matrix-spec-proposals/pull/4210)
works the other way: it removes the three legacy mention rules while its
flag is on. The flags that change the set are:
- `msc1767_enabled`:
[MSC1767](https://github.com/matrix-org/matrix-spec-proposals/pull/1767)
extensible events rules
- `msc3381_polls_enabled`:
[MSC3381](https://github.com/matrix-org/matrix-spec-proposals/pull/3381)
poll rules
- `msc3664_enabled`:
[MSC3664](https://github.com/matrix-org/matrix-spec-proposals/pull/3664)
reply rule
- `msc4028_push_encrypted_events`:
[MSC4028](https://github.com/matrix-org/matrix-spec-proposals/pull/4028)
encrypted event rule
- `msc4306_enabled`:
[MSC4306](https://github.com/matrix-org/matrix-spec-proposals/pull/4306)
thread subscription rules
- `msc4210_enabled`: removes `.m.rule.contains_display_name`,
`.m.rule.contains_user_name` and `.m.rule.roomnotif`
`GET /pushrules/` only lists the rules that exist for that
configuration. `GET` on a single rule that is not in the set returns
`404 M_NOT_FOUND`. The same set is used to evaluate notifications, and
to build the `m.push_rules` account data in `/sync`.
`PUT .../enabled` and `PUT .../actions` did not use that set. They
accepted any rule ID that Synapse knows about, whether or not the server
currently serves it. Clients could modify a rule the server had just
said did not exist. The request returned `200`, the change was stored,
and it was then ignored. Both endpoints now return `404 M_NOT_FOUND` for
such a rule, as [the spec
requires](https://spec.matrix.org/v1.19/client-server-api/#put_matrixclientv3pushrulesglobalkindruleidenabled)
for a rule that does not exist.
Before (`msc4210_enabled: true`):
```
GET /_matrix/client/v3/pushrules/global/override/.m.rule.roomnotif
404 {"errcode": "M_NOT_FOUND", "error": "..."}
PUT /_matrix/client/v3/pushrules/global/override/.m.rule.roomnotif/enabled
{"enabled": false}
200 {}
```
After:
```
GET /_matrix/client/v3/pushrules/global/override/.m.rule.roomnotif
404 {"errcode": "M_NOT_FOUND", "error": "..."}
PUT /_matrix/client/v3/pushrules/global/override/.m.rule.roomnotif/enabled
{"enabled": false}
404 {"errcode": "M_NOT_FOUND", "error": "Unknown rule"}
```
This is a no-op for default configurations. All of these flags are off
by default, and rules that are served are still modifiable as before.
Part of: MSC4354 (Experimental Feature tracking issue #19409)
Follows: #20312 (which plumbs `received_ts` into the `Event` struct,
allowing us to calculate this correctly and sanely)
- When writing to `sticky_events`, we now use the `received_ts` stamped
onto the event to calculate the expiry,
rather than taking a second clock reading.
- When emitting the `unsigned.sticky_duration_remaining_ms` (currently
called `unsigned.msc4354_sticky_duration_ttl_ms`) field,
we now also use the `received_ts` as a clamp on the start of the
stickiness,
causing us to calculate the same expiry timestamp as the one saved to
`sticky_events`
(**This is a bugfix, as previously we weren't using `received_ts` here
like the MSC requires.**)
- The calculation of the expiry is centralised into the `Event` struct,
which I hope you'll agree is a
huge sanity improvement.
- We now systematically prevent spammy or rejected events from counting
as sticky.
---------
Signed-off-by: Olivier 'reivilibre <oliverw@matrix.org>
Co-authored-by: Erik Johnston <erikj@element.io>
Motivated by work on MSC4354, because the sticky events MSC specs that
`received_ts` is used to calculate the
start of an event's stickiness. (It would therefore be convenient to
have it available as part of event metadata, so an event can expose a
cheap helper method to check if it is sticky.)
This commit adds `received_ts` into the `event.internal_metadata` struct,
alongside `stream_ordering`.
I conceptually treat it as a sibling to `stream_ordering` and populate
it in all the same places where `stream_ordering` is populated.
Because this needs `received_ts` to be added to the HTTP replication
response for event persistence,
I also add a temporary compatibility measure that keeps rolling worker
upgrades working during this
transition.
I intend to remove the compatibility code in the next release.
I have manually tested that
```
SYNAPSE_POSTGRES=1 SYNAPSE_POSTGRES_USER=postgres SYNAPSE_POSTGRES_HOST=localhost SYNAPSE_POSTGRES_PASSWORD=postgres poetry run trial -j 6 tests.replication.test_sharded_event_persister
```
fails only when both the compatibility code is removed and the
`"received_ts"` key is commented out in the response (to omit it).
The compatibility code causes the test to pass even when `"received_ts"`
is commented out.
---------
Signed-off-by: Olivier 'reivilibre <oliverw@matrix.org>
To use `reactor.callFromThread` we need to take the GIL. However, the
vast majority of use-cases on the Rust side involve using that from
Tokio reactor threads, where we do not want to take the GIL (as it can
potentially block for a long time).
In #20252 we added a pure Rust alternative to `callFromThread`.
`run_python_awaitable` is the last usage of `callFromThread`, and so we
replace it with `dispatch_to_twisted`.
---
The diff is a lot smaller if you hide pure whitespace changes.
---------
Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
