Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Postgres foundation: ONE shared pool, embedded migrations,
3 : //! transaction-scoped tenancy. Store implementations build on top.
4 :
5 : pub mod doc;
6 : pub mod entity;
7 : pub mod maintenance;
8 : pub mod outbox;
9 : pub mod seal;
10 : pub mod temporal;
11 :
12 : use antares_model::TenantId;
13 : use serde_json::Value;
14 : use sqlx::postgres::{PgPool, PgPoolOptions};
15 : use sqlx::{Postgres, Transaction};
16 :
17 : /// Embedded migrations, run at start (like Scorpio's Flyway, but once).
18 : ///
19 : /// `ANTARES_MIGRATE=0`/`false` skips that run on this process, so serving
20 : /// replicas do not race the same DDL and a deployment can migrate once from a
21 : /// separate job or init container against the same database. Unset — the
22 : /// default — migrates on boot exactly as before.
23 : pub static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("./migrations");
24 :
25 : /// Is this connect failure a database whose migration history can never match
26 : /// this binary? A database still booting is worth waiting for; a history from
27 : /// another release is not, and a caller that retries it reports the timeout
28 : /// instead of the cause. Only the history errors are permanent: an execution
29 : /// failure can be a lock the next attempt gets.
30 : /// The filter values a stored text column can equal. PostgreSQL `text` holds
31 : /// no NUL, so a request string carrying one names nothing the store holds;
32 : /// binding it would raise `22021` and turn a query that matches nothing into
33 : /// an error. Callers bind what this keeps.
34 5194 : pub(crate) fn storable<S: AsRef<str>>(values: &[S]) -> Vec<String> {
35 5194 : values
36 5194 : .iter()
37 5194 : .map(AsRef::as_ref)
38 6546 : .filter(|v| !v.contains('\0'))
39 5194 : .map(str::to_owned)
40 5194 : .collect()
41 5194 : }
42 :
43 79 : pub fn is_schema_mismatch(e: &sqlx::Error) -> bool {
44 : use sqlx::migrate::MigrateError::{
45 : Dirty, VersionMismatch, VersionMissing, VersionNotPresent, VersionTooNew, VersionTooOld,
46 : };
47 79 : match e {
48 15 : sqlx::Error::Migrate(m) => matches!(
49 15 : **m,
50 : VersionMissing(_)
51 : | VersionMismatch(_)
52 : | VersionNotPresent(_)
53 : | VersionTooOld(..)
54 : | VersionTooNew(..)
55 : | Dirty(_)
56 : ),
57 64 : _ => false,
58 : }
59 79 : }
60 :
61 : /// What the boot path does with the migrations a newer release applied
62 : /// (`newer`: version and description of each applied version above `newest`,
63 : /// this binary's own newest). A rollback starts the previous image on the
64 : /// schema the release it rolls back already migrated, and sqlx refuses every
65 : /// applied version it does not ship, so without this the previous image could
66 : /// never boot (docs/src/operations.md "Rollback"). A migration whose file name
67 : /// starts with `breaking_` is one this release cannot serve: the way back is
68 : /// the pre-release backup, so the boot stops with that version. `Ok(true)`
69 : /// means newer, compatible migrations are applied and tolerated.
70 363 : fn tolerate_newer(
71 363 : newest: i64,
72 363 : newer: &[(i64, String)],
73 363 : ) -> Result<bool, sqlx::migrate::MigrateError> {
74 363 : match newer.iter().find(|(_, d)| d.starts_with("breaking ")) {
75 4 : Some((v, _)) => Err(sqlx::migrate::MigrateError::VersionTooNew(*v, newest)),
76 359 : None => Ok(!newer.is_empty()),
77 : }
78 363 : }
79 :
80 : /// The first applied version at or below this binary's newest that `known`
81 : /// (the versions [`MIGRATOR`] ships) lacks: a history from another release,
82 : /// which the lenient run below would otherwise pass over.
83 8 : fn first_unknown_older(applied: &[i64], known: &[i64]) -> Option<i64> {
84 71 : applied.iter().copied().find(|v| !known.contains(v))
85 8 : }
86 :
87 : /// Version and description of each applied migration above `newest`.
88 355 : async fn newer_migrations(
89 355 : conn: &mut sqlx::PgConnection,
90 355 : newest: i64,
91 355 : ) -> Result<Vec<(i64, String)>, sqlx::Error> {
92 355 : sqlx::query_as(
93 355 : "SELECT version, description FROM _sqlx_migrations WHERE version > $1 ORDER BY version",
94 355 : )
95 355 : .bind(newest)
96 355 : .fetch_all(conn)
97 355 : .await
98 355 : .or_else(|e| match &e {
99 : // a database never migrated has no history table yet
100 7 : sqlx::Error::Database(d) if d.code().as_deref() == Some("42P01") => Ok(Vec::new()),
101 0 : _ => Err(e),
102 7 : })
103 355 : }
104 :
105 : /// Run [`MIGRATOR`], tolerating compatible migrations of a newer release
106 : /// ([`tolerate_newer`]). An unknown applied version at or below this binary's
107 : /// newest is still a history from another release and refused as before.
108 355 : async fn migrate(conn: &mut sqlx::PgConnection) -> Result<(), sqlx::Error> {
109 355 : let newest = MIGRATOR.iter().map(|m| m.version).max().unwrap_or(0);
110 355 : let newer = newer_migrations(conn, newest).await?;
111 355 : let tolerate = tolerate_newer(newest, &newer).map_err(|e| {
112 2 : tracing::error!(
113 : "the database holds a breaking migration of a newer release ({newer:?}): this \
114 : release cannot serve it; restore the backup taken before that release"
115 : );
116 2 : sqlx::Error::Migrate(Box::new(e))
117 2 : })?;
118 353 : if !tolerate {
119 351 : return MIGRATOR.run(conn).await.map_err(Into::into);
120 2 : }
121 2 : tracing::warn!(
122 : "the database holds migrations {newer:?} of a newer release, all marked compatible: \
123 : serving it as a rollback; roll forward again or restore the pre-release backup \
124 : before applying another schema change"
125 : );
126 2 : let applied: Vec<i64> =
127 2 : sqlx::query_scalar("SELECT version FROM _sqlx_migrations WHERE version <= $1")
128 2 : .bind(newest)
129 2 : .fetch_all(&mut *conn)
130 2 : .await?;
131 2 : let known: Vec<i64> = MIGRATOR.iter().map(|m| m.version).collect();
132 2 : if let Some(v) = first_unknown_older(&applied, &known) {
133 1 : return Err(sqlx::Error::Migrate(Box::new(
134 1 : sqlx::migrate::MigrateError::VersionMissing(v),
135 1 : )));
136 1 : }
137 1 : let mut lenient = sqlx::migrate::Migrator::with_migrations(MIGRATOR.iter().cloned().collect());
138 1 : lenient.set_ignore_missing(true);
139 1 : lenient.run(conn).await.map_err(Into::into)
140 355 : }
141 :
142 : /// The `ANTARES_MIGRATE` switch: off only for an explicit `0`/`false`.
143 365 : fn migrate_enabled(v: Option<&str>) -> bool {
144 365 : !matches!(v, Some("0" | "false"))
145 365 : }
146 :
147 : /// One shared pool for all tenants — never per-tenant pools.
148 : /// `max_connections` ≈ 2× the PG box's cores; the default suits a small dev
149 : /// Postgres, deployments size it via config.
150 : ///
151 : /// Every acquisition and session is bounded: a saturated pool fails the one
152 : /// request after 5 s instead of queueing forever; idle/aged connections are
153 : /// recycled; and each session carries `statement_timeout`/`lock_timeout` so
154 : /// a runaway query or lost lock can never wedge a pooled connection.
155 354 : pub async fn connect(url: &str, max_connections: u32) -> Result<PgPool, sqlx::Error> {
156 354 : connect_with(url, max_connections, std::time::Duration::from_secs(30)).await
157 354 : }
158 :
159 : /// How long a released connection may take to answer its liveness ping
160 : /// before it is dropped instead of going back to the pool.
161 : pub const RELEASE_PING: std::time::Duration = std::time::Duration::from_secs(2);
162 :
163 : /// [`connect`] with the per-session `statement_timeout` chosen by the
164 : /// deployment (`ANTARES_PG_STATEMENT_TIMEOUT_MS`).
165 421 : pub async fn connect_with(
166 421 : url: &str,
167 421 : max_connections: u32,
168 421 : statement_timeout: std::time::Duration,
169 421 : ) -> Result<PgPool, sqlx::Error> {
170 421 : let pool = options(max_connections, statement_timeout)
171 421 : .connect(url)
172 421 : .await?;
173 : // DDL is exempt from the per-session statement timeout: building an index
174 : // on a large attr_instances legitimately runs longer than a query ever
175 : // should. The lock timeout stays — a migration blocked on a lock must
176 : // fail rather than hold the boot path.
177 : //
178 : // `SET` is session-scoped and the pool has no release hook, so the
179 : // connection is DETACHED first: it never goes back into the pool, and no
180 : // request is ever served by a connection with the runaway-query wall
181 : // switched off. The pool opens a fresh one (running `after_connect`) on
182 : // demand.
183 355 : if migrate_enabled(std::env::var("ANTARES_MIGRATE").ok().as_deref()) {
184 : use sqlx::{Connection, Executor};
185 355 : let mut migrate_conn = pool.acquire().await?.detach();
186 355 : migrate_conn
187 355 : .execute(sqlx::raw_sql("SET statement_timeout = 0"))
188 355 : .await?;
189 355 : migrate(&mut migrate_conn).await?;
190 352 : migrate_conn.close().await?;
191 0 : }
192 352 : Ok(pool)
193 421 : }
194 :
195 : /// The write reserve (`ANTARES_PG_WRITE_RESERVE`): a second pool of
196 : /// `max_connections` with the same session settings and bounds as
197 : /// [`connect_with`], which only the batch writes draw from, so reads queued
198 : /// on the main pool cannot hold a pipeline's batch upsert behind them. It
199 : /// runs no migrations: [`connect_with`] already did, on the main pool.
200 1 : pub async fn connect_reserve(
201 1 : url: &str,
202 1 : max_connections: u32,
203 1 : statement_timeout: std::time::Duration,
204 1 : ) -> Result<PgPool, sqlx::Error> {
205 1 : options(max_connections, statement_timeout)
206 1 : .connect(url)
207 1 : .await
208 1 : }
209 :
210 : /// The bounds and session settings every broker pool shares.
211 422 : fn options(max_connections: u32, statement_timeout: std::time::Duration) -> PgPoolOptions {
212 : use std::time::Duration;
213 : // No parallel workers: a request reads a page, and a page whose filter
214 : // the planner thinks selective (a `q` under RLS is never an index
215 : // condition) got a Gather of two workers, two processes started per
216 : // statement. On the distinct-1000 rig (2 PostgreSQL cores, 1 000 req/s)
217 : // that took a `q` page from 2.5 ms to 106 ms. No JIT either: the
218 : // per-instance expiry pruning each `q` term carries puts a page of seven
219 : // Entities past the JIT cost thresholds, and LLVM spent 13.8 s
220 : // compiling 607 functions for a query that ran in 1 ms.
221 422 : let session = format!(
222 : "SET statement_timeout = '{}ms'; SET lock_timeout = '5s'; \
223 : SET max_parallel_workers_per_gather = 0; SET jit = off",
224 422 : statement_timeout.as_millis()
225 : );
226 422 : PgPoolOptions::new()
227 422 : .max_connections(max_connections)
228 422 : .acquire_timeout(Duration::from_secs(5))
229 422 : .idle_timeout(Duration::from_secs(600))
230 422 : .max_lifetime(Duration::from_secs(1800))
231 : // sqlx tests a released connection with a ping it never times out,
232 : // and holds its pool slot until the ping answers. Through the mesh
233 : // proxy, which accepts the bytes, a stream whose far side stalled
234 : // never answers: once each such stream held a slot for hours, until
235 : // all 20 were gone and every read answered 503. This ping is
236 : // bounded; one that fails or runs out drops the socket at once, and
237 : // the slot opens a fresh connection on demand. sqlx's own ping after
238 : // it costs one more round trip on a connection that just answered.
239 692279 : .after_release(|conn, _meta| {
240 692273 : Box::pin(async move {
241 : use sqlx::Connection;
242 692273 : match tokio::time::timeout(RELEASE_PING, conn.ping()).await {
243 692153 : Ok(Ok(())) => Ok(true),
244 0 : Ok(Err(e)) => Err(e),
245 2 : Err(_) => Err(sqlx::Error::PoolTimedOut),
246 : }
247 692155 : })
248 692273 : })
249 1236 : .after_connect(move |conn, _meta| {
250 1230 : let session = session.clone();
251 1230 : Box::pin(async move {
252 : use sqlx::Executor;
253 : // the only interpolated value is a validated integer of
254 : // milliseconds (parsed by the broker), never client text
255 1230 : conn.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(session)))
256 1230 : .await?;
257 1230 : Ok(())
258 1230 : })
259 1230 : })
260 422 : }
261 :
262 : /// Make RLS effective for this transaction — SET LOCAL only (transaction
263 : /// scoped), so a recycled pooled connection carries no tenant residue.
264 : /// Call first in EVERY transaction that touches tenant data.
265 657222 : pub async fn set_tenant(
266 657222 : tx: &mut Transaction<'_, Postgres>,
267 657222 : tenant: &TenantId,
268 657222 : ) -> Result<(), sqlx::Error> {
269 657222 : sqlx::query(crate::SET_TENANT_SQL)
270 657222 : .bind(tenant.as_str())
271 657222 : .execute(&mut **tx)
272 657222 : .await
273 657221 : .map(|_| ())
274 657221 : }
275 :
276 : /// Does the connected role bypass RLS (superuser or BYPASSRLS)? RLS is
277 : /// a belt only when the role wears it — the broker warns at startup when the
278 : /// belt is off.
279 : ///
280 : /// A probe that cannot answer fails CLOSED: an unreachable or erroring
281 : /// database is reported as bypassing, so the strict gate refuses the boot
282 : /// instead of passing on an error.
283 7 : pub async fn role_bypasses_rls(pool: &PgPool) -> bool {
284 7 : bypasses(
285 7 : sqlx::query_scalar(
286 7 : "SELECT rolsuper OR rolbypassrls FROM pg_roles WHERE rolname = current_user",
287 7 : )
288 7 : .fetch_one(pool)
289 7 : .await,
290 : )
291 7 : }
292 :
293 : /// What the connected server is, read ONCE at startup and served from
294 : /// `/q/health` afterwards: server version plus the version of each extension
295 : /// the broker's behaviour depends on. A probe that fails is not fatal — the
296 : /// broker serves fine without knowing its own server version — so the caller
297 : /// gets an empty object and health simply says nothing.
298 4 : pub async fn version_info(pool: &PgPool) -> Value {
299 : // The pool's own shape is what an operator sizes against, and it is
300 : // known without asking the server: how many connections this process may
301 : // hold, and how long a request waits for one before the broker answers
302 : // 503. Both are reported even when the version probe fails.
303 4 : let mut m = serde_json::Map::new();
304 4 : m.insert("engine".into(), "postgres".into());
305 4 : m.insert(
306 4 : "poolSize".into(),
307 4 : pool.options().get_max_connections().into(),
308 : );
309 4 : m.insert(
310 4 : "poolAcquireTimeoutSeconds".into(),
311 4 : pool.options().get_acquire_timeout().as_secs().into(),
312 : );
313 4 : let row: Result<(String, Option<String>, Option<String>), sqlx::Error> = sqlx::query_as(
314 4 : "SELECT current_setting('server_version'), \
315 4 : (SELECT extversion FROM pg_extension WHERE extname = 'postgis'), \
316 4 : (SELECT extversion FROM pg_extension WHERE extname = 'timescaledb')",
317 4 : )
318 4 : .fetch_one(pool)
319 4 : .await;
320 4 : match row {
321 4 : Ok((server, postgis, timescale)) => {
322 4 : m.insert("server".into(), server.into());
323 4 : if let Some(v) = postgis {
324 4 : m.insert("postgis".into(), v.into());
325 4 : }
326 4 : if let Some(v) = timescale {
327 2 : m.insert("timescaledb".into(), v.into());
328 4 : }
329 : }
330 0 : Err(e) => {
331 0 : tracing::warn!("server version probe failed ({e}); /q/health will not report it");
332 : }
333 : }
334 4 : Value::Object(m)
335 4 : }
336 :
337 : /// Open a transaction, timing what it cost to get one. `Pool::begin` is the
338 : /// only way to a `Transaction<'static>` (sqlx owns the pooled connection
339 : /// inside it), so the measurement covers the pool wait plus one BEGIN round
340 : /// trip; the round trip is sub-millisecond and the wait is what grows, up to
341 : /// the acquire timeout. Every transactional store call goes through here, so
342 : /// `antares_pg_transaction_begin_seconds` is where pool pressure shows.
343 645654 : pub(crate) async fn begin(
344 645654 : pool: &PgPool,
345 645654 : ) -> Result<sqlx::Transaction<'static, sqlx::Postgres>, sqlx::Error> {
346 645654 : let t0 = std::time::Instant::now();
347 645654 : let tx = pool.begin().await;
348 645647 : metrics::histogram!("antares_pg_transaction_begin_seconds").record(t0.elapsed().as_secs_f64());
349 645647 : tx
350 645647 : }
351 :
352 15 : fn bypasses(probe: Result<bool, sqlx::Error>) -> bool {
353 15 : probe.unwrap_or_else(|e| {
354 5 : tracing::error!(
355 : "row-level-security probe failed ({e}) — treating the role as RLS-bypassing"
356 : );
357 5 : true
358 5 : })
359 15 : }
360 :
361 : /// Arm the transaction-scoped `antares.service` escape (0001_init.sql) for
362 : /// the two internal cross-tenant jobs: outbox drain and temporal retention.
363 : /// NEVER call from a request path — request queries carry explicit tenant
364 : /// predicates and run under `set_tenant` only.
365 26342 : pub async fn set_service(tx: &mut Transaction<'_, Postgres>) -> Result<(), sqlx::Error> {
366 26342 : sqlx::query("SELECT set_config('antares.service', 'on', true)")
367 26342 : .execute(&mut **tx)
368 26342 : .await
369 26342 : .map(|_| ())
370 26342 : }
371 :
372 : /// The write's own claim on its tenant (ADR-0001: "every implicit tenant
373 : /// creation inserts its row in the same transaction as the document"). Two
374 : /// things ride on it being IN the document's transaction rather than before
375 : /// it: the tenant row and the rows it accounts for commit together, and the
376 : /// claim holds a row lock the purge's `SELECT … FOR UPDATE` conflicts with,
377 : /// so the purge waits for an in-flight write instead of stepping over it.
378 : /// Committed separately, a write racing a purge leaves rows behind for a
379 : /// tenant the inventory no longer names — readable to whoever sends that
380 : /// tenant header, listed by nothing, and reclaimable by no further purge.
381 : ///
382 : /// The lock is SHARED. An upsert whose `DO UPDATE` re-set the tenant id took
383 : /// the exclusive lock and wrote a row version on every write, so writes in
384 : /// one tenant serialised behind each other and a table with one row per
385 : /// tenant collected one dead tuple per write in the broker. `FOR SHARE`
386 : /// conflicts with `FOR UPDATE` exactly as the exclusive lock did, and is
387 : /// compatible with itself, so concurrent writers hold it together.
388 : ///
389 : /// Idempotent, so two concurrent first writes both succeed (vs Scorpio's
390 : /// CREATE DATABASE + Flyway deadlock).
391 36917 : pub async fn claim_tenant(
392 36917 : tx: &mut Transaction<'_, Postgres>,
393 36917 : tenant: &TenantId,
394 36917 : ) -> Result<(), sqlx::Error> {
395 : // ponytail: shared row lock, so many concurrent writers in one tenant
396 : // register on one row; if multixact contention on that row ever shows up
397 : // in `pg_stat_activity`, the upgrade is a shared advisory lock keyed by
398 : // the tenant, which the purge takes exclusively and which touches no heap
399 : // page at all.
400 36917 : let claimed =
401 36917 : sqlx::query_scalar::<_, i32>("SELECT 1 FROM tenants WHERE tenant_id = $1 FOR SHARE")
402 36917 : .bind(tenant.as_str())
403 36917 : .fetch_optional(&mut **tx)
404 36917 : .await?
405 36917 : .is_some();
406 36917 : if claimed {
407 36837 : return Ok(());
408 80 : }
409 : // First write for this tenant. `DO UPDATE` rather than `DO NOTHING`:
410 : // a concurrent first write must block here until the winner commits,
411 : // and `DO NOTHING` would return without taking its lock.
412 80 : sqlx::query(
413 80 : "INSERT INTO tenants (tenant_id) VALUES ($1) \
414 80 : ON CONFLICT (tenant_id) DO UPDATE SET tenant_id = EXCLUDED.tenant_id",
415 80 : )
416 80 : .bind(tenant.as_str())
417 80 : .execute(&mut **tx)
418 80 : .await
419 80 : .map(|_| ())
420 36917 : }
421 :
422 : /// The same claim outside a document transaction, for the callers that have
423 : /// no document to pair it with (test seeds, the maintenance paths).
424 245 : pub async fn ensure_tenant(pool: &PgPool, tenant: &TenantId) -> Result<(), sqlx::Error> {
425 245 : sqlx::query("INSERT INTO tenants (tenant_id) VALUES ($1) ON CONFLICT DO NOTHING")
426 245 : .bind(tenant.as_str())
427 245 : .execute(pool)
428 245 : .await
429 245 : .map(|_| ())
430 245 : }
431 :
432 : #[cfg(test)]
433 : mod tests {
434 : use super::{
435 : bypasses, first_unknown_older, is_schema_mismatch, migrate_enabled, tolerate_newer,
436 : };
437 :
438 : /// the lenient run of a rollback ignores missing versions,
439 : /// so an applied version at or below the newest that this binary lacks
440 : /// is found before it runs.
441 : #[test]
442 2 : fn a_rollback_still_refuses_an_unknown_older_version() {
443 2 : assert_eq!(first_unknown_older(&[1, 2, 3], &[1, 2, 3]), None);
444 2 : assert_eq!(first_unknown_older(&[1, 5, 2], &[1, 2, 3]), Some(5));
445 2 : assert_eq!(first_unknown_older(&[], &[1]), None);
446 2 : }
447 :
448 : /// The RLS probe answers a security gate: a probe that errors must read as
449 : /// "unsafe", never as "the role does not bypass RLS".
450 : #[test]
451 2 : fn rls_probe_fails_closed() {
452 2 : assert!(!bypasses(Ok(false)), "a role without BYPASSRLS is safe");
453 2 : assert!(bypasses(Ok(true)), "superuser/BYPASSRLS bypasses");
454 2 : assert!(
455 2 : bypasses(Err(sqlx::Error::PoolTimedOut)),
456 : "an unanswerable probe must not pass the gate"
457 : );
458 2 : assert!(bypasses(Err(sqlx::Error::RowNotFound)));
459 2 : }
460 :
461 : /// A database from another release is a permanent failure; a database
462 : /// that is not up yet is not. The boot path retries the second and dies on
463 : /// the first, so the classification decides which cause an operator is
464 : /// shown — the squash of the pre-1.0 migration set left a 0.1.0 database
465 : /// reporting "not reachable after 30 s" for a schema it would never match.
466 : #[test]
467 2 : fn a_history_from_another_release_is_permanent_and_a_booting_database_is_not() {
468 : use sqlx::migrate::MigrateError;
469 12 : for e in [
470 2 : MigrateError::VersionMissing(5),
471 2 : MigrateError::VersionMismatch(1),
472 2 : MigrateError::VersionNotPresent(9),
473 2 : MigrateError::VersionTooOld(1, 4),
474 2 : MigrateError::VersionTooNew(9, 4),
475 2 : MigrateError::Dirty(3),
476 2 : ] {
477 12 : let shown = e.to_string();
478 12 : assert!(
479 12 : is_schema_mismatch(&sqlx::Error::Migrate(Box::new(e))),
480 : "{shown} can never be resolved by waiting"
481 : );
482 : }
483 2 : assert!(
484 2 : !is_schema_mismatch(&sqlx::Error::PoolTimedOut),
485 : "a database still booting is worth another attempt"
486 : );
487 2 : assert!(!is_schema_mismatch(&sqlx::Error::RowNotFound));
488 2 : assert!(
489 2 : !is_schema_mismatch(&sqlx::Error::Migrate(Box::new(MigrateError::Execute(
490 2 : sqlx::Error::PoolTimedOut
491 2 : )))),
492 : "a migration that failed to execute may be a lock the next attempt gets"
493 : );
494 2 : }
495 :
496 : /// A rollback boots on a newer release's compatible migrations and stops
497 : /// on a breaking one, naming its version.
498 : #[test]
499 2 : fn a_rollback_tolerates_compatible_migrations_and_stops_on_a_breaking_one() {
500 : use sqlx::migrate::MigrateError;
501 2 : assert_eq!(tolerate_newer(16, &[]).ok(), Some(false));
502 2 : let compatible = [(17, "type counts".to_owned())];
503 2 : assert_eq!(tolerate_newer(16, &compatible).ok(), Some(true));
504 2 : let breaking = [
505 2 : (17, "type counts".to_owned()),
506 2 : (18, "breaking drop names".to_owned()),
507 2 : ];
508 2 : let refused = tolerate_newer(16, &breaking).err().map(|e| e.to_string());
509 2 : assert_eq!(
510 : refused,
511 2 : Some(MigrateError::VersionTooNew(18, 16).to_string())
512 : );
513 2 : let named = [(17, "not breaking yet".to_owned())];
514 2 : assert_eq!(tolerate_newer(16, &named).ok(), Some(true));
515 2 : }
516 :
517 : /// Migrations stay on unless a deployment explicitly turns them off.
518 : #[test]
519 2 : fn migrate_switch_defaults_on() {
520 2 : assert!(migrate_enabled(None));
521 2 : assert!(migrate_enabled(Some("1")));
522 2 : assert!(migrate_enabled(Some("true")));
523 2 : assert!(!migrate_enabled(Some("0")));
524 2 : assert!(!migrate_enabled(Some("false")));
525 2 : }
526 : }
|
