From 5b93913ae04e65349b9a8806f6bc0826d5cdaa4c Mon Sep 17 00:00:00 2001 From: Sam Gbafa Date: Thu, 15 Oct 2026 14:34:13 +0000 Subject: [PATCH 1/5] TC-737: seek quiet KV sync prefixes on PostgreSQL --- CHANGELOG.md | 1 + docs/kv-sync.md | 11 + tinycloud-core/src/db.rs | 21 +- tinycloud-core/src/kv_sync.rs | 254 ++++++++++++++ ...20261015_010000_current_kv_prefix_index.rs | 315 ++++++++++++++++++ tinycloud-core/src/migrations/mod.rs | 294 ++++++++++++++++ 6 files changed, 879 insertions(+), 17 deletions(-) create mode 100644 tinycloud-core/src/migrations/m20261015_010000_current_kv_prefix_index.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 18907086..ca23abd1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## [Unreleased] +- PostgreSQL `kv/sync` quiet-prefix polls can seek by byte-ordered `(space, key)` rather than scanning newer keys outside the prefix. The new index leaves empty-poll cursors and feed contents unchanged; SQLite already has the matching primary key (TC-737). - Account delegation session discovery keyset-pages delegation rows independently of ability fan-out, using a composite delegator/id index so large grant sets are not truncated (TC-1195). - `/attestation` quotes the SHA-256 of `key_binding` as REPORTDATA and returns it as `key_binding_sha256`; the quote no longer carries the raw binding. Before answering 200 the node checks that the returned TDX v4 quote's REPORTDATA field holds that digest (a structural check; it does not verify the quote). When the TEE can't produce a bound quote, the endpoint returns 503 instead of 200 `mode: "classic"`, and the dstack client reports a guest-agent error status instead of a parse error. `nonce` must be 1-64 characters of `A-Z a-z 0-9 - _` (400 otherwise), and the endpoint now shares the public routes' per-IP rate limit. Verifiers recompute the digest from their own nonce and expected DID and must still verify the quote; see `docs/attestation.md` (TC-963). diff --git a/docs/kv-sync.md b/docs/kv-sync.md index 7ed519bf..ee26687d 100644 --- a/docs/kv-sync.md +++ b/docs/kv-sync.md @@ -129,3 +129,14 @@ cursor. Read audit records every poll. Poll on demand, or no more often than every 60 seconds. + +On PostgreSQL, a `(space, key COLLATE "C")` index bounds a quiet-prefix poll +to keys inside that prefix even when other keys in the space keep changing. +The byte collation matches the prefix range predicate; the older feed-order +index still serves broad or active feeds. SQLite uses its existing +`(space, key)` primary key. This does not advance the cursor on an empty poll. + +MySQL does not get this index: its prefix predicate casts `key` to binary, +which does not seek the normal `(space, key)` index. MySQL's `kv/sync` +sequence lock is currently marked unsupported/untested for newly created +spaces. This change does not claim MySQL quiet-prefix performance. diff --git a/tinycloud-core/src/db.rs b/tinycloud-core/src/db.rs index d1e6d8c0..d4c59a6d 100644 --- a/tinycloud-core/src/db.rs +++ b/tinycloud-core/src/db.rs @@ -319,24 +319,11 @@ where } } -/// Apply the schema migrations on `pool`. -/// -/// On PostgreSQL sea-orm-migration runs them in one transaction; it runs -/// here inside an outer one that first lifts the pool's `statement_timeout` -/// (TC-1054), since a migration or its backfill may legitimately run for -/// minutes. The nested `BEGIN` becomes a savepoint, so the migrations stay -/// atomic as before. +/// Apply schema migrations on `pool`. The migrator keeps ordinary PostgreSQL +/// migrations transactional with an exempt statement timeout, but commits +/// before the concurrent current_kv index build and resumes afterward. pub async fn migrate(pool: &RecoveringConnection) -> Result<(), DbErr> { - let conn = pool.sea_orm_connection(); - if conn.get_database_backend() != sea_orm::DatabaseBackend::Postgres { - return Migrator::up(conn, None).await; - } - let transaction = conn.begin().await?; - transaction - .execute_unprepared(crate::db_pool::EXEMPT_STATEMENT_TIMEOUT) - .await?; - Migrator::up(&transaction, None).await?; - transaction.commit().await + Migrator::up(pool.sea_orm_connection(), None).await } impl SpaceDatabase diff --git a/tinycloud-core/src/kv_sync.rs b/tinycloud-core/src/kv_sync.rs index 891d308d..dfc74b9c 100644 --- a/tinycloud-core/src/kv_sync.rs +++ b/tinycloud-core/src/kv_sync.rs @@ -1252,6 +1252,57 @@ pub(crate) mod tests { ); } + #[tokio::test] + async fn kv_sync_quiet_prefix_does_not_expose_other_writes_in_cursor() { + let db = sqlite_db().await; + let owner = Owner::generate(); + host(&db, &owner).await; + run(&db, &owner, &[Op::Put("quiet/only", b"first")]).await; + let prefix = "quiet/".parse().unwrap(); + let mut request = KvSyncRequest { + limit: 10, + cursor: None, + cursor_key: [7; 32], + retention_grant: None, + }; + let (changes, more, cursor) = + kv_sync_read(db.connection(), &owner.space, &prefix, &request) + .await + .unwrap(); + assert_eq!(changes.len(), 1); + assert!(!more); + request.cursor = Some(cursor.clone()); + + for index in 0..20 { + run( + &db, + &owner, + &[Op::Put(&format!("busy/{index}"), b"outside")], + ) + .await; + } + let (changes, more, after_busy) = + kv_sync_read(db.connection(), &owner.space, &prefix, &request) + .await + .unwrap(); + assert!(changes.is_empty()); + assert!(!more); + assert_eq!( + after_busy, cursor, + "outside writes must not change the token" + ); + + run(&db, &owner, &[Op::Put("quiet/new", b"inside")]).await; + let (changes, more, advanced) = + kv_sync_read(db.connection(), &owner.space, &prefix, &request) + .await + .unwrap(); + assert_eq!(changes[0].key.as_str(), "quiet/new"); + assert_eq!(changes.len(), 1); + assert!(!more); + assert_ne!(advanced, cursor); + } + #[tokio::test] async fn kv_sync_bootstrap_skips_pre_floor_tombstones() { let db = sqlite_db().await; @@ -2679,6 +2730,209 @@ pub(crate) mod tests { (conn, locks) } + /// Run explicitly against PG18.6 (production engine): this fixture is + /// intentionally too large for the normal test suite. The one quiet key + /// predates 299,999 busy keys in its space; another space has 300,000 keys. + #[tokio::test] + #[ignore = "PG18.6 performance fixture: 2 spaces x 300k current_kv rows"] + async fn postgres_quiet_prefix_poll_seeks_by_key() { + let fixture = PostgresFixture::new("postgres_quiet_prefix_poll_seeks_by_key", 1) + .await + .expect("TINYCLOUD_TEST_POSTGRES_URL is required"); + let db = fixture.dbs[0].connection(); + let quiet = Owner::generate(); + let other = Owner::generate(); + host(&fixture.dbs[0], &quiet).await; + host(&fixture.dbs[0], &other).await; + let epoch = crate::hash::hash(b"tc737 benchmark epoch"); + for (space, with_quiet) in [(&quiet.space, true), (&other.space, false)] { + if with_quiet { + db.execute(Statement::from_sql_and_values( + DbBackend::Postgres, + "INSERT INTO current_kv (space, key, invocation, seq, epoch, epoch_seq, value, metadata, deleted) \ + VALUES ($1, 'quiet/only', $2, 10, $2, 0, $2, '{}', false)", + [SpaceIdWrap(space.clone()).into(), epoch.into()], + )) + .await + .unwrap(); + } + db.execute(Statement::from_sql_and_values( + DbBackend::Postgres, + "INSERT INTO current_kv (space, key, invocation, seq, epoch, epoch_seq, value, metadata, deleted) \ + SELECT $1, 'busy/' || lpad(n::text, 6, '0'), $2, n + 10, $2, 0, $2, '{}', false \ + FROM generate_series(1, $3::integer) AS n", + [ + SpaceIdWrap(space.clone()).into(), + epoch.into(), + (if with_quiet { 299_999 } else { 300_000 }).into(), + ], + )) + .await + .unwrap(); + } + db.execute(Statement::from_string( + DbBackend::Postgres, + "ANALYZE current_kv", + )) + .await + .unwrap(); + let statement = kv_sync_statement( + DbBackend::Postgres, + &quiet.space, + &"quiet/".parse().unwrap(), + 500, + Some(&KvSyncAnchor { + seq: 10, + epoch, + epoch_seq: 0, + }), + None, + ); + use sea_orm_migration::{MigrationTrait, SchemaManager}; + + let explain_sql = format!("EXPLAIN (ANALYZE, BUFFERS) {}", statement.sql); + let values = statement.values.map(|values| values.0).unwrap_or_default(); + db.execute_unprepared("DROP INDEX CONCURRENTLY idx_current_kv_space_key_c") + .await + .unwrap(); + let baseline = db + .query_all(Statement::from_sql_and_values( + DbBackend::Postgres, + explain_sql.clone(), + values.clone(), + )) + .await + .unwrap() + .into_iter() + .map(|row| row.try_get_by_index::(0).unwrap()) + .collect::>() + .join("\n"); + println!("TC-737 PostgreSQL pre-index quiet-prefix poll (2 x 300k):\n{baseline}"); + let build_db = db.clone(); + let build = tokio::spawn(async move { + let started = std::time::Instant::now(); + crate::migrations::m20261015_010000_current_kv_prefix_index::Migration + .up(&SchemaManager::new(&build_db)) + .await + .unwrap(); + started.elapsed() + }); + let active = tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + if build.is_finished() { + break false; + } + let progress = db + .query_one(Statement::from_string( + DbBackend::Postgres, + "SELECT 1 AS active FROM pg_stat_progress_create_index \ + WHERE relid = 'current_kv'::regclass", + )) + .await + .unwrap(); + if progress.is_some() { + break true; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!(active, "index build finished before lock sampling"); + let locks = db + .query_all(Statement::from_string( + DbBackend::Postgres, + "SELECT mode FROM pg_locks WHERE relation = 'current_kv'::regclass \ + AND pid IN (SELECT pid FROM pg_stat_progress_create_index \ + WHERE relid = 'current_kv'::regclass)", + )) + .await + .unwrap() + .into_iter() + .map(|row| row.try_get::("", "mode").unwrap()) + .collect::>(); + println!("TC-737 concurrent index table locks: {locks:?}"); + assert!( + !locks + .iter() + .any(|mode| mode == "ShareLock" || mode == "AccessExclusiveLock"), + "index build took a write-blocking lock: {locks:?}" + ); + let mut reads = 0; + let mut writes = 0; + let mut max_read = std::time::Duration::ZERO; + let mut max_write = std::time::Duration::ZERO; + while !build.is_finished() && writes < 10 { + let started = std::time::Instant::now(); + let row = tokio::time::timeout( + std::time::Duration::from_secs(2), + db.query_one(Statement::from_sql_and_values( + DbBackend::Postgres, + "SELECT key FROM current_kv WHERE space = $1 AND key = 'busy/000001'", + [SpaceIdWrap(quiet.space.clone()).into()], + )), + ) + .await + .unwrap() + .unwrap(); + assert!(row.is_some()); + max_read = max_read.max(started.elapsed()); + reads += 1; + let started = std::time::Instant::now(); + let result = tokio::time::timeout( + std::time::Duration::from_secs(2), + db.execute(Statement::from_sql_and_values( + DbBackend::Postgres, + "UPDATE current_kv SET metadata = '{}' WHERE space = $1 \ + AND key = 'busy/000001'", + [SpaceIdWrap(quiet.space.clone()).into()], + )), + ) + .await + .unwrap() + .unwrap(); + assert_eq!(result.rows_affected(), 1); + max_write = max_write.max(started.elapsed()); + writes += 1; + } + let build_time = build.await.unwrap(); + println!( + "TC-737 concurrent index build: {build_time:?}; \ + during build: {reads} reads (max {max_read:?}), \ + {writes} writes (max {max_write:?})" + ); + assert!( + reads > 0 && writes > 0, + "no read/write overlap with index build" + ); + let plan = db + .query_all(Statement::from_sql_and_values( + DbBackend::Postgres, + explain_sql, + values, + )) + .await + .unwrap() + .into_iter() + .map(|row| row.try_get_by_index::(0).unwrap()) + .collect::>() + .join("\n"); + println!("TC-737 PostgreSQL quiet-prefix poll (2 x 300k):\n{plan}"); + assert!( + plan.contains("idx_current_kv_space_key_c") && plan.contains("Index Cond:"), + "quiet-prefix poll must seek by key, not scan newer space changes:\n{plan}" + ); + let filtered = plan + .lines() + .filter_map(|line| line.split("Rows Removed by Filter: ").nth(1)) + .map(|count| count.trim().parse::().unwrap()) + .sum::(); + assert!( + filtered <= 10, + "quiet-prefix poll filtered {filtered} rows:\n{plan}" + ); + } + /// Keys the transaction already holds are not asked for again; the /// others are, and every one stays held until the transaction ends. #[tokio::test] diff --git a/tinycloud-core/src/migrations/m20261015_010000_current_kv_prefix_index.rs b/tinycloud-core/src/migrations/m20261015_010000_current_kv_prefix_index.rs new file mode 100644 index 00000000..130c5ea2 --- /dev/null +++ b/tinycloud-core/src/migrations/m20261015_010000_current_kv_prefix_index.rs @@ -0,0 +1,315 @@ +//! Byte-ordered key range seek for quiet-prefix kv/sync polls on PostgreSQL. +//! SQLite's `(space, key)` primary key already has byte ordering. + +use sea_orm_migration::prelude::*; +#[cfg(feature = "postgres")] +use sea_orm_migration::SchemaManagerConnection; + +pub(crate) const NAME: &str = "m20261015_010000_current_kv_prefix_index"; + +/// A dedicated connection holds the session advisory lock and has no +/// statement timeout. Try the lock without waiting in a PostgreSQL query: +/// CONCURRENTLY waits for old transactions, including a blocked lock query, +/// so a blocking advisory lock can deadlock two simultaneous node startups. +/// Closing the connection releases the lock and discards session settings. +#[cfg(feature = "postgres")] +async fn postgres_index(manager: &SchemaManager<'_>, create: bool) -> Result<(), DbErr> { + let SchemaManagerConnection::Connection(db) = manager.get_connection() else { + return Err(DbErr::Custom( + "concurrent current_kv index requires a connection outside a transaction".into(), + )); + }; + let mut connection = db + .get_postgres_connection_pool() + .acquire() + .await + .map_err(|error| DbErr::Conn(sea_orm::RuntimeErr::SqlxError(error)))?; + let result = async { + sea_orm::sqlx::query("SET statement_timeout = 0") + .execute(&mut *connection) + .await?; + loop { + let locked: bool = sea_orm::sqlx::query_scalar("SELECT pg_try_advisory_lock(737, 1)") + .fetch_one(&mut *connection) + .await?; + if locked { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + if create { + let valid: Option = sea_orm::sqlx::query_scalar( + "SELECT indisvalid FROM pg_index \ + WHERE indexrelid = to_regclass('idx_current_kv_space_key_c')", + ) + .fetch_optional(&mut *connection) + .await?; + // A cancelled concurrent build leaves an invalid index behind; + // IF NOT EXISTS alone would otherwise skip it forever. + if valid == Some(false) { + sea_orm::sqlx::query("DROP INDEX CONCURRENTLY idx_current_kv_space_key_c") + .execute(&mut *connection) + .await?; + } + sea_orm::sqlx::query( + "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_current_kv_space_key_c \ + ON current_kv (space, key COLLATE \"C\")", + ) + .execute(&mut *connection) + .await?; + } else { + sea_orm::sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS idx_current_kv_space_key_c") + .execute(&mut *connection) + .await?; + } + Ok::<(), sea_orm::sqlx::Error>(()) + } + .await; + let close = connection.close().await; + result.map_err(|error| DbErr::Exec(sea_orm::RuntimeErr::SqlxError(error)))?; + close.map_err(|error| DbErr::Conn(sea_orm::RuntimeErr::SqlxError(error))) +} + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + #[cfg(feature = "postgres")] + if manager.get_database_backend() == sea_orm::DbBackend::Postgres { + postgres_index(manager, true).await?; + } + Ok(()) + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + #[cfg(feature = "postgres")] + if manager.get_database_backend() == sea_orm::DbBackend::Postgres { + postgres_index(manager, false).await?; + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use sea_orm::{ConnectionTrait, Statement, TransactionTrait}; + + #[tokio::test] + async fn postgres_index_build_does_not_block_a_second_writer() { + let Some(schema) = crate::test_support::PostgresTestSchema::new( + "postgres_index_build_does_not_block_a_second_writer", + ) + .await + else { + return; + }; + let db = &schema.conn; + db.execute_unprepared("DROP INDEX idx_current_kv_space_key_c") + .await + .unwrap(); + let first = db.begin().await.unwrap(); + first + .execute(Statement::from_string( + sea_orm::DbBackend::Postgres, + "INSERT INTO current_kv (space, key, invocation, seq, epoch, epoch_seq, value, metadata, deleted) \ + VALUES ('space-a', 'busy/first', decode(repeat('ab', 34), 'hex'), 10, \ + decode(repeat('ab', 34), 'hex'), 0, decode(repeat('ab', 34), 'hex'), '{}', false)", + )) + .await + .unwrap(); + + let build_db = db.clone(); + let build = tokio::spawn(async move { Migration.up(&SchemaManager::new(&build_db)).await }); + let waiting = tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let row = db + .query_one(Statement::from_string( + sea_orm::DbBackend::Postgres, + "SELECT 1 AS active FROM pg_stat_activity \ + WHERE query LIKE 'CREATE INDEX%idx_current_kv_space_key_c%' \ + AND state = 'active' AND pid <> pg_backend_pid()", + )) + .await + .unwrap(); + if row.is_some() || build.is_finished() { + break; + } + tokio::task::yield_now().await; + } + }) + .await; + assert!(waiting.is_ok(), "index build never started"); + + let second = tokio::time::timeout( + std::time::Duration::from_secs(1), + db.execute(Statement::from_string( + sea_orm::DbBackend::Postgres, + "INSERT INTO current_kv (space, key, invocation, seq, epoch, epoch_seq, value, metadata, deleted) \ + VALUES ('space-a', 'busy/second', decode(repeat('ab', 34), 'hex'), 11, \ + decode(repeat('ab', 34), 'hex'), 0, decode(repeat('ab', 34), 'hex'), '{}', false)", + )), + ) + .await; + first.rollback().await.unwrap(); + build.await.unwrap().unwrap(); + assert!( + matches!(second, Ok(Ok(_))), + "index build blocked a writer: {second:?}" + ); + } + + #[tokio::test] + async fn postgres_concurrent_index_down_and_up_restore_history() { + use sea_orm_migration::MigratorTrait; + + let Some(schema) = crate::test_support::PostgresTestSchema::new( + "postgres_concurrent_index_down_and_up_restore_history", + ) + .await + else { + return; + }; + let db = &schema.conn; + let migrations = crate::migrations::Migrator::migrations(); + let index = migrations + .iter() + .position(|migration| migration.name() == NAME) + .unwrap(); + let steps = (migrations.len() - index) as u32; + crate::migrations::Migrator::down(db, Some(steps)) + .await + .unwrap(); + let absent = db + .query_one(Statement::from_string( + sea_orm::DbBackend::Postgres, + "SELECT 1 FROM pg_index WHERE indexrelid = to_regclass('idx_current_kv_space_key_c')", + )) + .await + .unwrap(); + assert!(absent.is_none()); + crate::migrations::Migrator::up(db, Some(steps)) + .await + .unwrap(); + let valid: bool = db + .query_one(Statement::from_string( + sea_orm::DbBackend::Postgres, + "SELECT indisvalid AS valid FROM pg_index \ + WHERE indexrelid = to_regclass('idx_current_kv_space_key_c')", + )) + .await + .unwrap() + .unwrap() + .try_get("", "valid") + .unwrap(); + assert!(valid); + } + + #[tokio::test] + async fn postgres_two_nodes_can_finish_the_same_concurrent_index_migration() { + use sea_orm_migration::MigratorTrait; + + let Some(schema) = crate::test_support::PostgresTestSchema::new( + "postgres_two_nodes_can_finish_the_same_concurrent_index_migration", + ) + .await + else { + return; + }; + let db = &schema.conn; + crate::migrations::Migrator::down(db, Some(1)) + .await + .unwrap(); + db.execute(Statement::from_string( + sea_orm::DbBackend::Postgres, + "INSERT INTO current_kv (space, key, invocation, seq, epoch, epoch_seq, value, metadata, deleted) \ + SELECT 'space-a', 'busy/' || lpad(n::text, 6, '0'), \ + decode(repeat('ab', 34), 'hex'), n, decode(repeat('ab', 34), 'hex'), 0, \ + decode(repeat('ab', 34), 'hex'), '{}', false FROM generate_series(1, 100000) n", + )) + .await + .unwrap(); + let barrier = std::sync::Arc::new(tokio::sync::Barrier::new(2)); + let run = || { + let db = db.clone(); + let barrier = barrier.clone(); + async move { + barrier.wait().await; + crate::migrations::Migrator::up(&db, None).await + } + }; + let (first, second) = tokio::join!(run(), run()); + assert!(first.is_ok(), "first node migration: {first:?}"); + assert!(second.is_ok(), "second node migration: {second:?}"); + } + #[tokio::test] + async fn postgres_retry_replaces_an_invalid_concurrent_index() { + use sea_orm_migration::MigratorTrait; + + let Some(schema) = crate::test_support::PostgresTestSchema::new( + "postgres_retry_replaces_an_invalid_concurrent_index", + ) + .await + else { + return; + }; + let db = &schema.conn; + crate::migrations::Migrator::down(db, Some(1)) + .await + .unwrap(); + db.execute_unprepared( + "INSERT INTO current_kv (space, key, invocation, seq, epoch, epoch_seq, value, metadata, deleted) \ + SELECT 'space-a', 'busy/' || n::text, decode(repeat('ab', 34), 'hex'), n, \ + decode(repeat('ab', 34), 'hex'), 0, decode(repeat('ab', 34), 'hex'), '{}', false \ + FROM generate_series(1, 2) n", + ) + .await + .unwrap(); + // A failed concurrent uniqueness check leaves an INVALID catalog entry + // with the same name as the desired index. + let failed = db + .execute_unprepared( + "CREATE UNIQUE INDEX CONCURRENTLY idx_current_kv_space_key_c \ + ON current_kv (space)", + ) + .await; + assert!(failed.is_err(), "duplicate spaces must fail the build"); + let invalid: bool = db + .query_one(Statement::from_string( + sea_orm::DbBackend::Postgres, + "SELECT NOT indisvalid AS invalid FROM pg_index \ + WHERE indexrelid = to_regclass('idx_current_kv_space_key_c')", + )) + .await + .unwrap() + .unwrap() + .try_get("", "invalid") + .unwrap(); + assert!(invalid); + let history = db + .query_one(Statement::from_string( + sea_orm::DbBackend::Postgres, + format!("SELECT 1 FROM seaql_migrations WHERE version = '{NAME}'"), + )) + .await + .unwrap(); + assert!(history.is_none(), "failed build must not be marked applied"); + + crate::migrations::Migrator::up(db, Some(1)).await.unwrap(); + let ddl: String = db + .query_one(Statement::from_string( + sea_orm::DbBackend::Postgres, + "SELECT pg_get_indexdef(indexrelid) AS ddl FROM pg_index \ + WHERE indexrelid = to_regclass('idx_current_kv_space_key_c') \ + AND indisvalid", + )) + .await + .unwrap() + .unwrap() + .try_get("", "ddl") + .unwrap(); + assert!(ddl.contains("(space, key COLLATE \"C\")"), "{ddl}"); + } +} diff --git a/tinycloud-core/src/migrations/mod.rs b/tinycloud-core/src/migrations/mod.rs index ee7f0d2f..479b37af 100644 --- a/tinycloud-core/src/migrations/mod.rs +++ b/tinycloud-core/src/migrations/mod.rs @@ -1,4 +1,6 @@ +use sea_orm::{ConnectionTrait, DatabaseConnection, DbBackend, Statement, TransactionTrait}; use sea_orm_migration::prelude::*; +use sea_orm_migration::{IntoSchemaManagerConnection, SchemaManagerConnection}; pub mod m20230510_101010_init_tables; pub mod m20260218_sql_database; pub mod m20260409_000000_hook_tables; @@ -25,9 +27,125 @@ pub mod m20261005_000000_deactivate_hook_subscriptions; pub mod m20261007_000000_database_alias; pub mod m20261007_010000_database_identity_fence; pub mod m20261015_000000_delegation_cursor_index; +pub mod m20261015_010000_current_kv_prefix_index; pub struct Migrator; +/// SeaORM wraps every PostgreSQL migration batch in a transaction. The +/// byte-ordered index cannot be built or dropped there: CONCURRENTLY must run +/// between committed batches. Keep ordinary migrations atomic on either side. +async fn postgres_migrate( + db: &DatabaseConnection, + steps: Option, + rollback: bool, +) -> Result<(), DbErr> { + let initial = db.begin().await?; + initial + .execute_unprepared(crate::db_pool::EXEMPT_STATEMENT_TIMEOUT) + .await?; + Migrator::install(&initial).await?; + let mut selected = if rollback { + Migrator::get_applied_migrations(&initial).await? + } else { + Migrator::get_pending_migrations(&initial).await? + }; + if rollback { + selected.reverse(); + } + let selected = selected + .into_iter() + .take(steps.map(|steps| steps as usize).unwrap_or(usize::MAX)) + .map(|migration| migration.name().to_owned()) + .collect::>(); + let mut migrations = Migrator::migrations(); + if rollback { + migrations.reverse(); + } + let mut transaction = Some(initial); + for migration in migrations { + if !selected.iter().any(|name| name == migration.name()) { + continue; + } + let concurrent = migration.name() == m20261015_010000_current_kv_prefix_index::NAME; + if concurrent { + if let Some(open) = transaction.take() { + open.commit().await?; + } + let manager = SchemaManager::new(db); + if rollback { + migration.down(&manager).await?; + } else { + migration.up(&manager).await?; + } + } else { + if transaction.is_none() { + let open = db.begin().await?; + open.execute_unprepared(crate::db_pool::EXEMPT_STATEMENT_TIMEOUT) + .await?; + transaction = Some(open); + } + let open = transaction.as_ref().expect("transaction opened above"); + let manager = SchemaManager::new(open); + if rollback { + migration.down(&manager).await?; + } else { + migration.up(&manager).await?; + } + } + if transaction.is_none() { + let open = db.begin().await?; + open.execute_unprepared(crate::db_pool::EXEMPT_STATEMENT_TIMEOUT) + .await?; + transaction = Some(open); + } + let connection: &dyn ConnectionTrait = + transaction.as_ref().expect("transaction opened above"); + if rollback { + connection + .execute(Statement::from_sql_and_values( + DbBackend::Postgres, + "DELETE FROM seaql_migrations WHERE version = $1", + [migration.name().to_owned().into()], + )) + .await?; + } else { + // Another node can finish the same concurrent build between the + // pending-migration read and this write. + let sql = if concurrent { + "INSERT INTO seaql_migrations (version, applied_at) VALUES ($1, $2) \ + ON CONFLICT (version) DO NOTHING" + } else { + "INSERT INTO seaql_migrations (version, applied_at) VALUES ($1, $2)" + }; + connection + .execute(Statement::from_sql_and_values( + DbBackend::Postgres, + sql, + [ + migration.name().to_owned().into(), + time::OffsetDateTime::now_utc().unix_timestamp().into(), + ], + )) + .await?; + } + } + if let Some(open) = transaction { + open.commit().await?; + } + Ok(()) +} + +/// Use SeaORM's standard runner on SQLite/MySQL; only PostgreSQL needs a +/// nontransactional break around the concurrent index. +struct StandardMigrator; + +#[async_trait::async_trait] +impl MigratorTrait for StandardMigrator { + fn migrations() -> Vec> { + Migrator::migrations() + } +} + #[async_trait::async_trait] impl MigratorTrait for Migrator { fn migrations() -> Vec> { @@ -58,8 +176,51 @@ impl MigratorTrait for Migrator { Box::new(m20261007_000000_database_alias::Migration), Box::new(m20261007_010000_database_identity_fence::Migration), Box::new(m20261015_000000_delegation_cursor_index::Migration), + Box::new(m20261015_010000_current_kv_prefix_index::Migration), ] } + + async fn up<'c, C>(db: C, steps: Option) -> Result<(), DbErr> + where + C: IntoSchemaManagerConnection<'c>, + { + match db.into_schema_manager_connection() { + SchemaManagerConnection::Connection(conn) + if conn.get_database_backend() == DbBackend::Postgres => + { + postgres_migrate(conn, steps, false).await + } + SchemaManagerConnection::Transaction(conn) + if conn.get_database_backend() == DbBackend::Postgres => + { + Err(DbErr::Custom( + "PostgreSQL migrations require a connection outside a transaction".into(), + )) + } + other => StandardMigrator::up(other, steps).await, + } + } + + async fn down<'c, C>(db: C, steps: Option) -> Result<(), DbErr> + where + C: IntoSchemaManagerConnection<'c>, + { + match db.into_schema_manager_connection() { + SchemaManagerConnection::Connection(conn) + if conn.get_database_backend() == DbBackend::Postgres => + { + postgres_migrate(conn, steps, true).await + } + SchemaManagerConnection::Transaction(conn) + if conn.get_database_backend() == DbBackend::Postgres => + { + Err(DbErr::Custom( + "PostgreSQL migrations require a connection outside a transaction".into(), + )) + } + other => StandardMigrator::down(other, steps).await, + } + } } #[cfg(test)] @@ -142,4 +303,137 @@ mod release_line_tests { .unwrap(); } } + + #[tokio::test] + async fn postgres_ordinary_migration_rolls_back_all_ddl_on_failure() { + let Some(url) = crate::test_support::postgres_test_url( + "postgres_ordinary_migration_rolls_back_all_ddl_on_failure", + ) else { + return; + }; + let admin = Database::connect(url.clone()).await.unwrap(); + let schema = format!( + "tc737_atomic_{}_{}", + std::process::id(), + time::OffsetDateTime::now_utc().unix_timestamp_nanos() + ); + admin + .execute(Statement::from_string( + DbBackend::Postgres, + format!("CREATE SCHEMA {schema}"), + )) + .await + .unwrap(); + let mut options = ConnectOptions::new(url); + options.set_schema_search_path(schema.clone()); + let db = Database::connect(options).await.unwrap(); + db.execute_unprepared("CREATE TABLE actor (id text PRIMARY KEY)") + .await + .unwrap(); + // The first ordinary migration creates `space` before `actor`. + // Conflict at actor must roll back space and the migration history. + assert!(Migrator::up(&db, Some(1)).await.is_err()); + let rolled_back: bool = db + .query_one(Statement::from_string( + DbBackend::Postgres, + "SELECT to_regclass('space') IS NULL AND \ + to_regclass('seaql_migrations') IS NULL AS rolled_back", + )) + .await + .unwrap() + .unwrap() + .try_get("", "rolled_back") + .unwrap(); + assert!(rolled_back, "ordinary migration must remain atomic"); + db.execute_unprepared("DROP TABLE actor").await.unwrap(); + Migrator::up(&db, None).await.unwrap(); + drop(db); + admin + .execute(Statement::from_string( + DbBackend::Postgres, + format!("DROP SCHEMA {schema} CASCADE"), + )) + .await + .unwrap(); + } + + #[tokio::test] + async fn postgres_tc1195_and_tc737_fresh_and_from_1206() { + let Some(url) = + crate::test_support::postgres_test_url("postgres_tc1195_and_tc737_fresh_and_from_1206") + else { + return; + }; + let admin = Database::connect(url.clone()).await.unwrap(); + let before_1195 = Migrator::migrations() + .iter() + .position(|migration| migration.name() == "m20261015_000000_delegation_cursor_index") + .unwrap() as u32; + // v1.20.6 ends immediately before TC-1195 in the migration list. + for (prior, with_737) in [ + ("fresh", false), + ("fresh", true), + ("v1.20.6", false), + ("v1.20.6", true), + ] { + let schema = format!( + "tc737_{}_{}_{}", + std::process::id(), + if with_737 { "both" } else { "only1195" }, + time::OffsetDateTime::now_utc().unix_timestamp_nanos() + ); + admin + .execute(Statement::from_string( + DbBackend::Postgres, + format!("CREATE SCHEMA {schema}"), + )) + .await + .unwrap(); + let mut options = ConnectOptions::new(url.clone()); + options.set_schema_search_path(schema.clone()); + let db = Database::connect(options).await.unwrap(); + if prior == "v1.20.6" { + Migrator::up(&db, Some(before_1195)).await.unwrap(); + } + Migrator::up( + &db, + Some(if prior == "fresh" { + before_1195 + 1 + u32::from(with_737) + } else { + 1 + u32::from(with_737) + }), + ) + .await + .unwrap(); + let ddl: String = db + .query_one(Statement::from_string( + DbBackend::Postgres, + "SELECT pg_get_indexdef(to_regclass('idx_delegation_delegator')) AS ddl", + )) + .await + .unwrap() + .unwrap() + .try_get("", "ddl") + .unwrap(); + assert!(ddl.contains("(delegator, id)"), "{prior}: {ddl}"); + let prefix_index: Option = db + .query_one(Statement::from_string( + DbBackend::Postgres, + "SELECT indisvalid AS valid FROM pg_index \ + WHERE indexrelid = to_regclass('idx_current_kv_space_key_c')", + )) + .await + .unwrap() + .map(|row| row.try_get("", "valid").unwrap()); + assert_eq!(prefix_index, with_737.then_some(true), "{prior}"); + drop(db); + admin + .execute(Statement::from_string( + DbBackend::Postgres, + format!("DROP SCHEMA {schema} CASCADE"), + )) + .await + .unwrap(); + } + } } From efa3b23bd6d7fe15b5014a5228f26f740de6833d Mon Sep 17 00:00:00 2001 From: Sam Gbafa Date: Thu, 15 Oct 2026 14:37:10 +0000 Subject: [PATCH 2/5] TC-737: guard release-line migration ordering --- tinycloud-core/src/migrations/mod.rs | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/tinycloud-core/src/migrations/mod.rs b/tinycloud-core/src/migrations/mod.rs index 479b37af..4ed12ab3 100644 --- a/tinycloud-core/src/migrations/mod.rs +++ b/tinycloud-core/src/migrations/mod.rs @@ -365,10 +365,21 @@ mod release_line_tests { return; }; let admin = Database::connect(url.clone()).await.unwrap(); - let before_1195 = Migrator::migrations() + let migrations = Migrator::migrations(); + let before_1195 = migrations .iter() .position(|migration| migration.name() == "m20261015_000000_delegation_cursor_index") .unwrap() as u32; + // Older same-timestamp migrations have historical registration order; + // the new release-line tail must never precede an already applied one. + for pair in migrations[before_1195 as usize..].windows(2) { + assert!( + pair[0].name() < pair[1].name(), + "new migration must follow its predecessor: {} before {}", + pair[0].name(), + pair[1].name() + ); + } // v1.20.6 ends immediately before TC-1195 in the migration list. for (prior, with_737) in [ ("fresh", false), From 798440d3b480e1ad21d524a63ce3dbb3c1cb3ec1 Mon Sep 17 00:00:00 2001 From: Sam Gbafa Date: Thu, 15 Oct 2026 15:07:00 +0000 Subject: [PATCH 3/5] TC-737: reject unsupported PostgreSQL maintenance migrations --- docs/releasing.md | 11 +++ tinycloud-core/src/migrations/mod.rs | 101 +++++++++++++++++++++++++++ 2 files changed, 112 insertions(+) diff --git a/docs/releasing.md b/docs/releasing.md index 27b306f5..ecf0d95d 100644 --- a/docs/releasing.md +++ b/docs/releasing.md @@ -222,6 +222,17 @@ gh workflow run docker.yml -R TinyCloudLabs/tinycloud-node --ref main \ release; only the node binary rolls back. Check `tinycloud-core/src/migrations` between the two revisions first: a binary rollback across a schema migration is not DB-safe. +- After TC-737's `m20261015_010000_current_kv_prefix_index` is recorded, + an older node binary cannot restart against that database: SeaORM rejects + unknown applied migration versions. Already-running older writers can keep + writing while PostgreSQL builds the additive index concurrently, but binary + rollback is not safe. Recover by rolling forward through the coordinated + release sequence, not by promoting an older image. +- The PostgreSQL migrator supports `up` and `down` across the concurrent index. + `fresh`, `refresh`, and `reset` reject before changing schema or migration + history; SeaORM's inherited implementations require one transaction and + cannot execute `CREATE/DROP INDEX CONCURRENTLY`. SQLite and MySQL keep their + standard migrator operations. - The sealed-config preflight is skipped for rollbacks, and the post-deploy `/version` check expects `rollback_image_version`. Run the same manual verification as for a release. diff --git a/tinycloud-core/src/migrations/mod.rs b/tinycloud-core/src/migrations/mod.rs index 4ed12ab3..adbe47de 100644 --- a/tinycloud-core/src/migrations/mod.rs +++ b/tinycloud-core/src/migrations/mod.rs @@ -146,6 +146,22 @@ impl MigratorTrait for StandardMigrator { } } +/// SeaORM's default maintenance methods bypass `up`/`down` and wrap their +/// PostgreSQL work in one transaction. Reject them before touching schema or +/// history rather than entering the nontransactional index migration. +fn standard_maintenance_only<'c>( + db: SchemaManagerConnection<'c>, + operation: &str, +) -> Result, DbErr> { + if db.get_database_backend() == DbBackend::Postgres { + Err(DbErr::Custom(format!( + "PostgreSQL migrations support up/down only; {operation} is unsupported" + ))) + } else { + Ok(db) + } +} + #[async_trait::async_trait] impl MigratorTrait for Migrator { fn migrations() -> Vec> { @@ -180,6 +196,39 @@ impl MigratorTrait for Migrator { ] } + async fn fresh<'c, C>(db: C) -> Result<(), DbErr> + where + C: IntoSchemaManagerConnection<'c>, + { + StandardMigrator::fresh(standard_maintenance_only( + db.into_schema_manager_connection(), + "fresh", + )?) + .await + } + + async fn refresh<'c, C>(db: C) -> Result<(), DbErr> + where + C: IntoSchemaManagerConnection<'c>, + { + StandardMigrator::refresh(standard_maintenance_only( + db.into_schema_manager_connection(), + "refresh", + )?) + .await + } + + async fn reset<'c, C>(db: C) -> Result<(), DbErr> + where + C: IntoSchemaManagerConnection<'c>, + { + StandardMigrator::reset(standard_maintenance_only( + db.into_schema_manager_connection(), + "reset", + )?) + .await + } + async fn up<'c, C>(db: C, steps: Option) -> Result<(), DbErr> where C: IntoSchemaManagerConnection<'c>, @@ -357,6 +406,58 @@ mod release_line_tests { .unwrap(); } + #[tokio::test] + async fn postgres_maintenance_methods_reject_without_mutation() { + let Some(schema) = crate::test_support::PostgresTestSchema::new( + "postgres_maintenance_methods_reject_without_mutation", + ) + .await + else { + return; + }; + let db = &schema.conn; + db.execute_unprepared("CREATE TABLE tc737_stray (id integer)") + .await + .unwrap(); + let before: i64 = db + .query_one(Statement::from_string( + DbBackend::Postgres, + "SELECT count(*) AS count FROM seaql_migrations", + )) + .await + .unwrap() + .unwrap() + .try_get("", "count") + .unwrap(); + for (name, result) in [ + ("fresh", Migrator::fresh(db).await), + ("refresh", Migrator::refresh(db).await), + ("reset", Migrator::reset(db).await), + ] { + assert!( + result + .unwrap_err() + .to_string() + .contains(&format!("{name} is unsupported")), + "{name} must reject before DDL" + ); + } + let row = db + .query_one(Statement::from_string( + DbBackend::Postgres, + "SELECT (SELECT count(*) FROM seaql_migrations) AS count, \ + to_regclass('tc737_stray') IS NOT NULL AS stray, \ + EXISTS(SELECT 1 FROM pg_index WHERE \ + indexrelid = to_regclass('idx_current_kv_space_key_c') AND indisvalid) AS valid", + )) + .await + .unwrap() + .unwrap(); + assert_eq!(row.try_get::("", "count").unwrap(), before); + assert!(row.try_get::("", "stray").unwrap()); + assert!(row.try_get::("", "valid").unwrap()); + } + #[tokio::test] async fn postgres_tc1195_and_tc737_fresh_and_from_1206() { let Some(url) = From c70cb97ddaab5ddb310598c3eb70760fe1258e39 Mon Sep 17 00:00:00 2001 From: Sam Gbafa Date: Thu, 15 Oct 2026 17:28:52 +0000 Subject: [PATCH 4/5] TC-737: bound quiet prefix polls and byte-order listings --- docs/kv-sync.md | 7 + docs/releasing.md | 6 + tinycloud-core/src/db.rs | 456 ++++++++++++++++++++++-- tinycloud-core/src/kv_sync.rs | 150 +++++++- tinycloud-node-server/src/routes/mod.rs | 71 +++- 5 files changed, 626 insertions(+), 64 deletions(-) diff --git a/docs/kv-sync.md b/docs/kv-sync.md index ee26687d..46f1f386 100644 --- a/docs/kv-sync.md +++ b/docs/kv-sync.md @@ -69,6 +69,13 @@ root feeds advertise `kv-sync-root` alongside `kv-sync-v1` in `/info`. the node's reads. It is not an error. Keep paging with the returned cursor; the next request makes progress. +The PostgreSQL node checks a non-root prefix by seeking the exact key and +its descendants in byte-key order before reading the sequence-ordered feed. +If no key has a change after the cursor (or a live key at the bootstrap floor), +it returns the same cursor without scanning unrelated writes in a busy space. +The probe reads at most the selected prefix's keys; positive polls retain +commit ordering, tombstone handling, and whole-invocation pages. + ### Pages and `limit` All keys written or deleted by one invocation (for example a batch put) share diff --git a/docs/releasing.md b/docs/releasing.md index ecf0d95d..eb4c29c8 100644 --- a/docs/releasing.md +++ b/docs/releasing.md @@ -228,6 +228,12 @@ gh workflow run docker.yml -R TinyCloudLabs/tinycloud-node --ref main \ writing while PostgreSQL builds the additive index concurrently, but binary rollback is not safe. Recover by rolling forward through the coordinated release sequence, not by promoting an older image. +- TC-737 also orders PostgreSQL `tinycloud.kv/list` pages by key bytes, + matching SQLite's BINARY order. Newly issued list cursors use version 2. + A valid version-1 cursor from before this cutover returns HTTP 400 with + "restart listing"; the client must omit `x-tinycloud-cursor` and begin a + new listing. The node rejects rather than silently skipping or repeating + keys when a database's native collation differs from byte order. - The PostgreSQL migrator supports `up` and `down` across the concurrent index. `fresh`, `refresh`, and `reset` reject before changing schema or migration history; SeaORM's inherited implementations require one transaction and diff --git a/tinycloud-core/src/db.rs b/tinycloud-core/src/db.rs index d4c59a6d..d21aba9f 100644 --- a/tinycloud-core/src/db.rs +++ b/tinycloud-core/src/db.rs @@ -3548,9 +3548,8 @@ async fn list_bounded( /// /// The empty prefix covers the whole space. That is where this rule departs /// from `ResourceId::extends`, under which an empty base path covers only the -/// empty key and keys starting with `/`. The departure is deliberate: a -/// path-less `kv/list` lists the whole space. `tinycloud.kv/sync` rejects an -/// empty prefix (TC-732), so a feed's scope never depends on the difference. +/// empty key and keys starting with `/`. Both path-less `kv/list` and the +/// root-scoped `kv/sync` feed cover the whole space. pub fn kv_prefix_covers(prefix: &str, key: &str) -> bool { prefix.is_empty() || key @@ -3563,6 +3562,20 @@ fn kv_list_cursor_in_prefix(prefix: &Path, cursor: Option<&Path>) -> bool { cursor.is_none_or(|cursor| kv_prefix_covers(prefix.as_str(), cursor.as_str())) } +/// Half-open byte range of descendants, and whether the bare key also matches. +pub(crate) fn kv_prefix_bounds(prefix: &str) -> Option<(String, String, bool)> { + if prefix.is_empty() { + return None; + } + let (lower, bare_prefix) = if prefix.ends_with('/') { + (prefix.to_owned(), false) + } else { + (format!("{prefix}/"), true) + }; + let upper = format!("{}0", &lower[..lower.len() - 1]); + Some((lower, upper, bare_prefix)) +} + /// Compare arbitrary resource columns in byte order on every backend. See /// `byte_ordered_resource` for why a column's native collation is not trusted. /// @@ -3574,15 +3587,7 @@ fn prefix_condition( column: impl Into + Clone, prefix: &str, ) -> Option { - if prefix.is_empty() { - return None; - } - let (lower, bare_prefix) = if prefix.ends_with('/') { - (prefix.to_owned(), false) - } else { - (format!("{prefix}/"), true) - }; - let upper = format!("{}0", &lower[..lower.len() - 1]); + let (lower, upper, bare_prefix) = kv_prefix_bounds(prefix)?; let descendants = Condition::all() .add(byte_ordered(backend, column.clone()).gte(lower)) .add(byte_ordered(backend, column.clone()).lt(upper)); @@ -3603,43 +3608,85 @@ pub(crate) fn kv_prefix_condition(backend: DbBackend, prefix: &str) -> Option after`) and orders -/// (`ORDER BY key`) in one collation -- the column's own -- inside SQL, so a -/// page boundary can never skip or repeat a key; membership is decided by the -/// byte-ordered `kv_prefix_condition`. One more row than `limit` is fetched to -/// detect truncation. -fn kv_list_statement( +#[derive(Debug, Clone, Copy)] +enum KvListBranch { + All, + Bare, + Descendants, +} + +/// On PostgreSQL, compare the keyset cursor and order rows in the same byte +/// collation used for prefix membership. Other backends retain their existing +/// native listing order (SQLite's is already byte order). +fn kv_list_branch_statement( backend: DbBackend, space_id: &SpaceId, prefix: &Path, - limit: Option, after: Option<&Path>, + row_limit: Option, + branch: KvListBranch, ) -> Statement { + let key = Expr::col((current_kv::Entity, current_kv::Column::Key)); let mut condition = Condition::all() .add(Expr::col(current_kv::Column::Deleted).eq(false)) .add( Expr::col((current_kv::Entity, current_kv::Column::Space)) .eq(SpaceIdWrap(space_id.clone())), ); - if let Some(prefix) = kv_prefix_condition(backend, prefix.as_str()) { - condition = condition.add(prefix); + match branch { + KvListBranch::All => { + if let Some(prefix) = kv_prefix_condition(backend, prefix.as_str()) { + condition = condition.add(prefix); + } + } + KvListBranch::Bare => { + condition = condition.add(byte_ordered(backend, key.clone()).eq(prefix.as_str())); + } + KvListBranch::Descendants => { + let (lower, upper, _) = + kv_prefix_bounds(prefix.as_str()).expect("descendants need a prefix"); + condition = condition + .add(byte_ordered(backend, key.clone()).gte(lower)) + .add(byte_ordered(backend, key.clone()).lt(upper)); + } } + let ordered_key: SimpleExpr = if backend == DbBackend::Postgres { + byte_ordered(backend, key.clone()) + } else { + key.clone().into() + }; if let Some(after) = after { - condition = condition - .add(Expr::col((current_kv::Entity, current_kv::Column::Key)).gt(after.as_str())); + condition = condition.add(ordered_key.clone().gt(after.as_str())); } let mut query = Query::select(); query .column((current_kv::Entity, current_kv::Column::Key)) .from(current_kv::Entity) .cond_where(condition) - .order_by((current_kv::Entity, current_kv::Column::Key), Order::Asc); - if let Some(limit) = limit { - query.limit(limit.saturating_add(1) as u64); + .order_by_expr(ordered_key, Order::Asc); + if let Some(row_limit) = row_limit { + query.limit(row_limit as u64); } backend.build(&query) } +fn kv_list_statement( + backend: DbBackend, + space_id: &SpaceId, + prefix: &Path, + limit: Option, + after: Option<&Path>, +) -> Statement { + kv_list_branch_statement( + backend, + space_id, + prefix, + after, + limit.map(|limit| limit.saturating_add(1)), + KvListBranch::All, + ) +} + /// List live keys inside `prefix` (by `kv_prefix_covers`), at most `limit` of /// them, strictly after the keyset cursor `after`. Callers validate `after` /// against `prefix`. @@ -3650,10 +3697,48 @@ async fn list_bounded_after( limit: Option, after: Option<&Path>, ) -> Result<(Vec, bool), DbErr> { - let statement = kv_list_statement(db.get_database_backend(), space_id, prefix, limit, after); - let mut list = db - .query_all(statement) - .await? + let backend = db.get_database_backend(); + let rows = if backend == DbBackend::Postgres + && !prefix.as_str().is_empty() + && !prefix.as_str().ends_with('/') + { + // An OR between the bare key and descendants can make a generic + // plan scan the whole space. The bare key sorts before every child. + let max_rows = limit.map(|limit| limit.saturating_add(1)); + let mut rows = Vec::new(); + if after.is_none() { + rows.extend( + db.query_all(kv_list_branch_statement( + backend, + space_id, + prefix, + None, + Some(1), + KvListBranch::Bare, + )) + .await?, + ); + } + if max_rows.is_none_or(|max_rows| rows.len() < max_rows) { + let remaining = max_rows.map(|max_rows| max_rows - rows.len()); + rows.extend( + db.query_all(kv_list_branch_statement( + backend, + space_id, + prefix, + after, + remaining, + KvListBranch::Descendants, + )) + .await?, + ); + } + rows + } else { + db.query_all(kv_list_statement(backend, space_id, prefix, limit, after)) + .await? + }; + let mut list = rows .into_iter() .map(|row| row.try_get::("", current_kv::Column::Key.as_str())) .collect::, _>>()? @@ -3865,7 +3950,7 @@ fn byte_ordered_resource(backend: DbBackend) -> SimpleExpr { } /// `column`, compared in byte order on `backend` (see `byte_ordered_resource`). -fn byte_ordered(backend: DbBackend, column: impl Into) -> SimpleExpr { +pub(crate) fn byte_ordered(backend: DbBackend, column: impl Into) -> SimpleExpr { // `$1` / `?` is sea-query's per-backend placeholder for the embedded // column expression, not for a bound value. match backend { @@ -5628,6 +5713,315 @@ mod test { ); } + #[tokio::test] + async fn postgres_kv_list_byte_pages_match_sqlite_with_tombstones() { + let Some(schema) = crate::test_support::PostgresTestSchema::new( + "postgres_kv_list_byte_pages_match_sqlite_with_tombstones", + ) + .await + else { + return; + }; + let sqlite = get_db().await.unwrap(); + let keys = [ + "docs/é", + "docs/Z", + "docs/a", + "docs", + "docs/界", + "docsecret/x", + "docs/Q", + ]; + for (conn, backend, name) in [ + (&schema.conn, DbBackend::Postgres, "pg-byte-order"), + (&sqlite.conn, DbBackend::Sqlite, "sqlite-byte-order"), + ] { + let space = seed_kv_keys(conn, name, &keys).await; + conn.execute(Statement::from_sql_and_values( + backend, + if backend == DbBackend::Postgres { + "UPDATE current_kv SET deleted = true WHERE space = $1 AND key = 'docs/Q'" + } else { + "UPDATE current_kv SET deleted = true WHERE space = ? AND key = 'docs/Q'" + }, + [SpaceIdWrap(space.clone()).into()], + )) + .await + .unwrap(); + assert_eq!( + kv_list_pages(conn, &space, "docs", 2).await, + vec![ + vec!["docs", "docs/Z"], + vec!["docs/a", "docs/é"], + vec!["docs/界"], + ], + "{backend:?}" + ); + assert_eq!( + kv_list_pages(conn, &space, "docs/", 1).await, + vec![ + vec!["docs/Z"], + vec!["docs/a"], + vec!["docs/é"], + vec!["docs/界"], + ], + "{backend:?}" + ); + } + } + + /// The auto and generic plans must seek by byte key under cross-space + /// skew. This is intentionally too large for the default test suite. + #[tokio::test] + #[ignore = "PG18.6 planner fixture: 2 spaces x 300k current_kv rows"] + async fn postgres_kv_prefix_actual_builders_seek_under_skew() { + let Some(schema) = crate::test_support::PostgresTestSchema::new( + "postgres_kv_prefix_actual_builders_seek_under_skew", + ) + .await + else { + return; + }; + let db = &schema.conn; + let rare = seed_kv_keys(db, "tc737-skew-rare", &["busy/only"]).await; + let dense = seed_kv_keys(db, "tc737-skew-dense", &[]).await; + let epoch = crate::hash::hash(b"tc737 skew epoch"); + for (space, prefix) in [(&rare, "noise/"), (&dense, "busy/")] { + db.execute(Statement::from_sql_and_values( + DbBackend::Postgres, + "INSERT INTO current_kv (space, key, invocation, seq, epoch, epoch_seq, value, metadata, deleted) \ + SELECT $1, $3::text || lpad(n::text, 6, '0'), $2, n + 10, $2, 0, $2, '{}', false \ + FROM generate_series(1, 300000) AS n", + [ + SpaceIdWrap(space.clone()).into(), + epoch.into(), + prefix.into(), + ], + )) + .await + .unwrap(); + } + db.execute(Statement::from_string( + DbBackend::Postgres, + "ANALYZE current_kv", + )) + .await + .unwrap(); + + async fn explain(db: &C, statement: Statement) -> String { + static NEXT_PLAN: AtomicU64 = AtomicU64::new(0); + let name = format!( + "tc737_builder_{}", + NEXT_PLAN.fetch_add(1, Ordering::Relaxed) + ); + let values = statement.values.map(|values| values.0).unwrap_or_default(); + let args = (1..=values.len()) + .map(|i| format!("${i}")) + .collect::>() + .join(", "); + db.execute(Statement::from_string( + DbBackend::Postgres, + format!("PREPARE {name} AS {}", statement.sql), + )) + .await + .unwrap(); + let execute = Statement::from_sql_and_values( + DbBackend::Postgres, + format!("EXECUTE {name}({args})"), + values, + ) + .to_string(); + let plan = db + .query_all(Statement::from_string( + DbBackend::Postgres, + format!("EXPLAIN (ANALYZE, BUFFERS) {execute}"), + )) + .await + .unwrap() + .into_iter() + .map(|row| row.try_get_by_index::(0).unwrap()) + .collect::>() + .join("\n"); + db.execute(Statement::from_string( + DbBackend::Postgres, + format!("DEALLOCATE {name}"), + )) + .await + .unwrap(); + plan + } + // seed_kv_keys wrote a real event_order row at this floor/position. + let pos = crate::kv_sync::KvSyncAnchor { + seq: 0, + epoch: crate::hash::hash(b"epoch-0"), + epoch_seq: 0, + }; + let floor = pos; + let bare: Path = "busy".parse().unwrap(); + let slash: Path = "busy/".parse().unwrap(); + let after: Path = "busy/000500".parse().unwrap(); + for mode in ["auto", "force_generic_plan"] { + let tx = db.begin().await.unwrap(); + tx.execute(Statement::from_string( + DbBackend::Postgres, + format!("SET LOCAL plan_cache_mode = {mode}"), + )) + .await + .unwrap(); + let actual_mode = tx + .query_one(Statement::from_string( + DbBackend::Postgres, + "SHOW plan_cache_mode", + )) + .await + .unwrap() + .unwrap() + .try_get_by_index::(0) + .unwrap(); + assert_eq!(actual_mode, mode); + println!("TC-737 skew planner mode: {actual_mode}"); + for (space, label) in [(&rare, "rare"), (&dense, "dense")] { + for (prefix, branch) in [ + (&bare, crate::kv_sync::KvPrefixProbe::Exact), + (&bare, crate::kv_sync::KvPrefixProbe::Descendants), + (&slash, crate::kv_sync::KvPrefixProbe::Descendants), + ] { + let plan = explain( + &tx, + crate::kv_sync::kv_sync_probe_statement( + DbBackend::Postgres, + space, + prefix, + if label == "rare" { Some(&pos) } else { None }, + if label == "rare" { Some(&floor) } else { None }, + branch, + ), + ) + .await; + println!( + "{mode} {label} probe {branch:?} {}:\n{plan}", + prefix.as_str() + ); + if label == "rare" && !matches!(branch, crate::kv_sync::KvPrefixProbe::Exact) { + assert!( + plan.contains("idx_current_kv_space_key_c") + && plan.lines().any( + |line| line.contains("Index Cond:") && line.contains("key") + ), + "{mode} rare prefix must seek by C key:\n{plan}" + ); + if mode == "force_generic_plan" { + assert!( + plan.contains("$1"), + "generic execution must retain parameter placeholders:\n{plan}" + ); + } + let filtered = plan + .lines() + .filter_map(|line| line.split("Rows Removed by Filter: ").nth(1)) + .map(|count| count.trim().parse::().unwrap()) + .sum::(); + assert!( + filtered <= 10, + "{mode} rare prefix filtered {filtered} unrelated rows:\n{plan}" + ); + } + } + for (prefix, after, branch) in [ + (&bare, None, KvListBranch::Bare), + (&slash, None, KvListBranch::All), + (&bare, None, KvListBranch::Descendants), + (&slash, Some(&after), KvListBranch::All), + (&bare, Some(&after), KvListBranch::Descendants), + ] { + let plan = explain( + &tx, + kv_list_branch_statement( + DbBackend::Postgres, + space, + prefix, + after, + Some(501), + branch, + ), + ) + .await; + println!( + "{mode} {label} list {branch:?} {} after={after:?}:\n{plan}", + prefix.as_str() + ); + assert!( + (matches!(branch, KvListBranch::Bare) + || plan.contains("idx_current_kv_space_key_c")) + && plan + .lines() + .any(|line| line.contains("Index Cond:") && line.contains("key")), + "{mode} {label} list must seek by key:\n{plan}" + ); + if label == "rare" { + let filtered = plan + .lines() + .filter_map(|line| line.split("Rows Removed by Filter: ").nth(1)) + .map(|count| count.trim().parse::().unwrap()) + .sum::(); + assert!( + filtered <= 10, + "{mode} rare list filtered {filtered}:\n{plan}" + ); + } + } + } + tx.rollback().await.unwrap(); + } + for prefix in [&bare, &slash] { + let state = crate::kv_sync::KvSyncState { + pos: Some(pos), + floor: Some(floor), + }; + let (changes, more, next) = + crate::kv_sync::kv_sync_page(db, &rare, prefix, 500, Some(&state)) + .await + .unwrap(); + assert!(changes.is_empty() && !more); + assert_eq!(next, state, "quiet poll must preserve the cursor state"); + let cursor_key = [7; 32]; + let cursor = + crate::kv_sync::encode_kv_sync_cursor(&cursor_key, &rare, prefix, &state).unwrap(); + let (changes, more, echo) = crate::kv_sync::kv_sync_read( + db, + &rare, + prefix, + &crate::kv_sync::KvSyncRequest { + limit: 500, + cursor: Some(cursor.clone()), + cursor_key, + retention_grant: None, + }, + ) + .await + .unwrap(); + assert!(changes.is_empty() && !more); + assert_eq!( + echo, cursor, + "outside writes cannot change the cursor bytes" + ); + } + assert_eq!( + kv_list_pages(db, &rare, "busy", 1).await, + vec![vec!["busy/only"]] + ); + assert_eq!( + list_bounded_after(db, &dense, &slash, Some(2), None) + .await + .unwrap() + .0 + .iter() + .map(Path::as_str) + .collect::>(), + vec!["busy/000001", "busy/000002"] + ); + } + #[test] fn kv_list_cursor_outside_prefix_is_rejected() { let path = |s: &str| s.parse::().unwrap(); diff --git a/tinycloud-core/src/kv_sync.rs b/tinycloud-core/src/kv_sync.rs index dfc74b9c..33b86052 100644 --- a/tinycloud-core/src/kv_sync.rs +++ b/tinycloud-core/src/kv_sync.rs @@ -45,7 +45,7 @@ use chacha20poly1305::{ use rand::{rngs::OsRng, RngCore}; use sea_orm::{ entity::prelude::*, - sea_query::{Expr, Order, Query, SelectStatement}, + sea_query::{Expr, ExprTrait, Order, Query, SelectStatement}, Condition, ConnectionTrait, DbBackend, QueryOrder, QueryResult, QuerySelect, Statement, }; use serde::{Deserialize, Serialize}; @@ -637,6 +637,19 @@ fn position_value(anchor: &KvSyncAnchor) -> Expr { ]) } +fn feed_condition(space: &SpaceId, floor: Option<&KvSyncAnchor>) -> Condition { + let mut condition = + Condition::all().add(col(current_kv::Column::Space).eq(SpaceIdWrap(space.clone()))); + if let Some(floor) = floor { + condition = condition.add( + Condition::any() + .add(col(current_kv::Column::Deleted).eq(false)) + .add(col(current_kv::Column::Seq).gt(floor.seq)), + ); + } + condition +} + /// The rows of `space` inside `prefix`, minus tombstones at or below /// `floor.seq`, with the feed's columns. fn feed_select( @@ -645,18 +658,10 @@ fn feed_select( prefix: &Path, floor: Option<&KvSyncAnchor>, ) -> SelectStatement { - let mut condition = - Condition::all().add(col(current_kv::Column::Space).eq(SpaceIdWrap(space.clone()))); + let mut condition = feed_condition(space, floor); if let Some(prefix) = crate::db::kv_prefix_condition(backend, prefix.as_str()) { condition = condition.add(prefix); } - if let Some(floor) = floor { - condition = condition.add( - Condition::any() - .add(col(current_kv::Column::Deleted).eq(false)) - .add(col(current_kv::Column::Seq).gt(floor.seq)), - ); - } let mut query = Query::select(); query .columns([ @@ -673,6 +678,89 @@ fn feed_select( query } +#[derive(Debug, Clone, Copy)] +pub(crate) enum KvPrefixProbe { + Exact, + Descendants, +} + +/// Seek keys in byte order before running the sequence-ordered feed. The +/// disjoint exact and descendant probes avoid the bare-prefix OR plan. +pub(crate) fn kv_sync_probe_statement( + backend: DbBackend, + space: &SpaceId, + prefix: &Path, + pos: Option<&KvSyncAnchor>, + floor: Option<&KvSyncAnchor>, + branch: KvPrefixProbe, +) -> Statement { + let key = col(current_kv::Column::Key); + let mut condition = feed_condition(space, floor); + condition = match branch { + KvPrefixProbe::Exact => { + condition.add(crate::db::byte_ordered(backend, key.clone()).eq(prefix.as_str())) + } + KvPrefixProbe::Descendants => { + let (lower, upper, _) = + crate::db::kv_prefix_bounds(prefix.as_str()).expect("probe needs a prefix"); + condition + .add(crate::db::byte_ordered(backend, key.clone()).gte(lower)) + .add(crate::db::byte_ordered(backend, key.clone()).lt(upper)) + } + }; + if let Some(pos) = pos { + condition = condition + .add(col(current_kv::Column::Seq).gte(pos.seq)) + .add(position_columns().gt(position_value(pos))); + } + let mut query = Query::select(); + query + .expr(Expr::cust("1")) + .from(current_kv::Entity) + .cond_where(condition) + .order_by_expr(crate::db::byte_ordered(backend, key), Order::Asc) + .limit(1); + backend.build(&query) +} + +/// Only called for a non-empty PostgreSQL prefix; other engines and the +/// whole-space feed retain their original single sequence-ordered query. +async fn prefix_has_feed_rows( + db: &C, + space: &SpaceId, + prefix: &Path, + pos: Option<&KvSyncAnchor>, + floor: Option<&KvSyncAnchor>, +) -> Result { + let backend = db.get_database_backend(); + if !prefix.as_str().ends_with('/') + && db + .query_one(kv_sync_probe_statement( + backend, + space, + prefix, + pos, + floor, + KvPrefixProbe::Exact, + )) + .await? + .is_some() + { + return Ok(true); + } + Ok(db + .query_one(kv_sync_probe_statement( + backend, + space, + prefix, + pos, + floor, + KvPrefixProbe::Descendants, + )) + .await? + .is_some()) +} + /// The feed query: rows in `prefix` at positions strictly after `pos`, in /// position order, at most `limit + 1` of them. /// @@ -756,7 +844,7 @@ async fn feed_rows( .collect() } -/// Read one page of the feed for `prefix` (non-empty) in `space`. +/// Read one page of the feed for `prefix` in `space` (empty means root). /// /// `state` is `None` on the first page, which anchors a fresh floor at the /// space's newest event. Otherwise both the position and floor anchors must @@ -794,6 +882,12 @@ pub(crate) async fn kv_sync_page( }; let backend = db.get_database_backend(); let floor = state.floor.as_ref(); + if backend == DbBackend::Postgres + && !prefix.as_str().is_empty() + && !prefix_has_feed_rows(db, space, prefix, state.pos.as_ref(), floor).await? + { + return Ok((Vec::new(), false, state)); + } let mut rows = feed_rows( db, kv_sync_statement(backend, space, prefix, limit, state.pos.as_ref(), floor), @@ -813,12 +907,16 @@ pub(crate) async fn kv_sync_page( ) .await?, ); - !feed_rows( - db, - kv_sync_statement(backend, space, prefix, 1, Some(&last), floor), - ) - .await? - .is_empty() + if backend == DbBackend::Postgres && !prefix.as_str().is_empty() { + prefix_has_feed_rows(db, space, prefix, Some(&last), floor).await? + } else { + !feed_rows( + db, + kv_sync_statement(backend, space, prefix, 1, Some(&last), floor), + ) + .await? + .is_empty() + } } else { rows.truncate(limit); true @@ -2798,8 +2896,8 @@ pub(crate) mod tests { let baseline = db .query_all(Statement::from_sql_and_values( DbBackend::Postgres, - explain_sql.clone(), - values.clone(), + explain_sql, + values, )) .await .unwrap() @@ -2905,6 +3003,20 @@ pub(crate) mod tests { reads > 0 && writes > 0, "no read/write overlap with index build" ); + let statement = kv_sync_probe_statement( + DbBackend::Postgres, + &quiet.space, + &"quiet/".parse().unwrap(), + Some(&KvSyncAnchor { + seq: 10, + epoch, + epoch_seq: 0, + }), + None, + KvPrefixProbe::Descendants, + ); + let explain_sql = format!("EXPLAIN (ANALYZE, BUFFERS) {}", statement.sql); + let values = statement.values.map(|values| values.0).unwrap_or_default(); let plan = db .query_all(Statement::from_sql_and_values( DbBackend::Postgres, diff --git a/tinycloud-node-server/src/routes/mod.rs b/tinycloud-node-server/src/routes/mod.rs index 94cf9f1d..a0ac469d 100644 --- a/tinycloud-node-server/src/routes/mod.rs +++ b/tinycloud-node-server/src/routes/mod.rs @@ -853,6 +853,13 @@ fn parse_positive_u64_header( } type KvCursorMac = Hmac; +const KV_LIST_CURSOR_VERSION: u8 = 2; + +#[derive(Debug)] +enum KvListCursorError { + Invalid, + RestartListing, +} #[derive(Debug, Clone, Serialize, Deserialize)] struct KvListCursorPayload { @@ -891,7 +898,7 @@ fn encode_kv_cursor( last: &Path, ) -> Result { let payload = KvListCursorPayload { - version: 1, + version: KV_LIST_CURSOR_VERSION, subject_digest: kv_subject_digest(subject), space: space.to_string(), prefix: prefix.to_string(), @@ -914,26 +921,33 @@ fn decode_kv_cursor( space: &SpaceId, prefix: &Path, limit: usize, -) -> Result { +) -> Result { if value.len() > 4096 { - return Err(()); + return Err(KvListCursorError::Invalid); } - let bytes = decode_config(value, URL_SAFE_NO_PAD).map_err(|_| ())?; - let wire: KvListCursorWire = serde_json::from_slice(&bytes).map_err(|_| ())?; - if wire.payload.version != 1 - || wire.payload.subject_digest != kv_subject_digest(subject) + let bytes = decode_config(value, URL_SAFE_NO_PAD).map_err(|_| KvListCursorError::Invalid)?; + let wire: KvListCursorWire = + serde_json::from_slice(&bytes).map_err(|_| KvListCursorError::Invalid)?; + let expected = kv_cursor_mac(key, &wire.payload).map_err(|_| KvListCursorError::Invalid)?; + if wire.payload.subject_digest != kv_subject_digest(subject) || wire.payload.space != space.to_string() || wire.payload.prefix != prefix.to_string() || wire.payload.action != "tinycloud.kv/list" || wire.payload.limit != limit + || !constant_time_eq(expected.as_bytes(), wire.mac.as_bytes()) { - return Err(()); + return Err(KvListCursorError::Invalid); } - let expected = kv_cursor_mac(key, &wire.payload)?; - if !constant_time_eq(expected.as_bytes(), wire.mac.as_bytes()) { - return Err(()); + if wire.payload.version == 1 { + return Err(KvListCursorError::RestartListing); } - wire.payload.last.parse().map_err(|_| ()) + if wire.payload.version != KV_LIST_CURSOR_VERSION { + return Err(KvListCursorError::Invalid); + } + wire.payload + .last + .parse() + .map_err(|_| KvListCursorError::Invalid) } fn constant_time_eq(left: &[u8], right: &[u8]) -> bool { @@ -1196,8 +1210,15 @@ fn kv_invoke_options_for_capabilities_with_cursor( .ok_or_else(|| (Status::BadRequest, "Invalid KV list cursor".to_string()))?; let limit = list_limit .ok_or_else(|| (Status::BadRequest, "Invalid KV list cursor".to_string()))?; - decode_kv_cursor(key, &value, subject, &space, &prefix, limit) - .map_err(|_| (Status::BadRequest, "Invalid KV list cursor".to_string())) + decode_kv_cursor(key, &value, subject, &space, &prefix, limit).map_err(|error| { + let message = match error { + KvListCursorError::Invalid => "Invalid KV list cursor", + KvListCursorError::RestartListing => { + "KV list cursor predates byte ordering; restart listing without x-tinycloud-cursor" + } + }; + (Status::BadRequest, message.to_string()) + }) }) .transpose()?; @@ -8519,6 +8540,28 @@ mod tests { let (status, body, _) = list_page(Some(forged)).await; assert_eq!(status, Status::BadRequest, "{body}"); assert!(body.contains("outside the requested prefix"), "{body}"); + // A pre-cutover cursor was signed for the native-collation ordering. + // It must not silently resume a byte-ordered page in the middle. + let mut legacy: KvListCursorWire = + serde_json::from_slice(&decode_config(&cursors[0], URL_SAFE_NO_PAD)?)?; + legacy.payload.version = 1; + legacy.mac = kv_cursor_mac(&cursor_key, &legacy.payload).expect("mac"); + let legacy = encode_config(serde_json::to_vec(&legacy)?, URL_SAFE_NO_PAD); + let (status, body, _) = list_page(Some(legacy.clone())).await; + assert_eq!(status, Status::BadRequest, "{body}"); + assert!(body.contains("restart listing"), "{body}"); + let (status, body, _) = list_page(None).await; + assert_eq!(status, Status::Ok, "{body}"); + assert_eq!(serde_json::from_str::>(&body)?, vec!["docs"]); + + // Editing the version without its MAC does not get the restart hint. + let mut tampered: KvListCursorWire = + serde_json::from_slice(&decode_config(&legacy, URL_SAFE_NO_PAD)?)?; + tampered.payload.version = 2; + let tampered = encode_config(serde_json::to_vec(&tampered)?, URL_SAFE_NO_PAD); + let (status, body, _) = list_page(Some(tampered)).await; + assert_eq!(status, Status::BadRequest, "{body}"); + assert!(body.contains("Invalid KV list cursor"), "{body}"); Ok(()) } From c02d6d30df2e24b231cccd80dded4f227c65d669 Mon Sep 17 00:00:00 2001 From: Sam Gbafa Date: Thu, 15 Oct 2026 18:07:49 +0000 Subject: [PATCH 5/5] fix: bound dense KV sync prefix continuations --- docs/releasing.md | 5 + tinycloud-core/src/db.rs | 274 +++++++++++++++++++++++++++++++++- tinycloud-core/src/kv_sync.rs | 194 ++++++++++++++++++++++-- 3 files changed, 462 insertions(+), 11 deletions(-) diff --git a/docs/releasing.md b/docs/releasing.md index eb4c29c8..1c8803b3 100644 --- a/docs/releasing.md +++ b/docs/releasing.md @@ -234,6 +234,11 @@ gh workflow run docker.yml -R TinyCloudLabs/tinycloud-node --ref main \ "restart listing"; the client must omit `x-tinycloud-cursor` and begin a new listing. The node rejects rather than silently skipping or repeating keys when a database's native collation differs from byte order. +- PostgreSQL `tinycloud.kv/sync` prefix polls first inspect at most 64 raw + sequence-index candidates; dense continuations use a bounded sequence feed, + while a full window without a match falls back to the C-ordered key index. + Empty polls preserve the supplied cursor bytes. The root feed and + SQLite/MySQL behavior are unchanged. - The PostgreSQL migrator supports `up` and `down` across the concurrent index. `fresh`, `refresh`, and `reset` reject before changing schema or migration history; SeaORM's inherited implementations require one transaction and diff --git a/tinycloud-core/src/db.rs b/tinycloud-core/src/db.rs index d21aba9f..6f43e54b 100644 --- a/tinycloud-core/src/db.rs +++ b/tinycloud-core/src/db.rs @@ -5770,8 +5770,9 @@ mod test { } } - /// The auto and generic plans must seek by byte key under cross-space - /// skew. This is intentionally too large for the default test suite. + /// Dense continuations seek by sequence, while quiet prefixes seek by + /// byte key, in both auto and generic plans. This fixture is too large + /// for the default suite. #[tokio::test] #[ignore = "PG18.6 planner fixture: 2 spaces x 300k current_kv rows"] async fn postgres_kv_prefix_actual_builders_seek_under_skew() { @@ -5801,6 +5802,60 @@ mod test { .await .unwrap(); } + // Keep actual event_order anchors at the middle, the next whole + // invocation, near the end, and at the end. Other bulk rows only + // shape the planner's space/prefix distribution. + let mut dense_anchors = Vec::new(); + for (seq, keys) in [ + (150_010, &["busy/150000"][..]), + ( + 150_011, + &["busy/150001", "busy/150002", "busy/150003", "busy/150004"][..], + ), + (299_910, &["busy/299900"][..]), + (300_010, &["busy/300000"][..]), + ] { + let epoch_id = crate::hash::hash(format!("tc737-dense-epoch-{seq}").as_bytes()); + epoch::ActiveModel { + seq: Set(seq), + id: Set(epoch_id), + space: Set(SpaceIdWrap(dense.clone())), + } + .insert(db) + .await + .unwrap(); + event_order::ActiveModel { + seq: Set(seq), + epoch: Set(epoch_id), + epoch_seq: Set(0), + event: Set(crate::hash::hash( + format!("tc737-dense-event-{seq}").as_bytes(), + )), + space: Set(SpaceIdWrap(dense.clone())), + } + .insert(db) + .await + .unwrap(); + for key in keys { + db.execute(Statement::from_sql_and_values( + DbBackend::Postgres, + "UPDATE current_kv SET seq = $2, epoch = $3 WHERE space = $1 AND key = $4", + [ + SpaceIdWrap(dense.clone()).into(), + seq.into(), + epoch_id.into(), + (*key).into(), + ], + )) + .await + .unwrap(); + } + dense_anchors.push(crate::kv_sync::KvSyncAnchor { + seq, + epoch: epoch_id, + epoch_seq: 0, + }); + } db.execute(Statement::from_string( DbBackend::Postgres, "ANALYZE current_kv", @@ -5850,6 +5905,18 @@ mod test { .unwrap(); plan } + fn execution_buffers(plan: &str) -> usize { + plan.lines() + .find(|line| line.trim_start().starts_with("Buffers: shared ")) + .expect("EXPLAIN must report execution buffers") + .split_whitespace() + .filter_map(|part| { + part.strip_prefix("hit=") + .or_else(|| part.strip_prefix("read=")) + .map(|count| count.parse::().unwrap()) + }) + .sum() + } // seed_kv_keys wrote a real event_order row at this floor/position. let pos = crate::kv_sync::KvSyncAnchor { seq: 0, @@ -5880,6 +5947,101 @@ mod test { .unwrap(); assert_eq!(actual_mode, mode); println!("TC-737 skew planner mode: {actual_mode}"); + for (label, space, prefix, cursor, floor, found, seen) in [ + ("rare-bare", &rare, &bare, &pos, &floor, false, 64), + ("rare-slash", &rare, &slash, &pos, &floor, false, 64), + ( + "dense-mid-bare", + &dense, + &bare, + &dense_anchors[0], + &dense_anchors[3], + true, + 64, + ), + ( + "dense-mid-slash", + &dense, + &slash, + &dense_anchors[0], + &dense_anchors[3], + true, + 64, + ), + ( + "dense-group-more", + &dense, + &slash, + &dense_anchors[1], + &dense_anchors[3], + true, + 64, + ), + ( + "dense-late", + &dense, + &bare, + &dense_anchors[2], + &dense_anchors[3], + true, + 64, + ), + ( + "dense-caught-up", + &dense, + &slash, + &dense_anchors[3], + &dense_anchors[3], + false, + 0, + ), + ] { + let statement = crate::kv_sync::kv_sync_candidate_statement( + DbBackend::Postgres, + space, + prefix, + Some(cursor), + Some(floor), + ); + let row = tx.query_one(statement.clone()).await.unwrap().unwrap(); + assert_eq!(row.try_get::("", "seen").unwrap(), seen, "{label}"); + assert_eq!(row.try_get::("", "found").unwrap(), found, "{label}"); + let plan = explain(&tx, statement).await; + println!("{mode} {label} bounded candidate:\n{plan}"); + assert!( + plan.contains("idx_current_kv_space_order") + && plan.contains("Limit") + && execution_buffers(&plan) <= 96, + "{mode} {label} preflight must stay bounded:\n{plan}" + ); + if mode == "force_generic_plan" { + assert!( + plan.contains("$1"), + "{mode} {label} was not generic:\n{plan}" + ); + } + if found && space == &dense { + let feed = explain( + &tx, + crate::kv_sync::kv_sync_bounded_feed_statement( + DbBackend::Postgres, + space, + prefix, + 500, + Some(cursor), + Some(floor), + ), + ) + .await; + println!("{mode} {label} bounded sequence feed:\n{feed}"); + assert!( + feed.contains("idx_current_kv_space_order") + && feed.contains("Limit") + && execution_buffers(&plan) + execution_buffers(&feed) <= 160, + "{mode} {label} preflight+feed scanned consumed keys:\n{plan}\n{feed}" + ); + } + } for (space, label) in [(&rare, "rare"), (&dense, "dense")] { for (prefix, branch) in [ (&bare, crate::kv_sync::KvPrefixProbe::Exact), @@ -6006,6 +6168,114 @@ mod test { "outside writes cannot change the cursor bytes" ); } + let mid = crate::kv_sync::KvSyncState { + pos: Some(dense_anchors[0]), + floor: Some(dense_anchors[3]), + }; + let (group, more, next) = crate::kv_sync::kv_sync_page(db, &dense, &slash, 1, Some(&mid)) + .await + .unwrap(); + assert_eq!( + group + .iter() + .map(|change| change.key.as_str()) + .collect::>(), + vec!["busy/150001", "busy/150002", "busy/150003", "busy/150004",] + ); + assert!(more); + assert_eq!(next.pos, Some(dense_anchors[1])); + let (following, more, _) = crate::kv_sync::kv_sync_page(db, &dense, &slash, 2, Some(&next)) + .await + .unwrap(); + assert_eq!( + following + .iter() + .map(|change| change.key.as_str()) + .collect::>(), + vec!["busy/150005", "busy/150006"] + ); + assert!(more); + let late = crate::kv_sync::KvSyncState { + pos: Some(dense_anchors[2]), + floor: Some(dense_anchors[3]), + }; + let (following, more, _) = crate::kv_sync::kv_sync_page(db, &dense, &bare, 2, Some(&late)) + .await + .unwrap(); + assert_eq!( + following + .iter() + .map(|change| change.key.as_str()) + .collect::>(), + vec!["busy/299901", "busy/299902"] + ); + assert!(more); + let caught_up = crate::kv_sync::KvSyncState { + pos: Some(dense_anchors[3]), + floor: Some(dense_anchors[3]), + }; + let (changes, more, next) = + crate::kv_sync::kv_sync_page(db, &dense, &slash, 500, Some(&caught_up)) + .await + .unwrap(); + assert!(changes.is_empty() && !more); + assert_eq!(next, caught_up); + // A qualifying key beyond a full 64-row window must still be found + // by the C-key fallback; the window is not a cursor advance. + let rare_next = crate::kv_sync::KvSyncAnchor { + seq: 300_011, + epoch: crate::hash::hash(b"tc737-skew-rare-next"), + epoch_seq: 0, + }; + epoch::ActiveModel { + seq: Set(rare_next.seq), + id: Set(rare_next.epoch), + space: Set(SpaceIdWrap(rare.clone())), + } + .insert(db) + .await + .unwrap(); + event_order::ActiveModel { + seq: Set(rare_next.seq), + epoch: Set(rare_next.epoch), + epoch_seq: Set(0), + event: Set(crate::hash::hash(b"tc737-skew-rare-next-event")), + space: Set(SpaceIdWrap(rare.clone())), + } + .insert(db) + .await + .unwrap(); + db.execute(Statement::from_sql_and_values( + DbBackend::Postgres, + "UPDATE current_kv SET seq = $2, epoch = $3 \ + WHERE space = $1 AND key = 'busy/only'", + [ + SpaceIdWrap(rare.clone()).into(), + rare_next.seq.into(), + rare_next.epoch.into(), + ], + )) + .await + .unwrap(); + for prefix in [&bare, &slash] { + let state = crate::kv_sync::KvSyncState { + pos: Some(pos), + floor: Some(floor), + }; + let (changes, more, next) = + crate::kv_sync::kv_sync_page(db, &rare, prefix, 500, Some(&state)) + .await + .unwrap(); + assert_eq!( + changes + .iter() + .map(|change| change.key.as_str()) + .collect::>(), + vec!["busy/only"] + ); + assert!(!more); + assert_eq!(next.pos, Some(rare_next)); + } assert_eq!( kv_list_pages(db, &rare, "busy", 1).await, vec![vec!["busy/only"]] diff --git a/tinycloud-core/src/kv_sync.rs b/tinycloud-core/src/kv_sync.rs index 33b86052..4d6d9393 100644 --- a/tinycloud-core/src/kv_sync.rs +++ b/tinycloud-core/src/kv_sync.rs @@ -684,8 +684,9 @@ pub(crate) enum KvPrefixProbe { Descendants, } -/// Seek keys in byte order before running the sequence-ordered feed. The -/// disjoint exact and descendant probes avoid the bare-prefix OR plan. +/// Seek keys in byte order when the sequence-index lookahead is full and +/// contains no scoped change. Separate exact/descendant queries avoid the +/// bare-prefix OR plan over an unbounded key range. pub(crate) fn kv_sync_probe_statement( backend: DbBackend, space: &SpaceId, @@ -723,8 +724,143 @@ pub(crate) fn kv_sync_probe_statement( backend.build(&query) } -/// Only called for a non-empty PostgreSQL prefix; other engines and the -/// whole-space feed retain their original single sequence-ordered query. +fn kv_sync_sequence_window( + backend: DbBackend, + space: &SpaceId, + pos: Option<&KvSyncAnchor>, + columns: &[current_kv::Column], + limit: u64, +) -> Statement { + let mut window = Query::select(); + window + .columns(columns.iter().cloned()) + .from(current_kv::Entity) + .and_where(col(current_kv::Column::Space).eq(SpaceIdWrap(space.clone()))); + if let Some(pos) = pos { + window + .and_where(col(current_kv::Column::Seq).gte(pos.seq)) + .and_where(position_columns().gt(position_value(pos))); + } + window + .order_by(current_kv::Column::Seq, Order::Asc) + .order_by(current_kv::Column::Epoch, Order::Asc) + .order_by(current_kv::Column::EpochSeq, Order::Asc) + .order_by(current_kv::Column::Key, Order::Asc) + .limit(limit); + backend.build(&window) +} + +fn kv_sync_window_filter( + prefix: &Path, + floor: Option<&KvSyncAnchor>, + values: &mut Vec, +) -> String { + let eligibility = floor.map(|floor| { + values.push(floor.seq.into()); + format!("(NOT deleted OR seq > ${})", values.len()) + }); + let (lower, upper, bare) = + crate::db::kv_prefix_bounds(prefix.as_str()).expect("window needs a prefix"); + values.push(lower.into()); + let lower = values.len(); + values.push(upper.into()); + let upper = values.len(); + let descendants = format!(r#"(key COLLATE "C" >= ${lower} AND key COLLATE "C" < ${upper})"#); + let scope = if bare { + values.push(prefix.as_str().into()); + format!(r#"(key COLLATE "C" = ${} OR {descendants})"#, values.len()) + } else { + descendants + }; + format!("({} AND {scope})", eligibility.as_deref().unwrap_or("true")) +} + +const KV_SYNC_CANDIDATE_WINDOW: u64 = 64; + +/// A bounded sequence-index lookahead before seeking the C-key index. Limit +/// raw rows *inside* the materialized CTE, then apply scope and floor outside +/// it: a dense prefix can resume cheaply, while a quiet prefix pays at most +/// one fixed-size window before the byte-key fallback. +pub(crate) fn kv_sync_candidate_statement( + backend: DbBackend, + space: &SpaceId, + prefix: &Path, + pos: Option<&KvSyncAnchor>, + floor: Option<&KvSyncAnchor>, +) -> Statement { + debug_assert_eq!(backend, DbBackend::Postgres); + let window = kv_sync_sequence_window( + backend, + space, + pos, + &[ + current_kv::Column::Key, + current_kv::Column::Deleted, + current_kv::Column::Seq, + ], + KV_SYNC_CANDIDATE_WINDOW, + ); + let mut values = window.values.map(|values| values.0).unwrap_or_default(); + let filter = kv_sync_window_filter(prefix, floor, &mut values); + Statement::from_sql_and_values( + backend, + format!( + "WITH candidates AS MATERIALIZED ({}) \ + SELECT count(*) AS seen, \ + COALESCE(bool_or({filter}), false) AS found \ + FROM candidates", + window.sql + ), + values, + ) +} + +/// A bounded feed for dense prefixes. If it yields a full page, or exhausts +/// the raw sequence window, its rows are the same next page as the normal +/// feed. Otherwise the caller falls back to the unbounded scoped query. +pub(crate) fn kv_sync_bounded_feed_statement( + backend: DbBackend, + space: &SpaceId, + prefix: &Path, + limit: usize, + pos: Option<&KvSyncAnchor>, + floor: Option<&KvSyncAnchor>, +) -> Statement { + debug_assert_eq!(backend, DbBackend::Postgres); + let window = kv_sync_sequence_window( + backend, + space, + pos, + &[ + current_kv::Column::Key, + current_kv::Column::Seq, + current_kv::Column::Epoch, + current_kv::Column::EpochSeq, + current_kv::Column::Value, + current_kv::Column::Metadata, + current_kv::Column::Deleted, + ], + limit.saturating_add(1) as u64, + ); + let mut values = window.values.map(|values| values.0).unwrap_or_default(); + let filter = kv_sync_window_filter(prefix, floor, &mut values); + Statement::from_sql_and_values( + backend, + format!( + "WITH candidates AS MATERIALIZED ({}) \ + SELECT key, seq, epoch, epoch_seq, value, metadata, deleted, \ + (SELECT count(*) FROM candidates) AS seen \ + FROM candidates WHERE {filter} \ + ORDER BY seq, epoch, epoch_seq", + window.sql + ), + values, + ) +} + +/// Only called for a non-empty PostgreSQL prefix; a bounded sequence-index +/// window handles dense continuations before the C-key seek handles quiet +/// prefixes. Other engines and the whole-space feed remain unchanged. async fn prefix_has_feed_rows( db: &C, space: &SpaceId, @@ -733,6 +869,19 @@ async fn prefix_has_feed_rows( floor: Option<&KvSyncAnchor>, ) -> Result { let backend = db.get_database_backend(); + let candidate = db + .query_one(kv_sync_candidate_statement( + backend, space, prefix, pos, floor, + )) + .await? + .ok_or_else(|| DbErr::Custom("missing KV sync candidate aggregate".into()))?; + let seen = candidate.try_get::("", "seen")?; + let found = candidate.try_get::("", "found")?; + if found || seen < KV_SYNC_CANDIDATE_WINDOW as i64 { + return Ok(found); + } + // A full window without a match cannot prove the rest of the space quiet. + // Seek only this prefix; never advance the cursor past other-space writes. if !prefix.as_str().ends_with('/') && db .query_one(kv_sync_probe_statement( @@ -888,11 +1037,38 @@ pub(crate) async fn kv_sync_page( { return Ok((Vec::new(), false, state)); } - let mut rows = feed_rows( - db, - kv_sync_statement(backend, space, prefix, limit, state.pos.as_ref(), floor), - ) - .await?; + let mut rows = if backend == DbBackend::Postgres && !prefix.as_str().is_empty() { + let window = db + .query_all(kv_sync_bounded_feed_statement( + backend, + space, + prefix, + limit, + state.pos.as_ref(), + floor, + )) + .await?; + let exhausted = window + .first() + .map(|row| row.try_get::("", "seen")) + .transpose()? + .is_some_and(|seen| seen < limit.saturating_add(1) as i64); + if window.len() > limit || exhausted { + window.into_iter().map(feed_row).collect::>()? + } else { + feed_rows( + db, + kv_sync_statement(backend, space, prefix, limit, state.pos.as_ref(), floor), + ) + .await? + } + } else { + feed_rows( + db, + kv_sync_statement(backend, space, prefix, limit, state.pos.as_ref(), floor), + ) + .await? + }; let more = if rows.len() <= limit { false } else {