diff --git a/CHANGELOG.md b/CHANGELOG.md index 3bcc51e0..1b1902d8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ - **`current_kv` C-index warning codes.** The log sidecar maps the three constant warnings of the TC-1274 `current_kv` C-index migration to `current_kv_c_index_missing`, `current_kv_c_index_invalid` and `current_kv_c_index_wrong_shape` (TC-1274); until then they would ship as `unclassified`. This is part of the same regenerated Vector configuration. - **Ingress.** The `/admin/stats` label takes effect once the ingress image is rebuilt from `deploy/ingress` and its digest is pinned in the compose; until then the deployed ingress logs it as ``. - Add `GET /admin/stats` (TC-1482), a read-only list of the lowercase wallet addresses that hold data on the node, for the scoreboard's protocol-user count. It is disabled (`404`, like an unknown route) unless the new `NODE_STATS_READ_TOKEN` is set to a value of at least 32 bytes (`openssl rand -hex 32`); a shorter value, or one equal to `TINYCLOUD_ADMIN_SECRET`, also disables it. The token is separate from `TINYCLOUD_ADMIN_SECRET` (neither opens the other's routes). An owner counts when they hold live KV in any space except `account`, or database artifacts in any space except `account` and `secrets`, except KV in the `public` space whose key starts with `.well-known/vault-` (vault metadata). Automatic sign-in records are excluded: the `account` space, the `secrets` table OpenKey's bootstrap creates, and the vault metadata `vault.unlock()` writes. App first-use databases (for example listen's `conversations`) count. So an owner who only signed in, or only signed in and unlocked the vault, is not listed. One SQL statement however many spaces exist. A result past 200,000 owners (or 400,000 distinct counted spaces) is a bare `507`, never a partial `200`. Neither the token nor any address is logged. The deployment workflow and compose files do not pass the variable yet. +- New migration `m20261020_000000_current_kv_key_c_index` adds `idx_current_kv_space_deleted_key_c ON current_kv (space, deleted, (key COLLATE "C"))` on PostgreSQL (TC-1274). The KV list range select compares `key COLLATE "C"`, which the existing index, built in the column's own collation, cannot seek on under a non-`C` database collation such as production's `en_US.UTF-8`; it read and filtered every live key of the space. No query changed: the planner matches the new index. **The index is performance-only: the node boots and serves without it, so the migration never locks `current_kv` for a build and never fails the boot over the index.** It does nothing when a valid index of that exact shape exists; builds it when `current_kv` is empty (a fresh database); and, when it is missing on a non-empty `current_kv`, builds nothing, logs one `current_kv_c_index_missing` warning with constant text and keeps serving (KV lists stay slower), and is recorded as applied, so an index built later is used without a restart. An index of that name that is invalid, not ready or not live logs `current_kv_c_index_invalid`; one of another shape logs `current_kv_c_index_wrong_shape` at error level (drop it with `DROP INDEX CONCURRENTLY`); neither stops the boot. **Operator step before deploy** (PostgreSQL), to avoid the slow lists: build the index with `CREATE INDEX CONCURRENTLY IF NOT EXISTS` outside any transaction and check `pg_index.indisvalid`, `indisready` and `indislive`; the exact statements and recovery are in `docs/releasing.md`. SQLite and MySQL: no change. Rollback: `DROP INDEX CONCURRENTLY IF EXISTS idx_current_kv_space_deleted_key_c`, and, for a binary rollback, `DELETE FROM seaql_migrations WHERE version = 'm20261020_000000_current_kv_key_c_index'` first (an older node refuses to start with a migration it does not ship). +- The SQL/DuckDB legacy-artifact fence scan is reused for 60 s instead of 1 s while the durable fence is off (TC-1274). It was an anti-join over every artifact row, re-run by the first SQL/DuckDB request each second. The durable and configured fences are still read on every request, so `fence on` is immediate; a stale answer can only delay a 503 (or lift one) by at most 60 s, and only around an artifact written by an older release or a restore. The argument is on `LEGACY_SCAN_TTL` in `database_migration.rs`. ## [1.21.2] - 2026-10-19 diff --git a/Cargo.lock b/Cargo.lock index 8bea3ae1..29d59d23 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9668,6 +9668,7 @@ dependencies = [ "tinycloud-auth", "tokio", "tracing", + "tracing-subscriber", "ucan-capabilities-object", "unicode-normalization", "uuid", diff --git a/docs/releasing.md b/docs/releasing.md index 562cafcc..658edb1f 100644 --- a/docs/releasing.md +++ b/docs/releasing.md @@ -57,6 +57,97 @@ a version into the `docker.yml` `image_version` input that `Cargo.toml` does not already claim. Both have happened (`0b48da2`, and the image tagged `1.10.0` built from a `1.9.0` tree); the `version-guard` job now blocks the second one. +## Operator steps before a node deploy + +A migration that indexes a large PostgreSQL table cannot be built by the boot: +the boot runs every pending migration in one transaction, so a plain +`CREATE INDEX` holds a SHARE lock on the table, blocking its writes, for the +whole build. For these, the operator builds the index online first. **Order: +operator step, then the node deploy.** + +An index that only speeds up reads is **performance-only**: the node boots and +serves without it, so its migration never builds it on a non-empty table and +never fails the boot over it. It logs a typed warning and keeps serving; an +index built later is picked up by the planner with no restart. + +### `idx_current_kv_space_deleted_key_c` (TC-1274, `m20261020_000000_current_kv_key_c_index`) + +Ships in the release after 1.21.2. It lets the KV list range select seek on +its byte-order key bounds instead of reading and filtering every live key of +the space. **Performance-only:** the node boots and serves without the index, +and KV lists are slower until it exists. When it is missing on a non-empty +`current_kv`, the node logs `current_kv_c_index_missing` (WARN) and keeps +serving. + +1. On the PlanetScale primary, **outside any transaction** (the console, or + `psql` without `BEGIN`; never through the node or a migration runner): + + ```sql + CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_current_kv_space_deleted_key_c ON current_kv (space, deleted, (key COLLATE "C")); + ``` + + Writes to `current_kv` continue during the build. It waits for transactions + that were open when it started, so it can sit in "waiting for old + transactions" while one is open; it does not block them. Build time grows + with the row count of `current_kv`. Measured locally on an unthrottled + PostgreSQL 16: 3.9 s for 1,000,000 rows (112 MB heap, 109 MB index). A + CPU-limited primary is slower; the index is about the size of the existing + `idx_current_kv_space_deleted_key`. + +2. Check the result. All three columns must be `t`: + + ```sql + SELECT i.indisvalid, i.indisready, i.indislive FROM pg_index i JOIN pg_class c ON c.oid = i.indexrelid WHERE c.relname = 'idx_current_kv_space_deleted_key_c'; + ``` + +3. If the build was cancelled or failed, the index is left INVALID, and + `IF NOT EXISTS` would skip it. Drop it and redo step 1: + + ```sql + DROP INDEX CONCURRENTLY idx_current_kv_space_deleted_key_c; + ``` + +4. Deploy the node. At boot the migration reads `pg_index` for that exact name + and **never takes a lock on `current_kv` for a build and never fails the + boot over the index**: + - a valid, ready, live btree of that name on `current_kv` with the shape + above (columns `space, deleted, key`, the third in collation + `pg_catalog.C`, the first two in their own columns' collations, default + operator classes, ascending, no predicate, + expression or INCLUDE): **no DDL, no lock, no warning**, and the index + keeps its OID; + - no such index and `current_kv` **empty** (a fresh or test database): the + migration builds it, which is instant; + - no such index and `current_kv` **not empty**: nothing is built. The node + logs one `current_kv_c_index_missing` warning, boots and serves with slower + KV lists, and the migration is recorded as applied. Run step 1 whenever + convenient; the planner uses the index as soon as it is valid. A restart + is not needed; + - the right shape but INVALID, not ready or not live (a failed or still + running `CREATE INDEX CONCURRENTLY`): one `current_kv_c_index_invalid` + warning. The index is left alone. If no build is running, follow step 3 + and redo step 1; + - the name with another shape (for example the existing + `(space, deleted, key)` shape, `COLLATE "POSIX"`, DESC, another operator + class, INCLUDE, a predicate, another table, or a non-index relation): one + `current_kv_c_index_wrong_shape` error-level log. The object is left + alone, the node serves, and lists are slower. Run + `DROP INDEX CONCURRENTLY idx_current_kv_space_deleted_key_c;` and then + step 1. + + The log text is constant (no key, space or value; production logs are + public, TC-961). The log shipper maps the `warn`/`error` record to + `node_warning` / `node_error` with an unclassified code, so look for the + event name in the node's own container log. + +Rollback: `DROP INDEX CONCURRENTLY IF EXISTS idx_current_kv_space_deleted_key_c;` +is safe at any time; the list range select goes back to filtering the space, +and nothing else uses the index. A node image from before this migration +refuses to start while `seaql_migrations` lists a migration it does not ship, so +a binary rollback also needs +`DELETE FROM seaql_migrations WHERE version = 'm20261020_000000_current_kv_key_c_index';` +first. The migration's `down` keeps the index. + ## Promoting to production (main-only) Production is built and promoted **only from `main`** (TC-940, 2026-10-13). diff --git a/tinycloud-core/Cargo.toml b/tinycloud-core/Cargo.toml index 70b1a742..330ab898 100644 --- a/tinycloud-core/Cargo.toml +++ b/tinycloud-core/Cargo.toml @@ -85,5 +85,6 @@ unicode-normalization = "0.1" sea-orm = { version = "1.1", features = ["runtime-tokio-rustls", "sqlx-sqlite"] } async-std = { version = "1", features = ["attributes"] } tokio = { workspace = true, features = ["test-util"] } +tracing-subscriber = "0.3" iri-string.workspace = true proptest = "1" diff --git a/tinycloud-core/src/database_migration.rs b/tinycloud-core/src/database_migration.rs index 9896aa4e..d5bc3814 100644 --- a/tinycloud-core/src/database_migration.rs +++ b/tinycloud-core/src/database_migration.rs @@ -438,6 +438,47 @@ async fn has_unmigrated_artifacts_exact( .any(|row| !is_digest_name(&row.2) && !registered.contains(&row))) } +/// How long `EffectiveFenceCache` reuses a legacy-artifact scan (TC-1274). +/// +/// This was 1 s, so while the durable fence was off every SQL/DuckDB request +/// more than 1 s after the last one re-ran `has_unmigrated_artifacts`, an +/// anti-join over every artifact row (~12% of the database's query CPU). +/// +/// Why 60 s of staleness is safe: +/// - **What the boolean gates.** `true` (a SQL/DuckDB artifact with a +/// non-digest name and no `database_legacy_artifact` row) makes every +/// SQL/DuckDB `/invoke`, and the share-email SQL store, answer 503 before +/// any database is touched. `false` lets the request on to `resolve`, which +/// reads `database_legacy_artifact` and `database_alias` live on every +/// request, quarantines a registered legacy name or routes it through its +/// alias. The cached value is only a tripwire for artifacts nobody has +/// inventoried yet. +/// - **The operator's fence is never cached.** The configured fence and the +/// durable `database_identity_fence` row are read on every call before the +/// cache. `fence on` takes effect at once, and `apply`, `set_alias` and +/// `clear_alias` all require that row, so every migration step runs behind a +/// live fence. `set_fence(false)` runs its own uncached scan. +/// - **Stale `false` (the unsafe direction) needs a writer this binary does +/// not have.** Artifact names come from `logical_name` (an N2 digest) or +/// from an alias, which `resolve` returns only when its legacy row is +/// registered; the legacy selector exists for N3's inventory only. So the +/// answer turns from `false` to `true` only if an older release or a restore +/// writes an unregistered legacy artifact. For up to 60 s requests for that +/// path then resolve to its digest name (an empty database) instead of +/// 503. That is the 1 s window made longer, and it overwrites nothing. A +/// digest artifact created in the window makes the later `apply` fail with +/// `Collision` for the operator to resolve. +/// - **Stale `true` is fail-closed.** After the last legacy artifact is +/// registered, requests keep getting a retryable 503 for at most 60 s, and +/// only while the durable fence is off. +/// - **Restarts and instances.** The cache lives in the process +/// (`SpaceDatabase`) and starts empty, so a restarted node scans on its +/// first request. A migration run or the `tinycloud-sql-identity` CLI is +/// another process and does not invalidate it; nothing above needs that. +/// Nodes sharing one database each hold their own entry, each at most 60 s +/// stale. Entries are aged from the start of their scan. +pub const LEGACY_SCAN_TTL: Duration = Duration::from_secs(60); + /// Caches only the expensive legacy-artifact inventory. The durable metadata /// fence is read on every check, so turning it on takes effect even while an /// earlier "unfenced" inventory result is cached. @@ -451,7 +492,7 @@ impl Default for EffectiveFenceCache { fn default() -> Self { Self { legacy: tokio::sync::Mutex::new(None), - ttl: Duration::from_secs(1), + ttl: LEGACY_SCAN_TTL, } } } @@ -1761,6 +1802,46 @@ mod tests { conn.close().await.unwrap(); } + /// The scan is reused for `LEGACY_SCAN_TTL`, measured on the monotonic + /// clock from the start of the scan, and re-run after it. The entry's + /// timestamp is moved instead of sleeping the TTL. + #[tokio::test] + async fn effective_fence_reuses_a_legacy_scan_for_the_ttl_and_rescans_after_it() { + use sea_orm::Database; + use sea_orm_migration::MigratorTrait; + + let conn = Database::connect("sqlite::memory:").await.unwrap(); + crate::migrations::Migrator::up(&conn, None).await.unwrap(); + let cache = EffectiveFenceCache::default(); + let aged = |age: Duration| Instant::now().checked_sub(age).expect("monotonic clock"); + + assert!(!cache.check(&conn, false).await.unwrap()); + artifact("sql", "test-space", "threads") + .insert(&conn) + .await + .unwrap(); + + cache.legacy.lock().await.as_mut().unwrap().0 = + aged(LEGACY_SCAN_TTL - Duration::from_secs(5)); + assert!( + !cache.check(&conn, false).await.unwrap(), + "a scan younger than the TTL is reused" + ); + + cache.legacy.lock().await.as_mut().unwrap().0 = + aged(LEGACY_SCAN_TTL + Duration::from_secs(1)); + assert!( + cache.check(&conn, false).await.unwrap(), + "a scan older than the TTL is re-run and sees the new artifact" + ); + let (checked_at, fenced) = cache.legacy.lock().await.unwrap(); + assert!(fenced); + assert!( + checked_at.elapsed() < Duration::from_secs(5), + "the re-run is timestamped now" + ); + } + #[tokio::test] async fn effective_fence_caches_legacy_scan_but_observes_metadata_fence_live() { use sea_orm::Database; diff --git a/tinycloud-core/src/db.rs b/tinycloud-core/src/db.rs index 22f61792..0c366811 100644 --- a/tinycloud-core/src/db.rs +++ b/tinycloud-core/src/db.rs @@ -9313,6 +9313,163 @@ mod test { exercise.expect("TC-731 PostgreSQL collation resilience"); } + /// `EXPLAIN ANALYZE` of the `list_bounded_after` range select for + /// `prefix`, one line per plan row. + async fn explain_kv_list( + conn: &C, + space: &SpaceId, + prefix: &str, + ) -> String { + let prefix: Path = prefix.parse().unwrap(); + let list = kv_list_statement(conn.get_database_backend(), space, &prefix, None, None); + let explain = Statement::from_sql_and_values( + conn.get_database_backend(), + match conn.get_database_backend() { + DbBackend::Postgres => { + format!( + "EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF) {}", + list.sql + ) + } + _ => format!("EXPLAIN QUERY PLAN {}", list.sql), + }, + list.values.expect("list statement is parameterised").0, + ); + conn.query_all(explain) + .await + .unwrap() + .into_iter() + .map(|row| { + row.try_get_by_index::(0) + .or_else(|_| row.try_get::("", "detail")) + .unwrap() + }) + .collect::>() + .join("\n") + } + + /// TC-1274: the range select seeks on its byte-order key bounds. + /// + /// `kv_prefix_condition` compares `key COLLATE "C"`, and + /// `idx_current_kv_space_deleted_key` is built in the column's own + /// collation. Under a linguistic collation (production's `en_US.UTF-8`; + /// here ICU `en`, which CI's C-collated PG16 would otherwise hide) the old + /// index can only seek on `(space, deleted)` and filter the rest of the + /// space's live keys. `idx_current_kv_space_deleted_key_c` makes the + /// bounds index conditions, with no query change. + #[tokio::test] + async fn postgres_kv_list_range_seeks_on_the_c_collation_index() { + let Some(fixture) = crate::test_support::PostgresTestSchema::new( + "postgres_kv_list_range_seeks_on_the_c_collation_index", + ) + .await + else { + return; + }; + let conn = &fixture.conn; + conn.execute(Statement::from_string( + DbBackend::Postgres, + "CREATE COLLATION tc1274_linguistic (provider = icu, locale = 'en')".to_string(), + )) + .await + .unwrap(); + conn.execute(Statement::from_string( + DbBackend::Postgres, + "ALTER TABLE current_kv ALTER COLUMN key TYPE character varying \ + COLLATE tc1274_linguistic" + .to_string(), + )) + .await + .unwrap(); + + let wanted: Vec = (0..20).map(|i| format!("docs/{i:03}")).collect(); + let others: Vec = (0..300).map(|i| format!("other/{i:03}")).collect(); + let keys: Vec<&str> = wanted.iter().chain(&others).map(String::as_str).collect(); + let space = seed_kv_keys(conn, "pg-range-index", &keys).await; + conn.execute(Statement::from_string( + DbBackend::Postgres, + "ANALYZE current_kv".to_string(), + )) + .await + .unwrap(); + + // Planner inputs the test controls: a table this small would be + // seq-scanned, and the test is about which index serves the range. + // Both runs are rolled back. + let plan = |index_present: bool, prefix: &'static str| { + let conn = conn.clone(); + let space = space.clone(); + async move { + let tx = conn.begin().await.unwrap(); + tx.execute_unprepared("SET LOCAL enable_seqscan = off") + .await + .unwrap(); + if !index_present { + tx.execute_unprepared("DROP INDEX idx_current_kv_space_deleted_key_c") + .await + .unwrap(); + } + let plan = explain_kv_list(&tx, &space, prefix).await; + tx.rollback().await.unwrap(); + plan + } + }; + for prefix in ["docs/", "docs"] { + let before = plan(false, prefix).await; + println!("TC-1274 plan without the C index, prefix {prefix:?}:\n{before}"); + assert!( + before.contains("Rows Removed by Filter: 300"), + "without the C index the whole space is filtered: {before}" + ); + + let after = plan(true, prefix).await; + println!("TC-1274 plan with the C index, prefix {prefix:?}:\n{after}"); + assert!( + after.contains("idx_current_kv_space_deleted_key_c"), + "{prefix:?}: {after}" + ); + assert!( + after + .lines() + .any(|line| line.contains("Index Cond") && line.contains("< 'docs0'")), + "{prefix:?}: the key bounds are Index Cond: {after}" + ); + assert!( + !after.contains("Rows Removed by Filter"), + "{prefix:?}: nothing is read and discarded: {after}" + ); + assert!(!after.contains("Seq Scan"), "{prefix:?}: {after}"); + } + + // Same answers either way. + let listed = kv_list_pages(conn, &space, "docs", 100).await; + assert_eq!(listed, vec![wanted]); + } + + /// TC-1274: SQLite needs no `COLLATE` index. Its `BINARY` bounds already + /// seek on `idx_current_kv_space_deleted_key` for a prefix ending in `/`, + /// so the migration is a no-op there. (A bare prefix is an `OR` with the + /// exact key, which SQLite does not seek; that is unchanged here.) + #[tokio::test] + async fn sqlite_kv_list_range_seeks_on_the_existing_index() { + let db = get_db().await.unwrap(); + let space = seed_kv_keys( + &db.conn, + "sqlite-range-index", + &["docs/a", "docs/b", "other/a"], + ) + .await; + + let plan = explain_kv_list(&db.conn, &space, "docs/").await; + println!("TC-1274 SQLite plan:\n{plan}"); + assert!( + plan.contains( + "idx_current_kv_space_deleted_key (space=? AND deleted=? AND key>? AND key, ) { diff --git a/tinycloud-core/src/epoch_head.rs b/tinycloud-core/src/epoch_head.rs index 43ccb566..dbb3e2f7 100644 --- a/tinycloud-core/src/epoch_head.rs +++ b/tinycloud-core/src/epoch_head.rs @@ -230,7 +230,7 @@ mod tests { use crate::migrations::Migrator; use crate::models::space; use crate::storage::memory::MemoryStore; - use crate::test_support::PostgresTestSchema; + use crate::test_support::{PostgresTestDatabase, PostgresTestSchema}; use crate::SpaceDatabase; use sea_orm::{DatabaseConnection, TransactionTrait}; use sea_orm_migration::MigratorTrait; @@ -601,7 +601,8 @@ mod tests { .position(|migration| migration.name() == "m20261007_010000_database_identity_fence") .unwrap() as u32 + 1; - let Some(fixture) = PostgresTestSchema::migrated_through(test, Some(v1_20_6)).await else { + let Some(fixture) = PostgresTestDatabase::migrated_through(test, Some(v1_20_6)).await + else { return; }; let conn = &fixture.conn; @@ -636,7 +637,13 @@ mod tests { migrate(&conn.clone().into()).await.unwrap(); - let db = postgres_db(&fixture).await; + let db: TestDb = SpaceDatabase::new( + fixture.connect(15).await, + MemoryStore::default(), + StaticSecret::new([0u8; 32].to_vec()).unwrap(), + ) + .await + .unwrap(); assert!(db.repair_epoch_heads(&linear.space).await.unwrap()); assert_eq!( cached(conn, &linear.space).await, diff --git a/tinycloud-core/src/migrations/m20261017_000000_epoch_order_space_index.rs b/tinycloud-core/src/migrations/m20261017_000000_epoch_order_space_index.rs index 443b4feb..07894f7a 100644 --- a/tinycloud-core/src/migrations/m20261017_000000_epoch_order_space_index.rs +++ b/tinycloud-core/src/migrations/m20261017_000000_epoch_order_space_index.rs @@ -156,7 +156,7 @@ mod tests { use crate::hash::{hash, Hash}; use crate::migrations::Migrator; use crate::models::{epoch, space}; - use crate::test_support::PostgresTestSchema; + use crate::test_support::{PostgresTestDatabase, PostgresTestSchema}; use crate::types::SpaceIdWrap; use sea_orm::{ ConnectionTrait, Database, DbBackend, DbConn, EntityTrait, PaginatorTrait, Statement, @@ -405,7 +405,7 @@ mod tests { /// until every migration has run. #[tokio::test] async fn postgres_operator_built_index_is_kept_without_blocking_writers() { - let Some(fixture) = PostgresTestSchema::migrated_through( + let Some(fixture) = PostgresTestDatabase::migrated_through( "postgres_operator_built_index_is_kept_without_blocking_writers", Some(v1_20_6_migrations()), ) @@ -516,7 +516,7 @@ mod tests { let test = "postgres_index_build_in_progress_says_wait"; for restricted in [false, true] { let Some(fixture) = - PostgresTestSchema::migrated_through(test, Some(steps_before_this())).await + PostgresTestDatabase::migrated_through(test, Some(steps_before_this())).await else { return; }; @@ -531,7 +531,7 @@ mod tests { .unwrap() .try_get("", "s") .unwrap(); - let role = format!("{schema}_node"); + let role = format!("{}_node", fixture.name()); let writer = db.begin().await.unwrap(); writer diff --git a/tinycloud-core/src/migrations/m20261020_000000_current_kv_key_c_index.rs b/tinycloud-core/src/migrations/m20261020_000000_current_kv_key_c_index.rs new file mode 100644 index 00000000..db6e3034 --- /dev/null +++ b/tinycloud-core/src/migrations/m20261020_000000_current_kv_key_c_index.rs @@ -0,0 +1,898 @@ +//! TC-1274: `current_kv (space, deleted, (key COLLATE "C"))` for the KV list +//! range select. `kv_prefix_condition` compares `key COLLATE "C"` so prefix +//! membership is byte order whatever the column's collation is, and +//! `idx_current_kv_space_deleted_key` is built in the column's own collation. +//! On a database whose collation is not `C` (production's is `en_US.UTF-8`) +//! the key bounds therefore cannot be index conditions: the scan reads every +//! live key of the space and filters. This index matches the `COLLATE "C"` +//! expression, so the planner seeks on the bounds. The old index stays for +//! point lookups. No query changes: the planner matches the expression. +//! +//! **The index is performance-only.** The node boots and serves without it +//! (KV lists are slower), so this migration never locks `current_kv` for a +//! build and never fails boot because the index is missing or broken. On +//! PostgreSQL, `up` reads `pg_index` for the index's exact name and: +//! +//! 1. valid, ready and live, with the right shape: does nothing; +//! 2. missing and `current_kv` empty (a fresh or test database): creates it, +//! which is instant on an empty table; +//! 3. missing and `current_kv` not empty: does **not** create it (a plain +//! `CREATE INDEX` takes a SHARE lock that blocks every KV write until the +//! boot transaction ends) and logs `current_kv_c_index_missing`; +//! 4. right shape but invalid, not ready or not live (a failed or running +//! `CREATE INDEX CONCURRENTLY`): logs `current_kv_c_index_invalid`; +//! 5. this name with another shape (or not an index on `current_kv`): logs +//! `current_kv_c_index_wrong_shape` at error level. +//! +//! In every case the migration is recorded as applied, so an index an operator +//! builds later is simply used by the planner. Operators build it online +//! before deploy, outside any transaction, with the statement and recovery +//! steps in `docs/releasing.md`. The warnings are constant text: production +//! logs are public (TC-961), so no key, space or value is ever logged. +//! +//! SQLite and MySQL: nothing to do. SQLite's `BINARY` collation is byte order, +//! so `idx_current_kv_space_deleted_key` already seeks on the bounds. MySQL has +//! no `COLLATE "C"` to match; the migration leaves it as it is. + +use sea_orm::{ConnectionTrait, DbBackend, Statement}; +use sea_orm_migration::prelude::*; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +const INDEX: &str = "idx_current_kv_space_deleted_key_c"; +const DEFINITION: &str = r#"(space, deleted, (key COLLATE "C"))"#; + +/// Longest the build on an empty table waits for a lock. The migration +/// transaction's own `lock_timeout` is 30 s; a `CREATE INDEX` that waits queues +/// every later KV writer behind it, so this is kept short and a timeout skips +/// the index instead of failing boot. +const CREATE_LOCK_TIMEOUT: &str = "2s"; + +const EVENT_MISSING: &str = "current_kv_c_index_missing"; +const EVENT_INVALID: &str = "current_kv_c_index_invalid"; +const EVENT_WRONG_SHAPE: &str = "current_kv_c_index_wrong_shape"; + +const MISSING_MESSAGE: &str = "idx_current_kv_space_deleted_key_c is missing; the node keeps \ +serving and KV list requests are slower until an operator builds it with CREATE INDEX CONCURRENTLY \ +(docs/releasing.md, section idx_current_kv_space_deleted_key_c)"; +const INVALID_MESSAGE: &str = "idx_current_kv_space_deleted_key_c is invalid or still being \ +built; the node keeps serving and KV list requests are slower. If no CREATE INDEX CONCURRENTLY is \ +running, run DROP INDEX CONCURRENTLY idx_current_kv_space_deleted_key_c; and then the CREATE INDEX \ +CONCURRENTLY in docs/releasing.md, section idx_current_kv_space_deleted_key_c"; +const WRONG_SHAPE_MESSAGE: &str = "idx_current_kv_space_deleted_key_c exists with the wrong \ +definition and is not used for KV lists; the node keeps serving and KV list requests are slower. \ +Run DROP INDEX CONCURRENTLY idx_current_kv_space_deleted_key_c; and then the CREATE INDEX \ +CONCURRENTLY in docs/releasing.md, section idx_current_kv_space_deleted_key_c"; + +/// The DDL `up` runs when the index is absent and `current_kv` is empty +/// (PostgreSQL only). +fn create_index_sql() -> String { + format!("CREATE INDEX IF NOT EXISTS {INDEX} ON current_kv {DEFINITION}") +} + +/// The statement an operator runs, in autocommit, before the node deploy. +#[cfg(test)] +fn operator_create_index_sql() -> String { + format!("CREATE INDEX CONCURRENTLY IF NOT EXISTS {INDEX} ON current_kv {DEFINITION}") +} + +/// What the current schema holds under [`INDEX`]'s name. +#[derive(Debug, PartialEq, Eq)] +enum IndexState { + /// No relation has that name. + Missing, + /// A valid, ready and live plain btree index of the right shape. + Valid, + /// The right shape, but not valid, ready and live. + Invalid, + /// Anything else with that name. + WrongShape, +} + +/// Classify [`INDEX`] on PostgreSQL. +/// +/// PostgreSQL stores `(key COLLATE "C")` on a bare column not as an index +/// expression but as the column `key` with the `C` collation of `pg_catalog` +/// recorded in `indcollation`, so that is the shape checked: columns +/// `space, deleted, key`, the third in collation `pg_catalog.C`, default +/// operator classes, ascending, the first two columns in their own columns' +/// collations (a query compares `space` in the column's collation, so an +/// index in another one cannot serve `space = $1`), no expressions, predicate or included +/// columns. An index on `key` in the database collation (the existing +/// `idx_current_kv_space_deleted_key`), on `COLLATE "POSIX"`, on another +/// expression, with a predicate, or with extra columns is not this index, and +/// the planner cannot use it for the byte-order bounds. +async fn postgres_index_state(manager: &SchemaManager<'_>) -> Result { + let row = manager + .get_connection() + .query_one(Statement::from_sql_and_values( + DbBackend::Postgres, + "SELECT c.relkind::text AS kind, + COALESCE(i.indrelid = 'current_kv'::regclass, false) AS on_table, + COALESCE(i.indisvalid AND i.indisready AND i.indislive, false) AS usable, + am.amname::text AS method, + i.indpred IS NULL AND i.indexprs IS NULL + AND i.indnatts = i.indnkeyatts AS plain, + array_to_string(ARRAY( + SELECT a.attname::text + FROM unnest(i.indkey::int2[]) WITH ORDINALITY AS k(attnum, ord) + LEFT JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = k.attnum + ORDER BY k.ord), ',') AS columns, + (SELECT n.nspname::text || '.' || co.collname::text + FROM unnest(i.indcollation::oid[]) WITH ORDINALITY AS k(coll, ord) + JOIN pg_collation co ON co.oid = k.coll + JOIN pg_namespace n ON n.oid = co.collnamespace + WHERE k.ord = 3) AS key_collation, + COALESCE((SELECT bool_and(oc.opcdefault) + FROM unnest(i.indclass::oid[]) AS k(opclass) + JOIN pg_opclass oc ON oc.oid = k.opclass), false) AS default_opclasses, + array_to_string(i.indoption::int2[], ',') AS options, + COALESCE((SELECT bool_and(k.coll = a.attcollation) + FROM unnest(i.indcollation::oid[]) WITH ORDINALITY AS k(coll, ord) + JOIN unnest(i.indkey::int2[]) WITH ORDINALITY AS kk(attnum, ord) + ON kk.ord = k.ord + JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = kk.attnum + WHERE k.ord IN (1, 2)), false) AS lead_collations_match_the_columns + FROM pg_class c + LEFT JOIN pg_index i ON i.indexrelid = c.oid + LEFT JOIN pg_am am ON am.oid = c.relam + WHERE c.relname = $1 + AND c.relnamespace = (SELECT relnamespace FROM pg_class + WHERE oid = 'current_kv'::regclass)", + [INDEX.into()], + )) + .await?; + let Some(row) = row else { + return Ok(IndexState::Missing); + }; + let kind: String = row.try_get("", "kind")?; + let on_table: bool = row.try_get("", "on_table")?; + let usable: bool = row.try_get("", "usable")?; + let method: Option = row.try_get("", "method")?; + let plain: Option = row.try_get("", "plain")?; + let columns: Option = row.try_get("", "columns")?; + let key_collation: Option = row.try_get("", "key_collation")?; + let default_opclasses: bool = row.try_get("", "default_opclasses")?; + let options: Option = row.try_get("", "options")?; + let lead_collations_match_the_columns: bool = + row.try_get("", "lead_collations_match_the_columns")?; + let shaped = kind == "i" + && on_table + && method.as_deref() == Some("btree") + && plain == Some(true) + && columns.as_deref() == Some("space,deleted,key") + && key_collation.as_deref() == Some("pg_catalog.C") + && default_opclasses + && options.as_deref() == Some("0,0,0") + && lead_collations_match_the_columns; + Ok(match (shaped, usable) { + (true, true) => IndexState::Valid, + (true, false) => IndexState::Invalid, + (false, _) => IndexState::WrongShape, + }) +} + +/// Whether `current_kv` has any row. An `EXISTS` probe stops at the first +/// row and takes only an ACCESS SHARE lock, which no writer conflicts with. +async fn current_kv_is_empty(manager: &SchemaManager<'_>) -> Result { + let row = manager + .get_connection() + .query_one(Statement::from_string( + DbBackend::Postgres, + "SELECT EXISTS (SELECT 1 FROM current_kv) AS present".to_string(), + )) + .await? + .ok_or_else(|| DbErr::Migration("current_kv probe returned no row".to_string()))?; + Ok(!row.try_get::("", "present")?) +} + +/// Build the index on the empty `current_kv`. The build runs in a savepoint +/// with a short `lock_timeout`, so a writer that appears between the probe and +/// the build, or a role that may not create indexes on the table, leaves the +/// transaction usable: `Ok(false)` means "skipped", never a failed boot. +async fn try_create_on_empty_table(manager: &SchemaManager<'_>) -> Result { + const SAVEPOINT: &str = "current_kv_c_index"; + let conn = manager.get_connection(); + let outer_timeout: String = conn + .query_one(Statement::from_string( + DbBackend::Postgres, + "SELECT current_setting('lock_timeout') AS value".to_string(), + )) + .await? + .ok_or_else(|| DbErr::Migration("lock_timeout probe returned no row".to_string()))? + .try_get("", "value")?; + conn.execute_unprepared(&format!("SAVEPOINT {SAVEPOINT}")) + .await?; + let built = async { + conn.execute_unprepared(&format!("SET LOCAL lock_timeout = '{CREATE_LOCK_TIMEOUT}'")) + .await?; + conn.execute_unprepared(&create_index_sql()).await + } + .await; + if built.is_err() { + conn.execute_unprepared(&format!("ROLLBACK TO SAVEPOINT {SAVEPOINT}")) + .await?; + } + conn.execute_unprepared(&format!("RELEASE SAVEPOINT {SAVEPOINT}")) + .await?; + if built.is_ok() { + // `SET LOCAL` outlives the released savepoint: give the migration + // transaction its own timeout back. + conn.execute(Statement::from_sql_and_values( + DbBackend::Postgres, + "SELECT set_config('lock_timeout', $1, true)", + [outer_timeout.into()], + )) + .await?; + } + Ok(built.is_ok()) +} + +/// Why the index was not built: constant values only. +const REASON_TABLE_NOT_EMPTY: &str = "table_not_empty"; +const REASON_BUILD_FAILED: &str = "build_failed"; + +fn warn_missing(reason: &'static str) { + tracing::warn!( + event = EVENT_MISSING, + index = INDEX, + reason, + "{MISSING_MESSAGE}" + ); +} + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + if manager.get_database_backend() != DbBackend::Postgres { + return Ok(()); + } + match postgres_index_state(manager).await? { + IndexState::Valid => {} + IndexState::Invalid => { + tracing::warn!(event = EVENT_INVALID, index = INDEX, "{INVALID_MESSAGE}"); + } + IndexState::WrongShape => { + tracing::error!( + event = EVENT_WRONG_SHAPE, + index = INDEX, + "{WRONG_SHAPE_MESSAGE}" + ); + } + IndexState::Missing => { + if !current_kv_is_empty(manager).await? { + warn_missing(REASON_TABLE_NOT_EMPTY); + } else if !try_create_on_empty_table(manager).await? { + warn_missing(REASON_BUILD_FAILED); + } + } + } + Ok(()) + } + + /// Keeps the index. It may be operator-built, and the previous release + /// runs the same range select, so it still benefits. Drop it by hand with + /// `DROP INDEX CONCURRENTLY` if it is ever unwanted. + async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> { + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::migrations::Migrator; + use crate::test_support::events::{capture_events, CapturedEvent}; + use crate::test_support::PostgresTestDatabase; + use sea_orm::{ConnectionTrait, Database, DbBackend, DbConn, Statement, TransactionTrait}; + use sea_orm_migration::MigratorTrait; + use std::time::Duration; + + /// A key the logs must never show. + const KEY_CANARY: &str = "secret-key-canary"; + + fn steps_before_this() -> u32 { + Migrator::migrations() + .iter() + .position(|migration| migration.name() == Migration.name()) + .unwrap() as u32 + } + + /// A database as of the previous release, in a database of its own: some + /// tests here run `CREATE INDEX CONCURRENTLY`, which waits for every open + /// transaction of its database (`PostgresTestDatabase`). + async fn before_this(test: &str) -> Option { + PostgresTestDatabase::migrated_through(test, Some(steps_before_this())).await + } + + #[tokio::test] + async fn sqlite_migration_is_a_no_op() { + let db = Database::connect("sqlite::memory:").await.unwrap(); + Migrator::up(&db, None).await.unwrap(); + let names: Vec = db + .query_all(Statement::from_string( + DbBackend::Sqlite, + "PRAGMA index_list('current_kv')".to_string(), + )) + .await + .unwrap() + .into_iter() + .map(|row| row.try_get("", "name").unwrap()) + .collect(); + assert!(names.contains(&"idx_current_kv_space_deleted_key".to_string())); + assert!(!names.iter().any(|name| name == INDEX), "{names:?}"); + } + + /// `(oid, indisvalid)` of `INDEX` on `current_kv`, or `None` if absent. + async fn postgres_index(db: &DbConn) -> Option<(u32, bool)> { + let row = db + .query_one(Statement::from_string( + DbBackend::Postgres, + format!( + "SELECT c.oid::int8 AS oid, i.indisvalid AS valid + FROM pg_index i JOIN pg_class c ON c.oid = i.indexrelid + WHERE i.indrelid = 'current_kv'::regclass AND c.relname = '{INDEX}'" + ), + )) + .await + .unwrap()?; + Some(( + row.try_get::("", "oid").unwrap() as u32, + row.try_get("", "valid").unwrap(), + )) + } + + /// The production entry point: one outer transaction (TC-1054). + async fn boot(db: &DbConn) -> Result<(), DbErr> { + crate::db::migrate(&db.clone().into()).await + } + + /// `boot`, giving up after `limit`. The timer starts when the future is + /// first polled: pass this to `events_of`, which polls it only once it + /// holds the capture lock, so waiting for another test's capture does not + /// count against `limit`. + async fn boot_within( + db: &DbConn, + limit: Duration, + ) -> Result, tokio::time::error::Elapsed> { + tokio::time::timeout(limit, boot(db)).await + } + + /// `up` alone, in a transaction under the boot's 30 s `lock_timeout`. + /// Returns the `lock_timeout` in force when `up` returned. + async fn run_up(db: &DbConn) -> String { + let transaction = db.begin().await.unwrap(); + transaction + .execute_unprepared("SET LOCAL lock_timeout = '30s'") + .await + .unwrap(); + Migration + .up(&SchemaManager::new(&transaction)) + .await + .unwrap(); + let timeout = transaction + .query_one(Statement::from_string( + DbBackend::Postgres, + "SELECT current_setting('lock_timeout') AS value".to_string(), + )) + .await + .unwrap() + .unwrap() + .try_get("", "value") + .unwrap(); + transaction.commit().await.unwrap(); + timeout + } + + async fn events_of(run: F) -> (F::Output, Vec) { + capture_events("current_kv_c_index_", run).await + } + + async fn recorded(db: &DbConn) -> i64 { + db.query_one(Statement::from_string( + DbBackend::Postgres, + format!( + "SELECT COUNT(*) AS n FROM seaql_migrations WHERE version = '{}'", + Migration.name() + ), + )) + .await + .unwrap() + .unwrap() + .try_get::("", "n") + .unwrap() + } + + /// One live key, so `current_kv` is not empty. Foreign keys are off for + /// the table: the rows they point at are not what is tested. + async fn insert_a_key(db: &DbConn) { + db.execute_unprepared("ALTER TABLE current_kv DISABLE TRIGGER ALL") + .await + .unwrap(); + db.execute_unprepared(&format!( + "INSERT INTO current_kv (space, key, invocation, seq, epoch, epoch_seq, value, \ + metadata, deleted) VALUES ('tinycloud:key:z:default', '{KEY_CANARY}', '\\x1e20', 1, \ + '\\x1e20', 0, '\\x1e20', '{{}}', false)" + )) + .await + .unwrap(); + } + + /// A transaction mid-write on `current_kv`, as `transact` is when it + /// upserts. `ROW EXCLUSIVE` conflicts with the `SHARE` lock a `CREATE + /// INDEX` takes, so anything that tried to build would queue behind it. + async fn open_writer(db: &DbConn) -> sea_orm::DatabaseTransaction { + let writer = db.begin().await.unwrap(); + writer + .execute_unprepared("LOCK TABLE current_kv IN ROW EXCLUSIVE MODE") + .await + .unwrap(); + writer + } + + /// The one `missing` warning, with the constant reason it was not built. + fn assert_missing(events: &[CapturedEvent], reason: &str, label: &str) { + assert_only( + events, + EVENT_MISSING, + tracing::Level::WARN, + MISSING_MESSAGE, + label, + ); + assert!( + events[0].fields.contains(&format!("reason=\"{reason}\"")), + "{label}: {:?}", + events[0] + ); + } + + fn assert_only( + events: &[CapturedEvent], + name: &str, + level: tracing::Level, + message: &str, + label: &str, + ) { + assert_eq!(events.len(), 1, "{label}: {events:?}"); + let event = &events[0]; + assert_eq!(event.event, name, "{label}"); + assert_eq!(event.level, level, "{label}"); + assert_eq!(event.message, message, "{label}"); + assert!( + !event.message.contains(KEY_CANARY) && !event.fields.contains(KEY_CANARY), + "{label}: {event:?}" + ); + } + + /// Branch 2 (fresh database): the index is built, instantly, silently. + #[tokio::test] + async fn postgres_fresh_database_gets_the_index() { + let Some(fixture) = + PostgresTestDatabase::migrated_through("postgres_fresh_database_gets_the_index", None) + .await + else { + return; + }; + assert!(postgres_index(&fixture.conn).await.unwrap().1); + } + + /// Branch 2: missing and `current_kv` empty. Created; no warning; the + /// boot transaction's own `lock_timeout` is back afterwards. + #[tokio::test] + async fn postgres_missing_index_on_an_empty_table_is_created() { + let Some(fixture) = + before_this("postgres_missing_index_on_an_empty_table_is_created").await + else { + return; + }; + let db = &fixture.conn; + assert_eq!(postgres_index(db).await, None); + + let (timeout, events) = events_of(run_up(db)).await; + + assert!(postgres_index(db).await.unwrap().1); + assert!(events.is_empty(), "{events:?}"); + assert_eq!(timeout, "30s"); + } + + /// Branch 2, but a writer holds `current_kv` between the probe and the + /// build (or the role may not build): the build gives up after + /// `CREATE_LOCK_TIMEOUT`, the transaction stays usable, the node boots + /// with one `missing` warning and the migration is recorded. + #[tokio::test] + async fn postgres_failed_build_on_an_empty_table_warns_and_boots() { + let Some(fixture) = + before_this("postgres_failed_build_on_an_empty_table_warns_and_boots").await + else { + return; + }; + let db = &fixture.conn; + let writer = open_writer(db).await; + let migrator = fixture.connect(1).await; + + // `up` alone: the transaction is usable after the failed build and + // has the boot's `lock_timeout` back. + let ((timeout, up_waited), events) = events_of(async { + let started = std::time::Instant::now(); + (run_up(&migrator).await, started.elapsed()) + }) + .await; + assert_eq!(timeout, "30s"); + assert!(up_waited < Duration::from_secs(10), "{up_waited:?}"); + assert_eq!(postgres_index(db).await, None); + assert_missing(&events, REASON_BUILD_FAILED, "up"); + + // The real boot: continues, records the migration. + let (booted, events) = events_of(boot_within(&migrator, Duration::from_secs(20))).await; + writer.rollback().await.unwrap(); + booted.expect("boot hung on the build").unwrap(); + assert_eq!(postgres_index(db).await, None); + assert_eq!(recorded(db).await, 1); + assert_missing(&events, REASON_BUILD_FAILED, "boot"); + } + + /// Branch 3: missing and `current_kv` not empty. No `CREATE INDEX`, so no + /// lock on `current_kv` (a writer mid-transaction does not hold the boot + /// up), no index, one `missing` warning with constant text, the node + /// boots and the migration is recorded. The operator's build afterwards + /// is a valid index of the right shape, and a restart is quiet. + #[tokio::test] + async fn postgres_missing_index_on_a_non_empty_table_is_skipped_with_a_warning() { + let Some(fixture) = + before_this("postgres_missing_index_on_a_non_empty_table_is_skipped_with_a_warning") + .await + else { + return; + }; + let db = &fixture.conn; + insert_a_key(db).await; + let writer = open_writer(db).await; + let migrator = fixture.connect(1).await; + + let (booted, events) = events_of(boot_within(&migrator, Duration::from_secs(5))).await; + writer.rollback().await.unwrap(); + + booted + .expect("boot waited behind an open current_kv write: it took a lock") + .unwrap(); + assert_eq!(postgres_index(db).await, None, "the index was built"); + assert_eq!(recorded(db).await, 1); + assert_missing(&events, REASON_TABLE_NOT_EMPTY, "non-empty"); + + // Later, the operator builds it with the documented statement. + db.execute_unprepared(&operator_create_index_sql()) + .await + .unwrap(); + let (oid, valid) = postgres_index(db).await.unwrap(); + assert!(valid); + let (_, events) = events_of(async { + boot(db).await.unwrap(); + run_up(db).await; + }) + .await; + assert!(events.is_empty(), "{events:?}"); + assert_eq!(postgres_index(db).await, Some((oid, true))); + } + + /// Branch 1: the operator built the index online with the documented + /// statement. Booting keeps that index (same OID), says nothing, and does + /// not lock `current_kv` against writers. + #[tokio::test] + async fn postgres_valid_index_is_kept_without_a_warning_or_a_lock() { + let Some(fixture) = + before_this("postgres_valid_index_is_kept_without_a_warning_or_a_lock").await + else { + return; + }; + let db = &fixture.conn; + insert_a_key(db).await; + db.execute_unprepared(&operator_create_index_sql()) + .await + .unwrap(); + let (oid, valid) = postgres_index(db).await.unwrap(); + assert!(valid); + let writer = open_writer(db).await; + let migrator = fixture.connect(1).await; + + let (booted, events) = events_of(boot_within(&migrator, Duration::from_secs(5))).await; + writer.rollback().await.unwrap(); + + booted + .expect("boot waited behind an open current_kv write") + .unwrap(); + assert_eq!(postgres_index(db).await, Some((oid, true))); + assert_eq!(recorded(db).await, 1); + assert!(events.is_empty(), "{events:?}"); + } + + /// Branch 4: the right shape but not valid, ready and live (what a failed + /// `CREATE INDEX CONCURRENTLY` leaves). One `invalid` warning; the index + /// is neither dropped nor rebuilt, boot continues, the migration is + /// recorded. Empty or not, the table is not built on. + #[tokio::test] + async fn postgres_invalid_index_warns_and_boots() { + let Some(fixture) = before_this("postgres_invalid_index_warns_and_boots").await else { + return; + }; + let db = &fixture.conn; + for (flag, populated) in [ + ("indisvalid", false), + ("indisready", false), + ("indislive", false), + ("indisvalid", true), + ] { + let label = format!("{flag} populated {populated}"); + if populated { + insert_a_key(db).await; + } + db.execute_unprepared(&format!("DROP INDEX IF EXISTS {INDEX}")) + .await + .unwrap(); + db.execute_unprepared(&format!("CREATE INDEX {INDEX} ON current_kv {DEFINITION}")) + .await + .unwrap(); + // What a failed CONCURRENTLY build leaves behind. Needs a + // superuser, as the CI and local test roles are. + db.execute_unprepared(&format!( + "UPDATE pg_index SET {flag} = false WHERE indexrelid = '{INDEX}'::regclass" + )) + .await + .unwrap(); + let oid = postgres_index(db).await.unwrap().0; + + let (_, events) = events_of(run_up(db)).await; + + assert_only( + &events, + EVENT_INVALID, + tracing::Level::WARN, + INVALID_MESSAGE, + &label, + ); + assert_eq!(postgres_index(db).await.unwrap().0, oid, "{label}: rebuilt"); + } + + // Through the real boot, recorded as applied. + let (booted, events) = events_of(boot(db)).await; + booted.unwrap(); + assert_eq!(recorded(db).await, 1); + assert_eq!(events.len(), 1, "{events:?}"); + } + + /// Remove whatever holds `INDEX`'s name (an index or a table) and `other_kv`. + async fn clear_index_name(db: &DbConn) { + db.execute_unprepared(&format!( + "DO $$ + DECLARE kind \"char\"; + BEGIN + SELECT relkind INTO kind FROM pg_class + WHERE relname = '{INDEX}' + AND relnamespace = (SELECT oid FROM pg_namespace WHERE nspname = current_schema()); + IF kind = 'i' THEN DROP INDEX {INDEX}; ELSIF kind = 'r' THEN DROP TABLE {INDEX}; END IF; + DROP TABLE IF EXISTS other_kv; + END $$" + )) + .await + .unwrap(); + } + + /// Branch 4 as it happens: the operator's `CREATE INDEX CONCURRENTLY` is + /// still running (held in its first wait by an open write), so the index + /// is not valid yet. The node boots with one `invalid` warning, without + /// waiting for the build and without disturbing it; the build then + /// finishes valid. This database is this test's own: the build waits for + /// every open transaction of its database. + #[tokio::test] + async fn postgres_index_build_in_progress_warns_and_boots() { + let Some(fixture) = before_this("postgres_index_build_in_progress_warns_and_boots").await + else { + return; + }; + let db = &fixture.conn; + insert_a_key(db).await; + let writer = open_writer(db).await; + let builder = fixture.connect(1).await; + let build = tokio::spawn(async move { + builder + .execute_unprepared(&operator_create_index_sql()) + .await + }); + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + while postgres_index(db).await.is_none() { + assert!( + tokio::time::Instant::now() < deadline, + "build never started" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } + assert!(!postgres_index(db).await.unwrap().1, "mid-build is invalid"); + let migrator = fixture.connect(1).await; + + let (booted, events) = events_of(boot_within(&migrator, Duration::from_secs(5))).await; + writer.rollback().await.unwrap(); + build.await.unwrap().unwrap(); + + booted.expect("boot waited for the build").unwrap(); + assert_only( + &events, + EVENT_INVALID, + tracing::Level::WARN, + INVALID_MESSAGE, + "build in progress", + ); + assert_eq!(recorded(db).await, 1); + assert!(postgres_index(db).await.unwrap().1, "build finished valid"); + } + + /// Branch 5: this name, another definition (or another kind of relation, + /// or another table). One error-level `wrong_shape` event; the object is + /// left alone, boot continues. + #[tokio::test] + async fn postgres_wrong_shape_under_the_index_name_warns_loudly_and_boots() { + let Some(fixture) = + before_this("postgres_wrong_shape_under_the_index_name_warns_loudly_and_boots").await + else { + return; + }; + let db = &fixture.conn; + for (setup, populated) in [ + // The existing index's shape: the database collation, not "C". + ( + format!("CREATE INDEX {INDEX} ON current_kv (space, deleted, key)"), + false, + ), + ( + format!( + r#"CREATE INDEX {INDEX} ON current_kv (space, deleted, (key COLLATE "POSIX"))"# + ), + false, + ), + // A collation on a leading column other than the column's own: a + // query compares `space` in the column's collation, so the + // planner cannot match this index for `space = $1`. + ( + format!( + r#"CREATE INDEX {INDEX} ON current_kv ((space COLLATE "C"), deleted, (key COLLATE "C"))"# + ), + false, + ), + ( + format!( + r#"CREATE INDEX {INDEX} ON current_kv ((space COLLATE "POSIX"), deleted, (key COLLATE "C"))"# + ), + false, + ), + ( + format!( + r#"CREATE INDEX {INDEX} ON current_kv (space, deleted, (lower(key) COLLATE "C"))"# + ), + false, + ), + ( + format!(r#"CREATE INDEX {INDEX} ON current_kv (space, (key COLLATE "C"))"#), + false, + ), + ( + format!( + r#"CREATE INDEX {INDEX} ON current_kv (space, (key COLLATE "C"), deleted)"# + ), + false, + ), + ( + format!( + r#"CREATE INDEX {INDEX} ON current_kv (space, deleted, (key COLLATE "C") DESC)"# + ), + false, + ), + ( + format!( + r#"CREATE INDEX {INDEX} ON current_kv (space, deleted, (key COLLATE "C") varchar_pattern_ops)"# + ), + false, + ), + ( + format!( + r#"CREATE INDEX {INDEX} ON current_kv (space, deleted, (key COLLATE "C")) INCLUDE (seq)"# + ), + false, + ), + ( + format!( + r#"CREATE INDEX {INDEX} ON current_kv (space, deleted, (key COLLATE "C")) WHERE deleted = false"# + ), + false, + ), + // Non-empty table too: still no build. + ( + format!("CREATE INDEX {INDEX} ON current_kv (space, deleted, key)"), + true, + ), + // Another table's index of this name. + ( + format!( + r#"CREATE TABLE other_kv (space text, deleted boolean, key varchar); + CREATE INDEX {INDEX} ON other_kv (space, deleted, (key COLLATE "C"))"# + ), + false, + ), + // Not an index. + (format!("CREATE TABLE {INDEX} (id int)"), false), + ] { + if populated { + insert_a_key(db).await; + } + clear_index_name(db).await; + db.execute_unprepared(&setup).await.unwrap(); + + let (_, events) = events_of(run_up(db)).await; + + assert_only( + &events, + EVENT_WRONG_SHAPE, + tracing::Level::ERROR, + WRONG_SHAPE_MESSAGE, + &setup, + ); + } + + // Through the real boot, recorded as applied. + let (booted, events) = events_of(boot(db)).await; + booted.unwrap(); + assert_eq!(recorded(db).await, 1); + assert_eq!(events.len(), 1, "{events:?}"); + } + + /// A different name is a different index: `up` looks only at its own + /// name, whatever else is on `current_kv` (existing indexes of any shape, + /// even this very definition under another name). + #[tokio::test] + async fn postgres_other_indexes_are_ignored() { + let Some(fixture) = before_this("postgres_other_indexes_are_ignored").await else { + return; + }; + let db = &fixture.conn; + for statement in [ + "CREATE INDEX idx_tc1274_other_collation ON current_kv (space, deleted, key)", + r#"CREATE INDEX idx_tc1274_other_name ON current_kv (space, deleted, (key COLLATE "C"))"#, + ] { + db.execute_unprepared(statement).await.unwrap(); + } + let others = || async { + db.query_all(Statement::from_string( + DbBackend::Postgres, + "SELECT c.relname::text AS name, c.oid::int8 AS oid FROM pg_class c \ + WHERE c.relname LIKE 'idx_tc1274_other_%' ORDER BY 1" + .to_string(), + )) + .await + .unwrap() + .into_iter() + .map(|row| { + ( + row.try_get::("", "name").unwrap(), + row.try_get::("", "oid").unwrap(), + ) + }) + .collect::>() + }; + let before = others().await; + assert_eq!(before.len(), 2); + + // Empty table: this index is still built. + let (_, events) = events_of(run_up(db)).await; + assert!(postgres_index(db).await.unwrap().1); + assert!(events.is_empty(), "{events:?}"); + assert_eq!(others().await, before); + + // Non-empty table without it: still the `missing` warning. + db.execute_unprepared(&format!("DROP INDEX {INDEX}")) + .await + .unwrap(); + insert_a_key(db).await; + let (_, events) = events_of(run_up(db)).await; + assert_eq!(postgres_index(db).await, None); + assert_missing(&events, REASON_TABLE_NOT_EMPTY, "other indexes"); + assert_eq!(others().await, before); + } +} diff --git a/tinycloud-core/src/migrations/mod.rs b/tinycloud-core/src/migrations/mod.rs index 3d0618ba..634b25de 100644 --- a/tinycloud-core/src/migrations/mod.rs +++ b/tinycloud-core/src/migrations/mod.rs @@ -28,6 +28,7 @@ pub mod m20261014_000000_delegation_cid_alias; pub mod m20261015_000000_delegation_cursor_index; pub mod m20261017_000000_epoch_order_space_index; pub mod m20261018_000000_epoch_head; +pub mod m20261020_000000_current_kv_key_c_index; pub struct Migrator; @@ -64,6 +65,7 @@ impl MigratorTrait for Migrator { Box::new(m20261015_000000_delegation_cursor_index::Migration), Box::new(m20261017_000000_epoch_order_space_index::Migration), Box::new(m20261018_000000_epoch_head::Migration), + Box::new(m20261020_000000_current_kv_key_c_index::Migration), ] } } diff --git a/tinycloud-core/src/test_support.rs b/tinycloud-core/src/test_support.rs index 103e3304..4d4f2b15 100644 --- a/tinycloud-core/src/test_support.rs +++ b/tinycloud-core/src/test_support.rs @@ -32,6 +32,21 @@ pub(crate) fn postgres_test_url(test: &str) -> Option { } } +/// A name no other fixture of this run has: the clock alone is not enough, +/// two tests starting together can read the same instant (a duplicate +/// `CREATE DATABASE` / `CREATE SCHEMA` fails the test), so a process-wide +/// counter breaks the tie. +fn unique_name(prefix: &str) -> String { + use std::sync::atomic::{AtomicU64, Ordering}; + static NEXT: AtomicU64 = AtomicU64::new(0); + format!( + "{prefix}_{}_{}_{}", + std::process::id(), + time::OffsetDateTime::now_utc().unix_timestamp_nanos(), + NEXT.fetch_add(1, Ordering::Relaxed) + ) +} + /// A migrated connection to a throwaway schema on the /// `TINYCLOUD_TEST_POSTGRES_URL` database (`search_path` is that schema). The /// schema is dropped when this goes out of scope, including while a failed @@ -130,6 +145,212 @@ impl Drop for PostgresTestSchema { } } +/// A migrated connection to a throwaway DATABASE on the server behind +/// `TINYCLOUD_TEST_POSTGRES_URL`, dropped (with its sessions) when this goes +/// out of scope. +/// +/// Use this instead of [`PostgresTestSchema`] for tests that run +/// `CREATE INDEX CONCURRENTLY`, or that hold a transaction open while one +/// runs. That statement waits for every open transaction of its *database* +/// that predates its snapshots, whatever schema they work in: in the shared +/// test database it queues behind the other tests' long migration +/// transactions (and they behind it, under enough parallelism), which looks +/// like a hang. A database of its own can only wait on its own test. +pub(crate) struct PostgresTestDatabase { + pub(crate) conn: sea_orm::DatabaseConnection, + admin_url: String, + url: String, + name: String, +} + +impl PostgresTestDatabase { + /// Like [`PostgresTestSchema::migrated_through`], in a database of its own. + pub(crate) async fn migrated_through(test: &str, steps: Option) -> Option { + use sea_orm::{ConnectionTrait, Database, DbBackend, Statement}; + use sea_orm_migration::MigratorTrait; + + let admin_url = postgres_test_url(test)?; + let name = unique_name("tc_test_db"); + let admin = Database::connect(sea_orm::ConnectOptions::new(admin_url.clone())) + .await + .expect("connect to PostgreSQL test server"); + admin + .execute(Statement::from_string( + DbBackend::Postgres, + format!("CREATE DATABASE {name}"), + )) + .await + .expect("create isolated test database"); + admin.close().await.expect("close admin connection"); + let mut fixture = Self { + conn: Default::default(), + url: url_with_database(&admin_url, &name), + admin_url, + name, + }; + fixture.conn = fixture.connect(4).await; + crate::migrations::Migrator::up(&fixture.conn, steps) + .await + .expect("migrate isolated test database"); + Some(fixture) + } + + /// The database's name; unique per fixture, so usable in cluster-wide + /// names such as roles. + pub(crate) fn name(&self) -> &str { + &self.name + } + + /// Another pool of `max_connections` on the same isolated database. + pub(crate) async fn connect(&self, max_connections: u32) -> sea_orm::DatabaseConnection { + let mut options = sea_orm::ConnectOptions::new(self.url.clone()); + options.max_connections(max_connections).sqlx_logging(false); + sea_orm::Database::connect(options) + .await + .expect("connect to isolated test database") + } +} + +impl Drop for PostgresTestDatabase { + fn drop(&mut self) { + use sea_orm::{ConnectOptions, ConnectionTrait, Database, DbBackend, Statement}; + + let (url, name) = (self.admin_url.clone(), self.name.clone()); + let dropped = std::thread::spawn(move || { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("database drop runtime") + .block_on(async move { + Database::connect(ConnectOptions::new(url)) + .await? + .execute(Statement::from_string( + DbBackend::Postgres, + format!("DROP DATABASE IF EXISTS {name} WITH (FORCE)"), + )) + .await + .map(|_| ()) + }) + }) + .join(); + if !matches!(dropped, Ok(Ok(()))) { + eprintln!("failed to drop isolated test database {}", self.name); + } + } +} + +/// `url` with its database name replaced, keeping credentials and query. +fn url_with_database(url: &str, database: &str) -> String { + let authority_end = url + .find("://") + .and_then(|scheme| url[scheme + 3..].find('/').map(|slash| scheme + 3 + slash)) + .expect("TINYCLOUD_TEST_POSTGRES_URL names a database"); + let query = url[authority_end..] + .find('?') + .map_or("", |at| &url[authority_end + at..]); + format!("{}/{database}{query}", &url[..authority_end]) +} + +/// Typed log events captured from the `tracing` calls a test makes. +/// +/// A thread-scoped subscriber can miss events while other threads of this +/// parallel test process register the same callsites, so this installs one +/// process-wide subscriber (first use) that records the WARN and ERROR +/// events carrying an `event` field. Nothing else in the library's tests +/// installs a subscriber. [`capture_events`] serializes its callers, so a +/// test sees only the events of its own run. +pub(crate) mod events { + use std::sync::{Arc, Mutex, OnceLock}; + use tracing_subscriber::{filter::LevelFilter, layer::SubscriberExt, Layer}; + + #[derive(Clone, Debug)] + pub(crate) struct CapturedEvent { + pub(crate) level: tracing::Level, + /// The `event` field. + pub(crate) event: String, + /// The log message. + pub(crate) message: String, + /// Every other field, rendered, for tests that look for leaked data. + pub(crate) fields: String, + } + + #[derive(Default)] + struct Fields { + event: Option, + message: String, + others: String, + } + + impl tracing::field::Visit for Fields { + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + if field.name() == "event" { + self.event = Some(value.to_string()); + } else { + self.record_debug(field, &value); + } + } + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + if field.name() == "message" { + self.message = format!("{value:?}"); + } else { + self.others + .push_str(&format!("{}={value:?} ", field.name())); + } + } + } + + struct Capture(Arc>>); + + impl Layer for Capture { + fn on_event( + &self, + event: &tracing::Event<'_>, + _: tracing_subscriber::layer::Context<'_, S>, + ) { + let mut fields = Fields::default(); + event.record(&mut fields); + if let Some(name) = fields.event { + self.0.lock().unwrap().push(CapturedEvent { + level: *event.metadata().level(), + event: name, + message: fields.message, + fields: fields.others, + }); + } + } + } + + static STORE: OnceLock>>> = OnceLock::new(); + static ONE_CAPTURE_AT_A_TIME: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + + /// Run `run`, and return its output with the captured events whose + /// `event` starts with `prefix`, in order. + pub(crate) async fn capture_events( + prefix: &str, + run: F, + ) -> (F::Output, Vec) { + let _alone = ONE_CAPTURE_AT_A_TIME.lock().await; + let store = STORE.get_or_init(|| { + let store = Arc::new(Mutex::new(Vec::new())); + tracing::subscriber::set_global_default( + tracing_subscriber::registry() + .with(Capture(store.clone()).with_filter(LevelFilter::WARN)), + ) + .expect("another tracing subscriber is already the process default"); + store + }); + store.lock().unwrap().clear(); + let output = run.await; + let captured = store + .lock() + .unwrap() + .drain(..) + .filter(|event| event.event.starts_with(prefix)) + .collect(); + (output, captured) + } +} + pub(crate) mod did_resolution { include!("../tests/support/did_resolution.rs"); }