diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9be0c4d0..b5a0d873 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -73,7 +73,7 @@ jobs: run: cargo run --manifest-path crates/graphql-orm/tests/fixtures/repository-aggregate-consumer/Cargo.toml --locked --no-default-features --features sqlite --example runtime_strings - name: Execute owned runtime migration preservation and compatibility run: | - cargo test -p graphql-orm --locked --no-default-features --features postgres --test runtime_migrations --test runtime_schema_ir --test runtime_migration_memory --test migration_planner --test migration_apply --test legacy_migration_history -- --test-threads=1 + cargo test -p graphql-orm --locked --no-default-features --features postgres --test runtime_mutations --test runtime_migrations --test runtime_schema_ir --test runtime_migration_memory --test migration_planner --test migration_apply --test legacy_migration_history -- --test-threads=1 cargo test -p graphql-orm --locked --no-default-features --features mssql --test runtime_migrations_unsupported for backend in sqlite postgres mssql; do cargo test --manifest-path crates/graphql-orm/tests/fixtures/migration-api-compatibility/Cargo.toml --locked --no-default-features --features "$backend" diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 941aefb4..f4e94c7c 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -155,7 +155,7 @@ jobs: cargo test -p graphql-orm --no-default-features --features sqlite,field-case-pascal,argument-case-snake --test federation_entity_keys_case_features --locked cargo check -p graphql-orm --no-default-features --features postgres --locked cargo test -p graphql-orm --no-default-features --features postgres --test federation_entity_keys_postgres --locked - cargo test -p graphql-orm --locked --no-default-features --features postgres --test runtime_migrations --test runtime_schema_ir --test runtime_migration_memory --test migration_planner --test migration_apply --test legacy_migration_history -- --test-threads=1 + cargo test -p graphql-orm --locked --no-default-features --features postgres --test runtime_mutations --test runtime_migrations --test runtime_schema_ir --test runtime_migration_memory --test migration_planner --test migration_apply --test legacy_migration_history -- --test-threads=1 cargo test -p graphql-orm --locked --no-default-features --features mssql --test runtime_migrations_unsupported for backend in sqlite postgres mssql; do cargo test --manifest-path crates/graphql-orm/tests/fixtures/migration-api-compatibility/Cargo.toml --locked --no-default-features --features "$backend" diff --git a/CHANGELOG.md b/CHANGELOG.md index 88ceb756..103e9d2e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,25 @@ This file is the authoritative user-facing release chronology. The former [release-notes ledger](docs/archive/2026/graphql-orm-release-notes.md) is retained for historical context. +## 0.41.0 - unreleased + +Companion macros crate: `graphql-orm-macros` **0.41.0** (alignment only). + +- Add schema-bound single-record runtime create, patch and exact-key delete on + SQLite/PostgreSQL pinned state-machine transactions. Reuse existing values, + handles, typed predicates and row decoding; preserve static repository APIs. +- Require explicit host intent/preimage/result authorization, structural CAS, + exact return projection grants and verified physical dependency environments. +- Add cancellation guards for runtime mutations and before-commit host journal + hooks. Caught errors/timeouts cannot commit partial runtime work. Static-only + transaction behavior and ambiguous commit limitations remain unchanged. +- Count backend-emitted predicate bindings before runtime target reads, including + duplicated SQLite suffix values and reused PostgreSQL native slots. +- Add conservative transaction commit-outcome inspection and safe runtime error + codes retained through the public transaction runner. +- Document unsupported modifying cascades and backend capabilities; include a + disposable create/update/delete/journal example and framework-neutral consumer. + ## 0.40.0 - unreleased Companion macros crate: `graphql-orm-macros` **0.40.0** (alignment only). diff --git a/Cargo.lock b/Cargo.lock index 84523071..9828a2f9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3146,7 +3146,7 @@ dependencies = [ [[package]] name = "graphql-orm" -version = "0.40.0" +version = "0.41.0" dependencies = [ "agql-auth", "async-graphql", @@ -3245,7 +3245,7 @@ dependencies = [ [[package]] name = "graphql-orm-macros" -version = "0.40.0" +version = "0.41.0" dependencies = [ "convert_case 0.7.1", "proc-macro2", diff --git a/Cargo.toml b/Cargo.toml index e629435a..37685b00 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -37,7 +37,7 @@ cynic-parser = { version = "=0.11.2", features = ["pretty"] } futures = "0.3" getrandom = "0.3" graphql-composition = "=0.12.2" -graphql-orm = { path = "crates/graphql-orm", version = "0.40.0", default-features = false } +graphql-orm = { path = "crates/graphql-orm", version = "0.41.0", default-features = false } graphql-orm-ai-tool-profiles = { path = "crates/graphql-orm-ai-tool-profiles", version = "0.15.0" } graphql-orm-backup = { path = "crates/graphql-orm-backup", version = "0.7.2", default-features = false } graphql-orm-operation-catalog = { path = "crates/graphql-orm-operation-catalog", version = "0.4.0" } diff --git a/MIGRATION.md b/MIGRATION.md index 857c4b09..ec280f7e 100644 --- a/MIGRATION.md +++ b/MIGRATION.md @@ -13,6 +13,18 @@ supersedes: [] `graphql-orm` is distributed from GitHub only. Use a reviewed full 40-character commit in `rev`; neither the runtime nor macros crate is published to crates.io. +## 0.40.0 to 0.41.0: transactional runtime mutations + +Adopt aligned ORM/macros 0.41.0. This is an additive, unpublished successor to A; +no stored-data migration, macro syntax or static GraphQL SDL change is required. +Use `TransactionMode::StateMachine`, a verified runtime mutation environment and +explicit host authority. The host pins its complete policy revision and external +DDL fence. Runtime mutation errors/cancellation now poison only transactions that +perform runtime work; static-only timeout behavior is unchanged. Use the guarded +before-commit hook for atomic host journal writes. Effects remain pending until +the runner commits. See [runtime mutations](docs/reference/graphql-orm/runtime-mutations.md) +for unsupported capabilities, exact omission/null semantics and ambiguous commits. + ## 0.39.0 to 0.40.0: owned runtime migration targets Adopt runtime and macros 0.40.0 together. This unpublished identity retains the released diff --git a/crates/graphql-orm-macros/Cargo.toml b/crates/graphql-orm-macros/Cargo.toml index aae982e7..55482615 100644 --- a/crates/graphql-orm-macros/Cargo.toml +++ b/crates/graphql-orm-macros/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "graphql-orm-macros" -version = "0.40.0" +version = "0.41.0" edition = "2024" authors = ["Toby Martin"] description = "Procedural macros for async-graphql and ORM-backed entities, relations, and CRUD operations." diff --git a/crates/graphql-orm-macros/README.md b/crates/graphql-orm-macros/README.md index 4c115919..feacd913 100644 --- a/crates/graphql-orm-macros/README.md +++ b/crates/graphql-orm-macros/README.md @@ -16,13 +16,13 @@ macro/runtime versions aligned: ```toml [dependencies] -graphql-orm = { git = "https://github.com/Dastari/graphql-orm.git", rev = "", version = "0.40.0", default-features = false, features = ["sqlite"] } +graphql-orm = { git = "https://github.com/Dastari/graphql-orm.git", rev = "", version = "0.41.0", default-features = false, features = ["sqlite"] } ``` Direct use is supported for tooling that needs the macro package: ```toml -graphql-orm-macros = { git = "https://github.com/Dastari/graphql-orm.git", rev = "", version = "0.40.0", default-features = false, features = ["sqlite"] } +graphql-orm-macros = { git = "https://github.com/Dastari/graphql-orm.git", rev = "", version = "0.41.0", default-features = false, features = ["sqlite"] } ``` The direct dependency still requires a compatible `graphql-orm` runtime in the @@ -183,6 +183,6 @@ See [host timestamp inputs, migration compatibility and cancellation limits](../ Complete SQLite/PostgreSQL text-group pages and SQL-visible ordinary aggregates: see [typed aggregates](../../docs/reference/graphql-orm/typed-aggregates.md). -Version 0.40.0 aligns with the owned runtime migration addition. Released host-managed +Version 0.41.0 aligns with the schema-bound transactional runtime mutation addition. Released host-managed timestamps, framework-neutral repository aggregate helpers and static GraphQL SDL remain unchanged. diff --git a/crates/graphql-orm-macros/fixtures/cross-crate-relations/Cargo.lock b/crates/graphql-orm-macros/fixtures/cross-crate-relations/Cargo.lock index b47ce13c..11e35273 100644 --- a/crates/graphql-orm-macros/fixtures/cross-crate-relations/Cargo.lock +++ b/crates/graphql-orm-macros/fixtures/cross-crate-relations/Cargo.lock @@ -1028,7 +1028,7 @@ dependencies = [ [[package]] name = "graphql-orm" -version = "0.40.0" +version = "0.41.0" dependencies = [ "async-graphql", "chacha20poly1305", @@ -1054,7 +1054,7 @@ dependencies = [ [[package]] name = "graphql-orm-macros" -version = "0.40.0" +version = "0.41.0" dependencies = [ "convert_case", "proc-macro2", diff --git a/crates/graphql-orm/Cargo.toml b/crates/graphql-orm/Cargo.toml index 9603b9d7..c4d4f496 100644 --- a/crates/graphql-orm/Cargo.toml +++ b/crates/graphql-orm/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "graphql-orm" -version = "0.40.0" +version = "0.41.0" edition = "2024" description = "Runtime support crate for graphql-orm-macros" license = "MIT" @@ -68,7 +68,7 @@ futures = "0.3" geo = { version = "0.33", optional = true, default-features = false } geo-types = { version = "0.7", optional = true } geojson = { version = "1", optional = true, default-features = true } -graphql-orm-macros = { path = "../graphql-orm-macros", version = "0.40.0", default-features = false } +graphql-orm-macros = { path = "../graphql-orm-macros", version = "0.41.0", default-features = false } graphql-orm-operation-catalog = { workspace = true } rust_decimal = { workspace = true } serde = { version = "1", features = ["derive"] } diff --git a/crates/graphql-orm/README.md b/crates/graphql-orm/README.md index 289c5cde..2ecf626c 100644 --- a/crates/graphql-orm/README.md +++ b/crates/graphql-orm/README.md @@ -29,7 +29,7 @@ backend: ```toml [dependencies] -graphql-orm = { git = "https://github.com/Dastari/graphql-orm.git", rev = "", version = "0.40.0", default-features = false, features = ["sqlite"] } +graphql-orm = { git = "https://github.com/Dastari/graphql-orm.git", rev = "", version = "0.41.0", default-features = false, features = ["sqlite"] } ``` This unpublished package has no docs.rs release. Use this Git README and the @@ -204,3 +204,7 @@ are documented in [runtime migrations](../../docs/reference/graphql-orm/runtime- Live capability checks reject FK actions/deferral the physical model cannot retain, including incoming system/unowned constraints, and recheck on the apply transaction. Static migration APIs remain unchanged. + +Schema-bound runtime create/patch/delete with pinned authority, structural CAS, +atomic host journals and cancellation poisoning are documented in +[runtime mutations](../../docs/reference/graphql-orm/runtime-mutations.md). diff --git a/crates/graphql-orm/examples/runtime_mutation_journal.rs b/crates/graphql-orm/examples/runtime_mutation_journal.rs new file mode 100644 index 00000000..5ff5e180 --- /dev/null +++ b/crates/graphql-orm/examples/runtime_mutation_journal.rs @@ -0,0 +1,169 @@ +//! Owned disposable runtime mutation + repository journal example. +#[cfg(any( + all(feature = "sqlite", not(any(feature = "postgres", feature = "mssql"))), + all(feature = "postgres", not(any(feature = "sqlite", feature = "mssql"))) +))] +mod example { + use graphql_orm::futures::future::BoxFuture; + use graphql_orm::{graphql::orm::*, prelude::*}; + use std::sync::Arc; + #[cfg(feature = "sqlite")] + type Backend = SqliteBackend; + #[cfg(feature = "postgres")] + type Backend = PostgresBackend; + #[derive(RepositoryEntity, Clone, serde::Serialize, serde::Deserialize)] + #[repository_entity(table = "example_journal", plural = "ExampleJournal")] + struct Journal { + #[primary_key] + #[graphql_orm(auto_generated = false)] + id: String, + action: String, + } + #[derive(GraphQLSchemaEntity, Clone, serde::Serialize, serde::Deserialize)] + #[graphql_entity(table = "example_records", plural = "ExampleRecords")] + struct Record { + #[primary_key] + #[graphql_orm(auto_generated = false)] + id: i64, + #[filterable(type = "string")] + label: String, + } + // Explicit public authority for this isolated example. A real host validates its + // pinned identity, policy/public revision, input/CAS fields and return fields here. + struct PublicAuthority(RuntimeProjection); + impl RuntimeWriteAuthority for PublicAuthority { + fn authorize_intent<'a>( + &'a self, + _: RuntimeWriteIntent<'a>, + _: &'a mut MutationContext<'_, Backend>, + ) -> BoxFuture<'a, Result> { + Box::pin(async move { RuntimeWriteGrant::new(self.0.clone(), None) }) + } + fn authorize_preimage<'a>( + &'a self, + _: RuntimePreimageCheck<'a>, + _: &'a mut MutationContext<'_, Backend>, + ) -> BoxFuture<'a, Result<(), RuntimeMutationError>> { + Box::pin(async { Ok(()) }) + } + fn authorize_result<'a>( + &'a self, + check: RuntimeResultCheck<'a>, + _: &'a mut MutationContext<'_, Backend>, + ) -> BoxFuture<'a, Result> { + Box::pin(async move { Ok(RuntimeReturnGrant::new(check.intent.returning().cloned())) }) + } + } + pub async fn run(db: Database) -> Result<(), Box> { + let schema = + Arc::new(RuntimeSchema::from_static_entities(&[Record::metadata()])?.validate()?); + let target = schema + .physical_schema::(Default::default())? + .with_static_entities(&[Journal::metadata()])?; + let ownership = ManagedTableSet::new(["example_records".into(), "example_journal".into()])?; + let plan = db + .schema() + .plan_owned_migration( + "example", + "isolated example", + &target, + &ownership, + PlanOptions::strict(), + ) + .await?; + db.schema() + .apply_owned_migration(&plan, ApplyOptions::default()) + .await?; + let environment = Arc::new( + db.schema() + .runtime_mutation_environment(schema.clone(), &target, &ownership) + .await?, + ); + let collection = schema.resolve_collection(&schema.schema().collections[0].id)?; + let id = schema.resolve_field(&collection, &FieldId::new("example_records.id")?)?; + let label = schema.resolve_field(&collection, &FieldId::new("example_records.label")?)?; + let projection = schema.resolve_projection(&collection, &[id.clone(), label.clone()])?; + let expected = schema.runtime_compare( + &collection, + &label, + RuntimeScalarOperator::Eq, + RuntimeValue::String("draft".into()), + Default::default(), + )?; + let requests = [ + schema.runtime_create_request( + &collection, + &[ + (id.clone(), RuntimeValue::Integer(7)), + (label.clone(), RuntimeValue::String("draft".into())), + ], + Some(projection.clone()), + Default::default(), + )?, + schema.runtime_update_request( + schema.runtime_key(&collection, &[(id.clone(), RuntimeValue::Integer(7))])?, + &[(label, RuntimeValue::String("published".into()))], + Some(expected), + Some(projection.clone()), + Default::default(), + )?, + schema.runtime_delete_request( + schema.runtime_key(&collection, &[(id, RuntimeValue::Integer(7))])?, + None, + Some(projection.clone()), + Default::default(), + )?, + ]; + for (index, request) in requests.into_iter().enumerate() { + let environment = environment.clone(); + let authority = PublicAuthority(projection.clone()); + let committed = db + .transaction(TransactionMode::StateMachine, move |tx| { + Box::pin(async move { + let pending = tx + .mutate_runtime(&environment, &request, &authority) + .await?; + tx.insert::(CreateJournalInput { + id: index.to_string(), + action: format!("{:?}", pending.action()), + }) + .await?; + Ok(pending) + }) + }) + .await?; + println!("Committed {:?}", committed.action()); + } + assert_eq!(Journal::count_all(&db).await?, 3); + db.pool().close().await; + Ok(()) + } +} +#[cfg(all(feature = "postgres", not(any(feature = "sqlite", feature = "mssql"))))] +#[path = "../tests/support/owned_postgres.rs"] +mod owned_postgres; +#[tokio::main] +pub async fn main() -> Result<(), Box> { + #[cfg(all(feature = "sqlite", not(any(feature = "postgres", feature = "mssql"))))] + example::run(graphql_orm::db::Database::connect_sqlite("sqlite::memory:").await?).await?; + #[cfg(all(feature = "postgres", not(any(feature = "sqlite", feature = "mssql"))))] + { + let mut owned = owned_postgres::OwnedPostgres::start("runtime-journal-example")?; + example::run(graphql_orm::db::Database::connect_postgres(&owned.url).await?).await?; + owned.cleanup()?; + } + #[cfg(any( + all(feature = "sqlite", not(any(feature = "postgres", feature = "mssql"))), + all(feature = "postgres", not(any(feature = "sqlite", feature = "mssql"))) + ))] + { + Ok(()) + } + #[cfg(not(any( + all(feature = "sqlite", not(any(feature = "postgres", feature = "mssql"))), + all(feature = "postgres", not(any(feature = "sqlite", feature = "mssql"))) + )))] + { + Err("runtime mutation example requires exactly SQLite or PostgreSQL".into()) + } +} diff --git a/crates/graphql-orm/src/db.rs b/crates/graphql-orm/src/db.rs index 040b64ee..5667e4b4 100644 --- a/crates/graphql-orm/src/db.rs +++ b/crates/graphql-orm/src/db.rs @@ -1485,6 +1485,18 @@ where return Err(classified); } }; + if context.runtime_cannot_commit() { + context + .rollback() + .await + .map_err(classify_transaction_error::)?; + return Err(crate::graphql::orm::TransactionError::Rejected( + OrmPublicError::with_message( + OrmErrorCode::Conflict, + "unfinished or failed runtime work requires rollback", + ), + )); + } context .commit_and_emit() .await diff --git a/crates/graphql-orm/src/graphql/errors.rs b/crates/graphql-orm/src/graphql/errors.rs index 91f6b9ec..cbc6cea5 100644 --- a/crates/graphql-orm/src/graphql/errors.rs +++ b/crates/graphql-orm/src/graphql/errors.rs @@ -88,6 +88,7 @@ pub struct OrmPublicError { /// Internal diagnostic text never exposed by default GraphQL extensions. pub(crate) internal: Option, pub(crate) retryable: bool, + pub(crate) runtime_mutation_code: Option<&'static str>, } impl fmt::Debug for OrmPublicError { @@ -125,6 +126,7 @@ impl OrmPublicError { correlation_id: None, internal: None, retryable: false, + runtime_mutation_code: None, } } @@ -136,6 +138,7 @@ impl OrmPublicError { correlation_id: None, internal: None, retryable: false, + runtime_mutation_code: None, } } @@ -151,6 +154,11 @@ impl OrmPublicError { self } + /// Safe runtime mutation code retained through the transaction runner, when applicable. + pub fn runtime_mutation_code(&self) -> Option<&'static str> { + self.runtime_mutation_code + } + /// Whether retrying the complete transaction is safe and recommended. pub const fn is_retryable(&self) -> bool { self.retryable diff --git a/crates/graphql-orm/src/graphql/orm/core.rs b/crates/graphql-orm/src/graphql/orm/core.rs index d0d49673..7baae4ac 100644 --- a/crates/graphql-orm/src/graphql/orm/core.rs +++ b/crates/graphql-orm/src/graphql/orm/core.rs @@ -674,7 +674,24 @@ pub struct RetentionPurgeEvent { pub affected: u32, } +/// Conservative commit outcome; unknown outcomes must not be blindly retried. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum TransactionCommitOutcome { + /// The runner confirmed rollback or rejected before transaction entry. + RolledBack, + /// Commit or rollback completion is not established by this error variant. + Unknown, +} + impl TransactionError { + /// Whether rollback is established; preserves the existing error variants. + pub const fn commit_outcome(&self) -> TransactionCommitOutcome { + match self { + Self::Rejected(_) => TransactionCommitOutcome::RolledBack, + Self::Retryable(_) | Self::Failed(_) => TransactionCommitOutcome::Unknown, + } + } + pub fn public_error(&self) -> &crate::graphql::errors::OrmPublicError { match self { Self::Rejected(error) | Self::Retryable(error) | Self::Failed(error) => error, @@ -829,6 +846,7 @@ pub struct MutationContext<'tx, B: WriteBackend = DefaultWriteBackend> { db: &'tx crate::db::Database, tx: B::Transaction<'tx>, mode: TransactionMode, + pub(crate) runtime_state: std::sync::Arc, deferred_events: Vec>>, deferred_actions: Vec>>, } @@ -1255,6 +1273,7 @@ where db, tx, mode, + runtime_state: Default::default(), deferred_events: Vec::new(), deferred_actions: Vec::new(), } @@ -1404,11 +1423,25 @@ where } } + pub(crate) fn runtime_cannot_commit(&self) -> bool { + self.runtime_state.cannot_commit() + } + pub async fn commit_and_emit(self) -> crate::Result<()> { + if self.runtime_cannot_commit() { + self.rollback().await?; + return Err(crate::graphql::errors::sqlx_error_from_public( + crate::graphql::errors::OrmPublicError::with_message( + crate::graphql::errors::OrmErrorCode::Conflict, + "unfinished or failed runtime work requires rollback", + ), + )); + } let Self { db, tx, mode: _, + runtime_state: _, deferred_events, deferred_actions, } = self; diff --git a/crates/graphql-orm/src/graphql/orm/mod.rs b/crates/graphql-orm/src/graphql/orm/mod.rs index d03887d2..b5bad81b 100644 --- a/crates/graphql-orm/src/graphql/orm/mod.rs +++ b/crates/graphql-orm/src/graphql/orm/mod.rs @@ -18,10 +18,14 @@ mod owned_schema; mod query; mod rls; mod runtime_migration; +#[cfg(any(feature = "sqlite", feature = "postgres", feature = "mssql"))] +mod runtime_mutation; mod runtime_query; mod runtime_record; mod runtime_relation; mod runtime_schema; +#[cfg(any(feature = "sqlite", feature = "postgres", feature = "mssql"))] +mod runtime_transaction; mod schema_manager; mod schema_module; mod search; @@ -45,6 +49,8 @@ pub use owned_schema::{IndexModel, IndexPredicateModel, OwnedSchemaModel, OwnedT pub use query::*; pub use rls::*; pub use runtime_migration::*; +#[cfg(any(feature = "sqlite", feature = "postgres", feature = "mssql"))] +pub use runtime_mutation::*; pub use runtime_query::*; pub use runtime_record::*; pub use runtime_relation::*; diff --git a/crates/graphql-orm/src/graphql/orm/runtime_migration.rs b/crates/graphql-orm/src/graphql/orm/runtime_migration.rs index 857392cd..cdb36116 100644 --- a/crates/graphql-orm/src/graphql/orm/runtime_migration.rs +++ b/crates/graphql-orm/src/graphql/orm/runtime_migration.rs @@ -1119,6 +1119,7 @@ impl<'db, B: RuntimeMigrationBackend> SchemaManager<'db, B> { } } Ok(RuntimeMutationEnvironment { + backend: B::DIALECT, schema, physical_baseline_hash: current.stable_hash(), ownership: ownership.clone(), @@ -1235,12 +1236,17 @@ pub struct VerifiedRuntimeDependency { /// Immutable schema-bound physical dependency certificate, not another schema IR. #[derive(Clone, Debug)] pub struct RuntimeMutationEnvironment { + backend: DatabaseBackend, schema: Arc, physical_baseline_hash: String, ownership: ManagedTableSet, incoming: BTreeMap>, } impl RuntimeMutationEnvironment { + /// Backend which verified this physical dependency certificate. + pub fn backend(&self) -> DatabaseBackend { + self.backend + } /// Validated schema pinned by the host generation. pub fn schema(&self) -> &Arc { &self.schema diff --git a/crates/graphql-orm/src/graphql/orm/runtime_mutation.rs b/crates/graphql-orm/src/graphql/orm/runtime_mutation.rs new file mode 100644 index 00000000..5dd1f3e3 --- /dev/null +++ b/crates/graphql-orm/src/graphql/orm/runtime_mutation.rs @@ -0,0 +1,949 @@ +//! Schema-bound, single-record writes on an ORM-owned pinned transaction. +use super::runtime_query::{placeholder, value_bind}; +use super::*; +use crate::graphql::errors::{OrmErrorCode, OrmPublicError}; +use futures::future::BoxFuture; +use std::{collections::BTreeSet, fmt, sync::Arc}; + +/// Safe runtime mutation category. No variant contains caller values or SQL. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub enum RuntimeMutationErrorCode { + InvalidInput, + MissingRequired, + FieldDenied, + Denied, + NotFound, + Conflict, + ConstraintViolation, + AppendOnly, + UnsupportedBackend, + UnsupportedGeneration, + UnsupportedCascade, + SchemaMismatch, + LimitExceeded, + RetryableTransaction, + HookFailed, + DatabaseFailed, + CommitUnknown, +} +impl RuntimeMutationErrorCode { + /// Stable machine code. + pub const fn as_str(self) -> &'static str { + match self { + Self::InvalidInput => "invalid_input", + Self::MissingRequired => "missing_required", + Self::FieldDenied => "field_denied", + Self::Denied => "denied", + Self::NotFound => "not_found", + Self::Conflict => "conflict", + Self::ConstraintViolation => "constraint_violation", + Self::AppendOnly => "append_only", + Self::UnsupportedBackend => "unsupported_backend", + Self::UnsupportedGeneration => "unsupported_generation", + Self::UnsupportedCascade => "unsupported_cascade", + Self::SchemaMismatch => "schema_mismatch", + Self::LimitExceeded => "limit_exceeded", + Self::RetryableTransaction => "retryable_transaction", + Self::HookFailed => "hook_failed", + Self::DatabaseFailed => "database_failed", + Self::CommitUnknown => "commit_unknown", + } + } +} +/// Redacted mutation failure. The source is available only through trusted error inspection. +pub struct RuntimeMutationError { + code: RuntimeMutationErrorCode, + source: Option>, +} +impl RuntimeMutationError { + /// Construct an explicit fail-closed host decision. + pub fn new(code: RuntimeMutationErrorCode) -> Self { + Self { code, source: None } + } + /// Safe category. + pub fn code(&self) -> RuntimeMutationErrorCode { + self.code + } + /// Retain a trusted diagnostic source without exposing it in formatting. + pub fn with_source(mut self, source: impl std::error::Error + Send + Sync + 'static) -> Self { + self.source = Some(Box::new(source)); + self + } +} +impl fmt::Debug for RuntimeMutationError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_tuple("RuntimeMutationError") + .field(&self.code) + .finish() + } +} +impl fmt::Display for RuntimeMutationError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.code.as_str()) + } +} +impl std::error::Error for RuntimeMutationError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + self.source.as_deref().map(|e| e as _) + } +} +impl From for RuntimeMutationError { + fn from(e: RuntimeRecordError) -> Self { + let code = if e.code() == RuntimeRecordErrorCode::SchemaMismatch { + RuntimeMutationErrorCode::SchemaMismatch + } else { + RuntimeMutationErrorCode::InvalidInput + }; + Self::new(code).with_source(e) + } +} +impl From for OrmPublicError { + fn from(e: RuntimeMutationError) -> Self { + use RuntimeMutationErrorCode as C; + let code = match e.code { + C::Denied | C::FieldDenied => OrmErrorCode::Forbidden, + C::NotFound => OrmErrorCode::NotFound, + C::Conflict => OrmErrorCode::Conflict, + C::ConstraintViolation => OrmErrorCode::ConstraintViolation, + C::RetryableTransaction => OrmErrorCode::ServiceUnavailable, + C::DatabaseFailed | C::CommitUnknown | C::HookFailed => OrmErrorCode::InternalError, + _ => OrmErrorCode::InvalidInput, + }; + let mut public = Self::with_message(code, e.code.as_str()); + public.runtime_mutation_code = Some(e.code.as_str()); + public.retryable = e.code == C::RetryableTransaction; + public + } +} +fn err(code: RuntimeMutationErrorCode) -> RuntimeMutationError { + RuntimeMutationError::new(code) +} +fn database_error(e: sqlx::Error) -> RuntimeMutationError { + let code = if B::is_retryable_write_error(&e) { + RuntimeMutationErrorCode::RetryableTransaction + } else if OrmPublicError::from_sqlx(&e).code == OrmErrorCode::ConstraintViolation { + RuntimeMutationErrorCode::ConstraintViolation + } else { + RuntimeMutationErrorCode::DatabaseFailed + }; + err(code).with_source(e) +} + +/// Bounds enforced before protected I/O and when decoding authoritative records. +#[derive(Clone, Copy, Debug)] +pub struct RuntimeMutationLimits { + pub max_input_fields: usize, + pub max_key_fields: usize, + pub max_value_bytes: usize, + pub max_record_bytes: usize, + pub max_json_depth: usize, + pub query: RuntimeQueryLimits, +} +impl Default for RuntimeMutationLimits { + fn default() -> Self { + Self { + max_input_fields: 128, + max_key_fields: 16, + max_value_bytes: 1024 * 1024, + max_record_bytes: 4 * 1024 * 1024, + max_json_depth: 32, + query: RuntimeQueryLimits::default(), + } + } +} +/// Exact operation; no bulk, nested writes or implicit upsert. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum RuntimeMutationAction { + Create, + Update, + Delete, +} +/// Complete, non-null primary key bound to a validated schema. +#[derive(Clone)] +pub struct RuntimeKey { + schema: Arc, + collection: RuntimeCollectionHandle, + fields: Vec<(RuntimeFieldHandle, RuntimeValue)>, +} +/// Immutable validated mutation request. Omission is distinct from SQL NULL. +#[derive(Clone)] +pub struct RuntimeMutationRequest { + schema: Arc, + collection: RuntimeCollectionHandle, + action: RuntimeMutationAction, + fields: Vec<(RuntimeFieldHandle, RuntimeValue)>, + key: Option, + expected: Option, + returning: Option, + limits: RuntimeMutationLimits, +} +macro_rules! redacted_debug { + ($($ty:ty),+ $(,)?) => {$(impl fmt::Debug for $ty { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.write_str(concat!(stringify!($ty), "([redacted])")) } + })+}; +} +redacted_debug!(RuntimeKey, RuntimeMutationRequest); +impl RuntimeKey { + /// Complete key members in declared primary-key order; trusted host use only. + pub fn fields(&self) -> &[(RuntimeFieldHandle, RuntimeValue)] { + &self.fields + } + pub fn collection(&self) -> &RuntimeCollectionHandle { + &self.collection + } +} +/// Borrowed intent. Hosts must validate their complete policy/public revision here. +#[derive(Clone, Copy)] +pub struct RuntimeWriteIntent<'a> { + request: &'a RuntimeMutationRequest, +} +impl<'a> RuntimeWriteIntent<'a> { + pub fn action(self) -> RuntimeMutationAction { + self.request.action + } + pub fn collection(self) -> &'a RuntimeCollectionHandle { + &self.request.collection + } + pub fn fields(self) -> &'a [(RuntimeFieldHandle, RuntimeValue)] { + &self.request.fields + } + pub fn key(self) -> Option<&'a RuntimeKey> { + self.request.key.as_ref() + } + pub fn expected(self) -> Option<&'a RuntimePredicate> { + self.request.expected.as_ref() + } + pub fn returning(self) -> Option<&'a RuntimeProjection> { + self.request.returning.as_ref() + } +} +/// Authoritative preimage from the pinned transaction, never a caller preview. +pub struct RuntimePreimageCheck<'a> { + pub intent: RuntimeWriteIntent<'a>, + pub record: &'a RuntimeRecord, +} +/// Actual staged result, including defaults; delete uses its authorized preimage. +pub struct RuntimeResultCheck<'a> { + pub intent: RuntimeWriteIntent<'a>, + pub preimage: Option<&'a RuntimeRecord>, + pub record: &'a RuntimeRecord, +} +redacted_debug!( + RuntimeWriteIntent<'_>, + RuntimePreimageCheck<'_>, + RuntimeResultCheck<'_> +); +/// Explicit approval of the entire intent, internal policy projection and row restriction. +pub struct RuntimeWriteGrant { + projection: RuntimeProjection, + predicate: Option, +} +impl RuntimeWriteGrant { + /// Reject cross-schema/collection predicates; this does not narrow input approval. + pub fn new( + projection: RuntimeProjection, + predicate: Option, + ) -> Result { + if predicate.as_ref().is_some_and(|p| { + !p.belongs_to( + projection.schema_fingerprint(), + projection.collection().id(), + ) + }) { + return Err(err(RuntimeMutationErrorCode::SchemaMismatch)); + } + Ok(Self { + projection, + predicate, + }) + } +} +/// Explicit final-row approval and exact authorized return projection. +pub struct RuntimeReturnGrant { + projection: Option, +} +impl RuntimeReturnGrant { + pub fn new(projection: Option) -> Self { + Self { projection } + } +} +redacted_debug!(RuntimeWriteGrant, RuntimeReturnGrant); +/// Required host authority; there is no default allow implementation. +pub trait RuntimeWriteAuthority: Send + Sync { + fn authorize_intent<'a>( + &'a self, + check: RuntimeWriteIntent<'a>, + tx: &'a mut MutationContext<'_, B>, + ) -> BoxFuture<'a, Result>; + fn authorize_preimage<'a>( + &'a self, + check: RuntimePreimageCheck<'a>, + tx: &'a mut MutationContext<'_, B>, + ) -> BoxFuture<'a, Result<(), RuntimeMutationError>>; + fn authorize_result<'a>( + &'a self, + check: RuntimeResultCheck<'a>, + tx: &'a mut MutationContext<'_, B>, + ) -> BoxFuture<'a, Result>; +} +/// Pending work, not committed success or an event. Only the runner establishes commit. +pub struct RuntimeMutationEffect { + action: RuntimeMutationAction, + record: Option, +} +impl RuntimeMutationEffect { + pub fn action(&self) -> RuntimeMutationAction { + self.action + } + pub fn record(&self) -> Option<&RuntimeRecord> { + self.record.as_ref() + } +} +redacted_debug!(RuntimeMutationEffect); +/// Atomic host journal hook. Invoke only through `run_runtime_before_commit`. +pub trait RuntimeMutationHook: Send + Sync { + fn before_commit<'a>( + &'a self, + tx: &'a mut MutationContext<'_, B>, + pending: &'a RuntimeMutationEffect, + ) -> BoxFuture<'a, Result<(), RuntimeMutationError>>; +} +/// Additive opt-in capability; existing third-party backends gain no required methods. +pub trait RuntimeMutationBackend: TransactionBackend + RuntimeRowDecoder { + const RUNTIME_MUTATIONS_SUPPORTED: bool; +} +#[cfg(feature = "sqlite")] +impl RuntimeMutationBackend for SqliteBackend { + const RUNTIME_MUTATIONS_SUPPORTED: bool = true; +} +#[cfg(feature = "postgres")] +impl RuntimeMutationBackend for PostgresBackend { + const RUNTIME_MUTATIONS_SUPPORTED: bool = true; +} +#[cfg(feature = "mssql")] +impl RuntimeMutationBackend for MssqlBackend { + const RUNTIME_MUTATIONS_SUPPORTED: bool = false; +} + +fn collection<'a>( + schema: &'a ValidatedRuntimeSchema, + handle: &RuntimeCollectionHandle, +) -> Result<&'a RuntimeCollection, RuntimeMutationError> { + if &schema.fingerprint() != handle.schema_fingerprint() { + return Err(err(RuntimeMutationErrorCode::SchemaMismatch)); + } + schema + .schema() + .collections + .iter() + .find(|c| &c.id == handle.id()) + .ok_or_else(|| err(RuntimeMutationErrorCode::SchemaMismatch)) +} +pub(super) fn check_value( + field: &RuntimeFieldHandle, + value: &RuntimeValue, + limits: RuntimeMutationLimits, +) -> Result<(), RuntimeMutationError> { + if matches!(value, RuntimeValue::Null) { + if !field.nullable() { + return Err(err(RuntimeMutationErrorCode::InvalidInput)); + } + } else if value.kind() != Some(field.value_kind()) { + return Err(err(RuntimeMutationErrorCode::InvalidInput)); + } + if let RuntimeValue::Json(value) = value { + let mut stack = vec![(value, 1usize)]; + while let Some((node, depth)) = stack.pop() { + if depth > limits.max_json_depth { + return Err(err(RuntimeMutationErrorCode::LimitExceeded)); + } + match node { + serde_json::Value::Array(a) => stack.extend(a.iter().map(|v| (v, depth + 1))), + serde_json::Value::Object(o) => stack.extend(o.values().map(|v| (v, depth + 1))), + _ => {} + } + } + } + let bytes = + serde_json::to_vec(value).map_err(|_| err(RuntimeMutationErrorCode::InvalidInput))?; + if bytes.len() > limits.max_value_bytes { + return Err(err(RuntimeMutationErrorCode::LimitExceeded)); + } + Ok(()) +} +impl ValidatedRuntimeSchema { + /// Validate all primary-key members, preserving declared order. + pub fn runtime_key( + &self, + handle: &RuntimeCollectionHandle, + fields: &[(RuntimeFieldHandle, RuntimeValue)], + ) -> Result { + let c = collection(self, handle)?; + if fields.len() != c.primary_key.len() { + return Err(err(RuntimeMutationErrorCode::InvalidInput)); + } + let mut ordered = Vec::with_capacity(fields.len()); + for id in &c.primary_key { + let expected = self.resolve_field(handle, id)?; + let matches = fields + .iter() + .filter(|(f, _)| f == &expected) + .collect::>(); + if matches.len() != 1 { + return Err(err(RuntimeMutationErrorCode::InvalidInput)); + } + check_value(&expected, &matches[0].1, RuntimeMutationLimits::default())?; + ordered.push(matches[0].clone()); + } + Ok(RuntimeKey { + schema: Arc::new(self.clone()), + collection: handle.clone(), + fields: ordered, + }) + } + /// Create input: omitted defaults/generators are evaluated at write time. + pub fn runtime_create_request( + &self, + collection: &RuntimeCollectionHandle, + fields: &[(RuntimeFieldHandle, RuntimeValue)], + returning: Option, + limits: RuntimeMutationLimits, + ) -> Result { + self.mutation_request( + collection, + RuntimeMutationAction::Create, + fields, + None, + None, + returning, + limits, + ) + } + /// Patch input: omitted fields remain unchanged; keys are immutable. + pub fn runtime_update_request( + &self, + key: RuntimeKey, + fields: &[(RuntimeFieldHandle, RuntimeValue)], + expected: Option, + returning: Option, + limits: RuntimeMutationLimits, + ) -> Result { + let collection = key.collection.clone(); + self.mutation_request( + &collection, + RuntimeMutationAction::Update, + fields, + Some(key), + expected, + returning, + limits, + ) + } + /// Delete exactly one complete key; incoming modifying actions are rejected at execution. + pub fn runtime_delete_request( + &self, + key: RuntimeKey, + expected: Option, + returning: Option, + limits: RuntimeMutationLimits, + ) -> Result { + let collection = key.collection.clone(); + self.mutation_request( + &collection, + RuntimeMutationAction::Delete, + &[], + Some(key), + expected, + returning, + limits, + ) + } + #[allow(clippy::too_many_arguments)] + fn mutation_request( + &self, + handle: &RuntimeCollectionHandle, + action: RuntimeMutationAction, + fields: &[(RuntimeFieldHandle, RuntimeValue)], + key: Option, + expected: Option, + returning: Option, + limits: RuntimeMutationLimits, + ) -> Result { + let c = collection(self, handle)?; + if c.append_only && action != RuntimeMutationAction::Create { + return Err(err(RuntimeMutationErrorCode::AppendOnly)); + } + if fields.len() > limits.max_input_fields || c.primary_key.len() > limits.max_key_fields { + return Err(err(RuntimeMutationErrorCode::LimitExceeded)); + } + if action == RuntimeMutationAction::Update && fields.is_empty() { + return Err(err(RuntimeMutationErrorCode::InvalidInput)); + } + if key + .as_ref() + .is_some_and(|k| k.schema.fingerprint() != self.fingerprint()) + || expected + .as_ref() + .is_some_and(|p| !p.belongs_to(&self.fingerprint(), &c.id)) + { + return Err(err(RuntimeMutationErrorCode::SchemaMismatch)); + } + if let Some(p) = &returning { + self.resolve_projection(handle, p.fields())?; + } + if returning + .as_ref() + .is_some_and(|p| p.fields().len() > limits.query.max_projection_fields) + { + return Err(err(RuntimeMutationErrorCode::LimitExceeded)); + } + if let Some(p) = &expected { + p.check_mutation_limits(limits)?; + } + let mut seen = BTreeSet::new(); + for (field, value) in fields { + if self.resolve_field(handle, field.id())? != *field { + return Err(err(RuntimeMutationErrorCode::SchemaMismatch)); + } + if !seen.insert(field.id()) { + return Err(err(RuntimeMutationErrorCode::InvalidInput)); + } + let definition = c + .fields + .iter() + .find(|f| &f.id == field.id()) + .ok_or_else(|| err(RuntimeMutationErrorCode::InvalidInput))?; + if definition.generated + || (action != RuntimeMutationAction::Create && c.primary_key.contains(field.id())) + { + return Err(err(RuntimeMutationErrorCode::FieldDenied)); + } + check_value(field, value, limits)?; + } + let input_size = fields.iter().try_fold(0usize, |size, (_, value)| { + let bytes = serde_json::to_vec(value) + .map_err(|_| err(RuntimeMutationErrorCode::InvalidInput))?; + size.checked_add(bytes.len()) + .filter(|size| *size <= limits.max_record_bytes) + .ok_or_else(|| err(RuntimeMutationErrorCode::LimitExceeded)) + })?; + let _ = input_size; + if let Some(key) = &key { + for (f, v) in &key.fields { + check_value(f, v, limits)?; + } + } + for field in &c.fields { + if field.generated && field.value_kind != RuntimeValueKind::Uuid { + return Err(err(RuntimeMutationErrorCode::UnsupportedGeneration)); + } + if action == RuntimeMutationAction::Create + && !seen.contains(&field.id) + && !field.generated + && field.default.is_none() + && !field.nullable + { + return Err(err(RuntimeMutationErrorCode::MissingRequired)); + } + } + Ok(RuntimeMutationRequest { + schema: Arc::new(self.clone()), + collection: handle.clone(), + action, + fields: fields.to_vec(), + key, + expected, + returning, + limits, + }) + } +} + +fn bind_value( + backend: DatabaseBackend, + field: &RuntimeFieldHandle, + value: &RuntimeValue, + values: &mut Vec, +) -> String { + if let Some(value) = value_bind(value) { + values.push(value); + placeholder(backend, values.len(), field.value_kind()) + } else { + "NULL".into() + } +} +fn key_sql( + backend: DatabaseBackend, + fields: &[(RuntimeFieldHandle, RuntimeValue)], + values: &mut Vec, +) -> String { + fields + .iter() + .map(|(f, v)| { + format!( + "{} = {}", + backend.quote_identifier(f.physical_column()), + bind_value(backend, f, v, values) + ) + }) + .collect::>() + .join(" AND ") +} +fn columns(backend: DatabaseBackend, projection: &RuntimeProjection) -> String { + projection + .fields() + .iter() + .map(|f| backend.quote_identifier(f.physical_column())) + .collect::>() + .join(", ") +} +fn bounded_record( + record: &RuntimeRecord, + projection: &RuntimeProjection, + limits: RuntimeMutationLimits, +) -> Result<(), RuntimeMutationError> { + for field in projection.fields() { + match record.state(field)? { + RuntimeFieldState::Null => check_value(field, &RuntimeValue::Null, limits)?, + RuntimeFieldState::Value(value) => check_value(field, value, limits)?, + RuntimeFieldState::Unloaded => { + return Err(err(RuntimeMutationErrorCode::DatabaseFailed)); + } + } + } + if record.to_json()?.len() > limits.max_record_bytes { + return Err(err(RuntimeMutationErrorCode::LimitExceeded)); + } + Ok(()) +} +impl MutationContext<'_, B> { + /// Execute pending schema-bound work. Any error or cancellation poisons this transaction. + /// + /// The host must pin the physical environment and complete policy revision under its + /// schema/DDL fence. A successful effect remains uncommitted until the outer runner returns. + pub async fn mutate_runtime( + &mut self, + environment: &RuntimeMutationEnvironment, + request: &RuntimeMutationRequest, + authority: &dyn RuntimeWriteAuthority, + ) -> Result { + let guard = self.runtime_state.enter(); // first poll, before all awaits and validation + let result = self + .mutate_runtime_inner(environment, request, authority) + .await?; + guard.complete(); + Ok(result) + } + /// Execute an atomic journal hook with cancellation poisoning, even after successful DML. + pub async fn run_runtime_before_commit( + &mut self, + hook: &dyn RuntimeMutationHook, + pending: &RuntimeMutationEffect, + ) -> Result<(), RuntimeMutationError> { + let guard = self.runtime_state.enter(); + if !B::RUNTIME_MUTATIONS_SUPPORTED { + return Err(err(RuntimeMutationErrorCode::UnsupportedBackend)); + } + if self.transaction_mode() != TransactionMode::StateMachine { + return Err(err(RuntimeMutationErrorCode::InvalidInput)); + } + hook.before_commit(self, pending).await?; + guard.complete(); + Ok(()) + } + async fn runtime_select( + &mut self, + request: &RuntimeMutationRequest, + projection: &RuntimeProjection, + key: &[(RuntimeFieldHandle, RuntimeValue)], + predicate: Option<&RuntimePredicate>, + lock: bool, + ) -> Result, RuntimeMutationError> { + let mut values = Vec::new(); + let mut condition = key_sql(B::DIALECT, key, &mut values); + if let Some(predicate) = predicate { + condition = format!( + "({condition}) AND ({})", + predicate.render(B::DIALECT, &mut values) + ); + } + if values.len() > request.limits.query.max_bind_parameters { + return Err(err(RuntimeMutationErrorCode::LimitExceeded)); + } + let sql = format!( + "SELECT {} FROM {} WHERE {} LIMIT 2{}", + columns(B::DIALECT, projection), + B::DIALECT.quote_identifier(request.collection.physical_table()), + condition, + if lock && B::DIALECT == DatabaseBackend::Postgres { + " FOR UPDATE" + } else { + "" + } + ); + let rows = self + .fetch_rows(&sql, &values) + .await + .map_err(database_error::)?; + if rows.len() > 1 { + return Err(err(RuntimeMutationErrorCode::DatabaseFailed)); + } + rows.first() + .map(|row| { + let record = projection.decode_row::(row)?; + bounded_record(&record, projection, request.limits)?; + Ok(record) + }) + .transpose() + } + async fn mutate_runtime_inner( + &mut self, + environment: &RuntimeMutationEnvironment, + request: &RuntimeMutationRequest, + authority: &dyn RuntimeWriteAuthority, + ) -> Result { + use RuntimeMutationErrorCode as E; + if !B::RUNTIME_MUTATIONS_SUPPORTED + || B::READ_ONLY + || !B::RUNTIME_ROW_DECODING_SUPPORTED + || !matches!( + B::DIALECT, + DatabaseBackend::Sqlite | DatabaseBackend::Postgres + ) + { + return Err(err(E::UnsupportedBackend)); + } + if self.transaction_mode() != TransactionMode::StateMachine { + return Err(err(E::InvalidInput)); + } + if environment.schema().fingerprint() != request.schema.fingerprint() + || environment.backend() != B::DIALECT + { + return Err(err(E::SchemaMismatch)); + } + let c = collection(&request.schema, &request.collection)?; + if request.action == RuntimeMutationAction::Delete + && environment + .incoming_dependencies(&c.physical_table) + .iter() + .any(|d| !matches!(d.on_delete, DeletePolicy::Restrict)) + { + return Err(err(E::UnsupportedCascade)); + } + let intent = RuntimeWriteIntent { request }; + let grant = authority.authorize_intent(intent, self).await?; + // Validate every host-supplied handle before selecting any protected values. + request + .schema + .resolve_projection(&request.collection, grant.projection.fields())?; + if let Some(p) = &grant.predicate { + p.check_mutation_limits(request.limits)?; + } + let mut binds = request + .fields + .iter() + .filter(|(_, value)| !matches!(value, RuntimeValue::Null)) + .count(); + if request.action == RuntimeMutationAction::Create { + binds = binds.saturating_add(c.fields.iter().filter(|field| field.generated).count()); + } else { + binds = binds.saturating_add(c.primary_key.len()); + for predicate in [grant.predicate.as_ref(), request.expected.as_ref()] + .into_iter() + .flatten() + { + binds = binds.saturating_add(predicate.mutation_bind_count(B::DIALECT)); + } + } + binds = binds.max( + c.primary_key.len().saturating_add( + grant + .predicate + .as_ref() + .map_or(0, |predicate| predicate.mutation_bind_count(B::DIALECT)), + ), + ); + if binds > request.limits.query.max_bind_parameters { + return Err(err(E::LimitExceeded)); + } + let mut fields = grant.projection.fields().to_vec(); + let key_fields = c + .primary_key + .iter() + .map(|id| request.schema.resolve_field(&request.collection, id)) + .collect::, _>>()?; + fields.extend(key_fields.clone()); + fields.extend(request.fields.iter().map(|(f, _)| f.clone())); + if let Some(expected) = &request.expected { + fields.extend(expected.referenced_fields().into_iter().cloned()); + } + if let Some(returning) = &request.returning { + fields.extend(returning.fields().iter().cloned()); + } + let mut seen = BTreeSet::new(); + fields.retain(|f| seen.insert(f.id().clone())); + if fields.len() > request.limits.query.max_projection_fields { + return Err(err(E::LimitExceeded)); + } + let projection = request + .schema + .resolve_projection(&request.collection, &fields)?; + let keys_projection = request + .schema + .resolve_projection(&request.collection, &key_fields)?; + let preimage = if let Some(key) = &request.key { + let record = self + .runtime_select( + request, + &projection, + &key.fields, + grant.predicate.as_ref(), + true, + ) + .await? + .ok_or_else(|| err(E::NotFound))?; + authority + .authorize_preimage( + RuntimePreimageCheck { + intent, + record: &record, + }, + self, + ) + .await + .map_err(|error| { + if matches!(error.code(), E::Denied | E::FieldDenied) { + err(E::NotFound).with_source(error) + } else { + error + } + })?; + if let Some(expected) = &request.expected { + if self + .runtime_select( + request, + &keys_projection, + &key.fields, + Some(expected), + false, + ) + .await? + .is_none() + { + return Err(err(E::Conflict)); + } + } + Some(record) + } else { + None + }; + let mut values = Vec::new(); + let table = B::DIALECT.quote_identifier(&c.physical_table); + let sql = match request.action { + RuntimeMutationAction::Create => { + let mut entries = request.fields.clone(); + for field in c.fields.iter().filter(|f| f.generated) { + entries.push(( + request + .schema + .resolve_field(&request.collection, &field.id)?, + RuntimeValue::Uuid(uuid::Uuid::new_v4()), + )); + } + if entries.is_empty() { + format!("INSERT INTO {table} DEFAULT VALUES") + } else { + let names = entries + .iter() + .map(|(f, _)| B::DIALECT.quote_identifier(f.physical_column())) + .collect::>() + .join(", "); + let binds = entries + .iter() + .map(|(f, v)| bind_value(B::DIALECT, f, v, &mut values)) + .collect::>() + .join(", "); + format!("INSERT INTO {table} ({names}) VALUES ({binds})") + } + } + RuntimeMutationAction::Update | RuntimeMutationAction::Delete => { + let assignments = request + .fields + .iter() + .map(|(f, v)| { + format!( + "{} = {}", + B::DIALECT.quote_identifier(f.physical_column()), + bind_value(B::DIALECT, f, v, &mut values) + ) + }) + .collect::>() + .join(", "); + let key = request.key.as_ref().ok_or_else(|| err(E::InvalidInput))?; + let mut condition = key_sql(B::DIALECT, &key.fields, &mut values); + for p in [grant.predicate.as_ref(), request.expected.as_ref()] + .into_iter() + .flatten() + { + condition = + format!("({condition}) AND ({})", p.render(B::DIALECT, &mut values)); + } + if request.action == RuntimeMutationAction::Update { + format!("UPDATE {table} SET {assignments} WHERE {condition}") + } else { + format!("DELETE FROM {table} WHERE {condition}") + } + } + }; + if values.len() > request.limits.query.max_bind_parameters { + return Err(err(E::LimitExceeded)); + } + let sql = format!("{sql} RETURNING {}", columns(B::DIALECT, &keys_projection)); + let rows = self + .fetch_rows(&sql, &values) + .await + .map_err(database_error::)?; + if rows.is_empty() { + return Err(err(E::Conflict)); + } + if rows.len() != 1 { + return Err(err(E::DatabaseFailed)); + } + let actual_keys = keys_projection.decode_row::(&rows[0])?; + bounded_record(&actual_keys, &keys_projection, request.limits)?; + let key = key_fields + .iter() + .map(|f| actual_keys.value(f).map(|v| (f.clone(), v.clone()))) + .collect::, _>>()?; + let result = if request.action == RuntimeMutationAction::Delete { + preimage.as_ref().ok_or_else(|| err(E::NotFound))?.clone() + } else { + self.runtime_select(request, &projection, &key, grant.predicate.as_ref(), false) + .await? + .ok_or_else(|| err(E::Denied))? + }; + let return_grant = authority + .authorize_result( + RuntimeResultCheck { + intent, + preimage: preimage.as_ref(), + record: &result, + }, + self, + ) + .await?; + if return_grant.projection != request.returning { + return Err(err(E::FieldDenied)); + } + let record = request + .returning + .as_ref() + .map(|p| result.project(p)) + .transpose()?; + Ok(RuntimeMutationEffect { + action: request.action, + record, + }) + } +} diff --git a/crates/graphql-orm/src/graphql/orm/runtime_query.rs b/crates/graphql-orm/src/graphql/orm/runtime_query.rs index 2da63c27..38f26f5a 100644 --- a/crates/graphql-orm/src/graphql/orm/runtime_query.rs +++ b/crates/graphql-orm/src/graphql/orm/runtime_query.rs @@ -230,6 +230,103 @@ impl fmt::Debug for RuntimePredicate { } impl RuntimePredicate { + /// Fields whose values affect this predicate, for independent host authorization. + pub fn referenced_fields(&self) -> Vec<&RuntimeFieldHandle> { + fn visit<'a>(expr: &'a PredicateExpr, fields: &mut Vec<&'a RuntimeFieldHandle>) { + match expr { + PredicateExpr::Constant(_) => {} + PredicateExpr::IsNull { field, .. } + | PredicateExpr::Compare { field, .. } + | PredicateExpr::List { field, .. } + | PredicateExpr::Between { field, .. } => { + if !fields.contains(&field) { + fields.push(field); + } + } + PredicateExpr::And(nodes) | PredicateExpr::Or(nodes) => { + for node in nodes { + visit(node, fields); + } + } + PredicateExpr::Not(node) => visit(node, fields), + } + } + let mut fields = Vec::new(); + visit(&self.expr, &mut fields); + fields + } + #[cfg(any(feature = "sqlite", feature = "postgres", feature = "mssql"))] + pub(crate) fn mutation_bind_count(&self, backend: DatabaseBackend) -> usize { + // Count emitted values, not logical operands or placeholder occurrences. + // Dialects may duplicate a value (SQLite suffixes) or reference one native + // slot repeatedly (PostgreSQL). Keep preflight coupled to execution rendering. + let mut values = Vec::new(); + self.render(backend, &mut values); + values.len() + } + + #[cfg(any(feature = "sqlite", feature = "postgres", feature = "mssql"))] + pub(crate) fn check_mutation_limits( + &self, + limits: super::RuntimeMutationLimits, + ) -> Result<(), super::RuntimeMutationError> { + if self.nodes > limits.query.max_predicate_nodes + || self.depth > limits.query.max_predicate_depth + || self.binds > limits.query.max_bind_parameters + { + return Err(super::RuntimeMutationError::new( + super::RuntimeMutationErrorCode::LimitExceeded, + )); + } + fn visit( + expr: &PredicateExpr, + limits: super::RuntimeMutationLimits, + bytes: &mut usize, + ) -> Result<(), super::RuntimeMutationError> { + use super::{RuntimeMutationError, RuntimeMutationErrorCode}; + let mut check = |field: &RuntimeFieldHandle, value: &RuntimeValue| { + super::runtime_mutation::check_value(field, value, limits)?; + let size = serde_json::to_vec(value) + .map_err(|_| RuntimeMutationError::new(RuntimeMutationErrorCode::InvalidInput))? + .len(); + *bytes = bytes + .checked_add(size) + .filter(|n| *n <= limits.max_record_bytes) + .ok_or_else(|| { + RuntimeMutationError::new(RuntimeMutationErrorCode::LimitExceeded) + })?; + Ok::<_, RuntimeMutationError>(()) + }; + match expr { + PredicateExpr::Constant(_) | PredicateExpr::IsNull { .. } => {} + PredicateExpr::Compare { field, value, .. } => check(field, value)?, + PredicateExpr::List { field, values, .. } => { + if values.len() > limits.query.max_values_per_list { + return Err(RuntimeMutationError::new( + RuntimeMutationErrorCode::LimitExceeded, + )); + } + for value in values { + check(field, value)?; + } + } + PredicateExpr::Between { field, low, high } => { + check(field, low)?; + check(field, high)?; + } + PredicateExpr::And(nodes) | PredicateExpr::Or(nodes) => { + for node in nodes { + visit(node, limits, bytes)?; + } + } + PredicateExpr::Not(node) => visit(node, limits, bytes)?, + } + Ok(()) + } + visit(&self.expr, limits, &mut 0)?; + Ok(()) + } + pub(crate) fn belongs_to(&self, schema: &SchemaFingerprint, collection: &CollectionId) -> bool { &self.schema == schema && &self.collection == collection } @@ -1600,6 +1697,76 @@ mod tests { .unwrap() } + #[cfg(any(feature = "sqlite", feature = "postgres", feature = "mssql"))] + #[test] + fn mutation_counts_emitted_backend_slots_including_nested_native_reuse() { + let schema = schema(); + let collection = schema.resolve_collection(&collection_id("items")).unwrap(); + let name = schema + .resolve_field(&collection, &field_id("name")) + .unwrap(); + for literal in ["", "π%_\\"] { + let compare = |op| { + schema + .runtime_compare( + &collection, + &name, + op, + RuntimeValue::String(literal.into()), + RuntimeQueryLimits::default(), + ) + .unwrap() + }; + let suffix = compare(RuntimeScalarOperator::EndsWith); + let prefix = compare(RuntimeScalarOperator::StartsWith); + let nested = schema + .runtime_and( + &collection, + vec![ + suffix.clone(), + schema + .runtime_or( + &collection, + vec![ + schema + .runtime_not( + &collection, + prefix.clone(), + Default::default(), + ) + .unwrap(), + suffix.clone(), + ], + Default::default(), + ) + .unwrap(), + ], + Default::default(), + ) + .unwrap(); + for backend in [DatabaseBackend::Sqlite, DatabaseBackend::Postgres] { + let suffix_slots = if backend == DatabaseBackend::Sqlite { + 2 + } else { + 1 + }; + assert_eq!(suffix.mutation_bind_count(backend), suffix_slots); + assert_eq!(prefix.mutation_bind_count(backend), 1); + assert_eq!(nested.mutation_bind_count(backend), 2 * suffix_slots + 1); + // Existing surrounding query values change numbering, not slot cost. + let mut values = vec![SqlValue::Int(0)]; + let sql = nested.render(backend, &mut values); + assert_eq!(nested.mutation_bind_count(backend), values.len() - 1); + if backend == DatabaseBackend::Postgres { + for slot in ["$2", "$3", "$4"] { + assert_eq!(sql.matches(slot).count(), 2); + } + assert_eq!(values.len(), 4); + } + } + } + } + #[test] fn renderer_quotes_handles_and_binds_hostile_values() { let schema = schema(); diff --git a/crates/graphql-orm/src/graphql/orm/runtime_transaction.rs b/crates/graphql-orm/src/graphql/orm/runtime_transaction.rs new file mode 100644 index 00000000..98375bd7 --- /dev/null +++ b/crates/graphql-orm/src/graphql/orm/runtime_transaction.rs @@ -0,0 +1,60 @@ +//! Monotonic transaction state for cancellation-safe runtime work. +use std::sync::{ + Arc, + atomic::{AtomicBool, AtomicUsize, Ordering}, +}; + +#[derive(Default)] +pub(crate) struct RuntimeTransactionState { + unfinished: AtomicUsize, + poisoned: AtomicBool, +} +impl RuntimeTransactionState { + pub(crate) fn enter(self: &Arc) -> RuntimeOperationGuard { + self.unfinished.fetch_add(1, Ordering::SeqCst); + RuntimeOperationGuard { + state: self.clone(), + complete: false, + } + } + pub(crate) fn cannot_commit(&self) -> bool { + self.poisoned.load(Ordering::SeqCst) || self.unfinished.load(Ordering::SeqCst) != 0 + } +} +pub(crate) struct RuntimeOperationGuard { + state: Arc, + complete: bool, +} +impl RuntimeOperationGuard { + pub(crate) fn complete(mut self) { + self.complete = true; + } +} +impl Drop for RuntimeOperationGuard { + fn drop(&mut self) { + if !self.complete { + self.state.poisoned.store(true, Ordering::SeqCst); + } + self.state.unfinished.fetch_sub(1, Ordering::SeqCst); + } +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn dropped_and_forgotten_operations_never_allow_commit() { + let state = Arc::new(RuntimeTransactionState::default()); + assert!(!state.cannot_commit()); + let guard = state.enter(); + assert!(state.cannot_commit()); + guard.complete(); + assert!(!state.cannot_commit()); + drop(state.enter()); + state.enter().complete(); + assert!(state.cannot_commit()); + let state = Arc::new(RuntimeTransactionState::default()); + std::mem::forget(state.enter()); + assert!(state.cannot_commit()); + } +} diff --git a/crates/graphql-orm/tests/fixtures/backend-coexistence/Cargo.lock b/crates/graphql-orm/tests/fixtures/backend-coexistence/Cargo.lock index a051d5ca..5e05234c 100644 --- a/crates/graphql-orm/tests/fixtures/backend-coexistence/Cargo.lock +++ b/crates/graphql-orm/tests/fixtures/backend-coexistence/Cargo.lock @@ -1350,7 +1350,7 @@ dependencies = [ [[package]] name = "graphql-orm" -version = "0.40.0" +version = "0.41.0" dependencies = [ "agql-auth", "async-graphql", @@ -1420,7 +1420,7 @@ dependencies = [ [[package]] name = "graphql-orm-macros" -version = "0.40.0" +version = "0.41.0" dependencies = [ "convert_case", "proc-macro2", diff --git a/crates/graphql-orm/tests/fixtures/federation-end-to-end/Cargo.lock b/crates/graphql-orm/tests/fixtures/federation-end-to-end/Cargo.lock index 604781a8..3a2d0b81 100644 --- a/crates/graphql-orm/tests/fixtures/federation-end-to-end/Cargo.lock +++ b/crates/graphql-orm/tests/fixtures/federation-end-to-end/Cargo.lock @@ -2508,7 +2508,7 @@ dependencies = [ [[package]] name = "graphql-orm" -version = "0.40.0" +version = "0.41.0" dependencies = [ "async-graphql", "chacha20poly1305", @@ -2531,7 +2531,7 @@ dependencies = [ [[package]] name = "graphql-orm-macros" -version = "0.40.0" +version = "0.41.0" dependencies = [ "convert_case 0.7.1", "proc-macro2", diff --git a/crates/graphql-orm/tests/fixtures/migration-api-compatibility/Cargo.lock b/crates/graphql-orm/tests/fixtures/migration-api-compatibility/Cargo.lock index a549e290..cf4a5e33 100644 --- a/crates/graphql-orm/tests/fixtures/migration-api-compatibility/Cargo.lock +++ b/crates/graphql-orm/tests/fixtures/migration-api-compatibility/Cargo.lock @@ -1004,7 +1004,7 @@ dependencies = [ [[package]] name = "graphql-orm" -version = "0.40.0" +version = "0.41.0" dependencies = [ "async-graphql", "chacha20poly1305", @@ -1030,7 +1030,7 @@ dependencies = [ [[package]] name = "graphql-orm-macros" -version = "0.40.0" +version = "0.41.0" dependencies = [ "convert_case", "proc-macro2", diff --git a/crates/graphql-orm/tests/fixtures/repository-aggregate-consumer/Cargo.lock b/crates/graphql-orm/tests/fixtures/repository-aggregate-consumer/Cargo.lock index d288e123..bb534097 100644 --- a/crates/graphql-orm/tests/fixtures/repository-aggregate-consumer/Cargo.lock +++ b/crates/graphql-orm/tests/fixtures/repository-aggregate-consumer/Cargo.lock @@ -1004,7 +1004,7 @@ dependencies = [ [[package]] name = "graphql-orm" -version = "0.40.0" +version = "0.41.0" dependencies = [ "async-graphql", "chacha20poly1305", @@ -1030,7 +1030,7 @@ dependencies = [ [[package]] name = "graphql-orm-macros" -version = "0.40.0" +version = "0.41.0" dependencies = [ "convert_case", "proc-macro2", diff --git a/crates/graphql-orm/tests/fixtures/repository-aggregate-consumer/examples/runtime_mutation_journal.rs b/crates/graphql-orm/tests/fixtures/repository-aggregate-consumer/examples/runtime_mutation_journal.rs new file mode 100644 index 00000000..ab892fab --- /dev/null +++ b/crates/graphql-orm/tests/fixtures/repository-aggregate-consumer/examples/runtime_mutation_journal.rs @@ -0,0 +1,4 @@ +//! Compile and execute the public host example without a direct async-graphql dependency. +#[path = "../../../../examples/runtime_mutation_journal.rs"] +mod host_example; +fn main() -> Result<(), Box> { host_example::main() } diff --git a/crates/graphql-orm/tests/runtime_mutations.rs b/crates/graphql-orm/tests/runtime_mutations.rs new file mode 100644 index 00000000..fe082a45 --- /dev/null +++ b/crates/graphql-orm/tests/runtime_mutations.rs @@ -0,0 +1,1369 @@ +//! Runtime mutation conformance: each PostgreSQL case owns a verified disposable server. +#![cfg(any( + all(feature = "sqlite", not(any(feature = "postgres", feature = "mssql"))), + all(feature = "postgres", not(any(feature = "sqlite", feature = "mssql"))) +))] +use futures::future::BoxFuture; +use graphql_orm::{graphql::errors::OrmPublicError, graphql::orm::*, prelude::*}; +use std::{ + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, + time::Duration, +}; +#[cfg(feature = "sqlite")] +type Backend = SqliteBackend; +#[cfg(feature = "postgres")] +type Backend = PostgresBackend; +#[cfg(feature = "postgres")] +#[path = "support/owned_postgres.rs"] +mod owned_postgres; + +#[derive(RepositoryEntity, Clone, serde::Serialize, serde::Deserialize)] +#[repository_entity(table = "runtime_journal", plural = "RuntimeJournal")] +struct Journal { + #[primary_key] + #[graphql_orm(auto_generated = false)] + id: String, + message: String, +} +fn definition() -> RuntimeSchema { + let mut fields = Vec::new(); + for (name, kind, nullable, default) in [ + ("id", RuntimeValueKind::Integer, false, None), + ("tenant", RuntimeValueKind::String, false, None), + ("label", RuntimeValueKind::String, false, None), + ( + "score", + RuntimeValueKind::Integer, + false, + Some(RuntimeDefault::Literal("7".into())), + ), + ("optional", RuntimeValueKind::String, true, None), + ( + "occurred", + RuntimeValueKind::DateTime, + false, + Some(RuntimeDefault::CurrentTimestamp), + ), + ] { + fields.push(RuntimeField { + id: FieldId::new(name).unwrap(), + api_name: name.into(), + physical_column: name.into(), + value_kind: kind, + nullable, + unique: false, + filterable: true, + sortable: true, + generated: false, + default, + }); + } + RuntimeSchema { + format_version: RUNTIME_SCHEMA_FORMAT_VERSION, + collections: vec![RuntimeCollection { + id: CollectionId::new("records").unwrap(), + api_type_name: "Record".into(), + api_plural_name: "Records".into(), + physical_table: "runtime_records".into(), + primary_key: vec![FieldId::new("id").unwrap()], + append_only: false, + retention_purge: false, + fields, + relations: vec![], + indexes: vec![], + composite_unique: vec![], + default_order: vec![], + }], + } +} +struct Fixture { + db: Database, + environment: Arc, + #[cfg(feature = "postgres")] + owned: owned_postgres::OwnedPostgres, +} +impl Fixture { + async fn new() -> Self { + Self::with_definition(definition()).await + } + async fn with_definition(definition: RuntimeSchema) -> Self { + #[cfg(feature = "sqlite")] + let db = Database::::connect_sqlite_with_options( + "sqlite::memory:", + graphql_orm::db::ConnectionOptions::default().max_connections(1), + ) + .await + .unwrap(); + #[cfg(feature = "postgres")] + let owned = owned_postgres::OwnedPostgres::start("runtime-mutations").unwrap(); + #[cfg(feature = "postgres")] + let db = Database::::connect_postgres(&owned.url) + .await + .unwrap(); + let schema = Arc::new(definition.validate().unwrap()); + let target = schema + .physical_schema::(Default::default()) + .unwrap() + .with_static_entities(&[Journal::metadata()]) + .unwrap(); + let ownership = + ManagedTableSet::new(["runtime_records".into(), "runtime_journal".into()]).unwrap(); + let plan = db + .schema() + .plan_owned_migration( + "init", + "runtime test", + &target, + &ownership, + PlanOptions::strict(), + ) + .await + .unwrap(); + db.schema() + .apply_owned_migration(&plan, ApplyOptions::default()) + .await + .unwrap(); + let environment = Arc::new( + db.schema() + .runtime_mutation_environment(schema, &target, &ownership) + .await + .unwrap(), + ); + Self { + db, + environment, + #[cfg(feature = "postgres")] + owned, + } + } + fn schema(&self) -> &ValidatedRuntimeSchema { + self.environment.schema() + } + fn collection(&self) -> RuntimeCollectionHandle { + self.schema() + .resolve_collection(&CollectionId::new("records").unwrap()) + .unwrap() + } + fn field(&self, name: &str) -> RuntimeFieldHandle { + self.schema() + .resolve_field(&self.collection(), &FieldId::new(name).unwrap()) + .unwrap() + } + fn projection(&self) -> RuntimeProjection { + self.schema() + .resolve_projection( + &self.collection(), + &[ + self.field("id"), + self.field("label"), + self.field("score"), + self.field("optional"), + self.field("occurred"), + ], + ) + .unwrap() + } + fn authority(&self) -> Authority { + Authority { + projection: self.projection(), + predicate: Some( + self.schema() + .runtime_compare( + &self.collection(), + &self.field("tenant"), + RuntimeScalarOperator::Eq, + RuntimeValue::String("north".into()), + Default::default(), + ) + .unwrap(), + ), + expected_policy_revision: 1, + pinned_policy_revision: 1, + deny_intent: false, + deny_preimage: false, + preimage_calls: Arc::new(std::sync::atomic::AtomicUsize::new(0)), + deny_result: false, + pending_result: false, + staged: Arc::new(AtomicBool::new(false)), + wrong_return: false, + } + } + fn create(&self, id: i64) -> RuntimeMutationRequest { + self.schema() + .runtime_create_request( + &self.collection(), + &[ + (self.field("id"), RuntimeValue::Integer(id)), + (self.field("tenant"), RuntimeValue::String("north".into())), + (self.field("label"), RuntimeValue::String("initial".into())), + ], + Some(self.projection()), + Default::default(), + ) + .unwrap() + } + fn key(&self, id: i64) -> RuntimeKey { + self.schema() + .runtime_key( + &self.collection(), + &[(self.field("id"), RuntimeValue::Integer(id))], + ) + .unwrap() + } + async fn run( + &self, + request: RuntimeMutationRequest, + authority: Authority, + ) -> Result { + let environment = self.environment.clone(); + self.db + .transaction(TransactionMode::StateMachine, move |tx| { + Box::pin(async move { + tx.mutate_runtime(&environment, &request, &authority) + .await + .map_err(OrmPublicError::from) + }) + }) + .await + } + async fn count(&self) -> i64 { + let rows = Backend::fetch_rows( + self.db.pool(), + "SELECT COUNT(*) AS n FROM runtime_records", + &[], + ) + .await + .unwrap(); + Backend::try_get_i64(&rows[0], "n").unwrap() + } + async fn finish(mut self) { + self.db.pool().close().await; + #[cfg(feature = "postgres")] + self.owned.cleanup().unwrap(); + #[cfg(feature = "sqlite")] + { + let _ = &mut self; + } + } +} +struct Authority { + projection: RuntimeProjection, + predicate: Option, + expected_policy_revision: u64, + pinned_policy_revision: u64, + deny_intent: bool, + deny_preimage: bool, + preimage_calls: Arc, + deny_result: bool, + pending_result: bool, + staged: Arc, + wrong_return: bool, +} +impl RuntimeWriteAuthority for Authority { + fn authorize_intent<'a>( + &'a self, + _check: RuntimeWriteIntent<'a>, + _tx: &'a mut MutationContext<'_, Backend>, + ) -> BoxFuture<'a, Result> { + Box::pin(async move { + if self.expected_policy_revision != self.pinned_policy_revision { + return Err(RuntimeMutationError::new( + RuntimeMutationErrorCode::SchemaMismatch, + )); + } + if self.deny_intent { + return Err(RuntimeMutationError::new( + RuntimeMutationErrorCode::FieldDenied, + )); + } + RuntimeWriteGrant::new(self.projection.clone(), self.predicate.clone()) + }) + } + fn authorize_preimage<'a>( + &'a self, + _check: RuntimePreimageCheck<'a>, + _tx: &'a mut MutationContext<'_, Backend>, + ) -> BoxFuture<'a, Result<(), RuntimeMutationError>> { + Box::pin(async move { + self.preimage_calls.fetch_add(1, Ordering::SeqCst); + if self.deny_preimage { + return Err(RuntimeMutationError::new(RuntimeMutationErrorCode::Denied)); + } + Ok(()) + }) + } + fn authorize_result<'a>( + &'a self, + check: RuntimeResultCheck<'a>, + tx: &'a mut MutationContext<'_, Backend>, + ) -> BoxFuture<'a, Result> { + Box::pin(async move { + if self.pending_result { + tx.insert::(CreateJournalInput { + id: "pending".into(), + message: "staged".into(), + }) + .await + .map_err(|e| { + RuntimeMutationError::new(RuntimeMutationErrorCode::HookFailed).with_source(e) + })?; + tx.queue_event(19u32); + self.staged.store(true, Ordering::SeqCst); + std::future::pending::<()>().await; + } + if self.deny_result { + return Err(RuntimeMutationError::new(RuntimeMutationErrorCode::Denied)); + } + Ok(RuntimeReturnGrant::new(if self.wrong_return { + None + } else { + check.intent.returning().cloned() + })) + }) + } +} +struct JournalHook { + staged: Arc, + pending: bool, + fail: bool, +} +impl RuntimeMutationHook for JournalHook { + fn before_commit<'a>( + &'a self, + tx: &'a mut MutationContext<'_, Backend>, + _pending: &'a RuntimeMutationEffect, + ) -> BoxFuture<'a, Result<(), RuntimeMutationError>> { + Box::pin(async move { + tx.insert::(CreateJournalInput { + id: "hook".into(), + message: "staged".into(), + }) + .await + .map_err(|e| { + RuntimeMutationError::new(RuntimeMutationErrorCode::HookFailed).with_source(e) + })?; + tx.queue_event(20u32); + self.staged.store(true, Ordering::SeqCst); + if self.pending { + std::future::pending::<()>().await; + } + if self.fail { + return Err(RuntimeMutationError::new( + RuntimeMutationErrorCode::HookFailed, + )); + } + Ok(()) + }) + } +} +#[tokio::test] +async fn create_patch_cas_delete_defaults_and_private_return_projection() { + let f = Fixture::new().await; + let row = f.run(f.create(1), f.authority()).await.unwrap(); + assert_eq!(row.record().unwrap().integer(&f.field("score")).unwrap(), 7); + assert!(matches!( + row.record().unwrap().state(&f.field("optional")).unwrap(), + RuntimeFieldState::Null + )); + assert!(row.record().unwrap().datetime(&f.field("occurred")).is_ok()); + assert!(row.record().unwrap().value(&f.field("tenant")).is_err()); + let expected = f + .schema() + .runtime_compare( + &f.collection(), + &f.field("score"), + RuntimeScalarOperator::Eq, + RuntimeValue::Integer(7), + Default::default(), + ) + .unwrap(); + let patch = f + .schema() + .runtime_update_request( + f.key(1), + &[(f.field("score"), RuntimeValue::Integer(-123))], + Some(expected.clone()), + Some(f.projection()), + Default::default(), + ) + .unwrap(); + let row = f.run(patch.clone(), f.authority()).await.unwrap(); + assert_eq!( + row.record().unwrap().integer(&f.field("score")).unwrap(), + -123 + ); + assert_eq!( + row.record().unwrap().string(&f.field("label")).unwrap(), + "initial" + ); + let conflict = f.run(patch, f.authority()).await.unwrap_err(); + assert_eq!( + conflict.public_error().runtime_mutation_code(), + Some("conflict") + ); + assert_eq!( + conflict.commit_outcome(), + TransactionCommitOutcome::RolledBack + ); + let delete = f + .schema() + .runtime_delete_request(f.key(1), None, Some(f.projection()), Default::default()) + .unwrap(); + f.run(delete, f.authority()).await.unwrap(); + assert_eq!(f.count().await, 0); + f.finish().await; +} +#[tokio::test] +async fn caught_result_timeout_rolls_back_record_journal_events_and_reuses_connection() { + let f = Fixture::new().await; + let environment = f.environment.clone(); + let request = f.create(1); + let mut authority = f.authority(); + authority.pending_result = true; + let staged = authority.staged.clone(); + let mut events = f.db.ensure_event_sender::().subscribe(); + let result = + f.db.transaction(TransactionMode::StateMachine, move |tx| { + Box::pin(async move { + assert!( + tokio::time::timeout( + Duration::from_millis(150), + tx.mutate_runtime(&environment, &request, &authority) + ) + .await + .is_err() + ); + Ok(()) + }) + }) + .await; + assert!(staged.load(Ordering::SeqCst)); + assert!(matches!(result, Err(TransactionError::Rejected(_)))); + assert_eq!(f.count().await, 0); + assert_eq!(Journal::count_all(&f.db).await.unwrap(), 0); + assert!(events.try_recv().is_err()); + f.run(f.create(2), f.authority()).await.unwrap(); + assert_eq!(f.count().await, 1); + f.finish().await; +} +#[tokio::test] +async fn caught_hook_timeout_and_error_poison_successful_mutations() { + let f = Fixture::new().await; + for pending in [true, false] { + let environment = f.environment.clone(); + let request = f.create(1); + let authority = f.authority(); + let mut events = f.db.ensure_event_sender::().subscribe(); + let result = + f.db.transaction(TransactionMode::StateMachine, move |tx| { + Box::pin(async move { + let effect = tx + .mutate_runtime(&environment, &request, &authority) + .await?; + let hook = JournalHook { + staged: Arc::new(AtomicBool::new(false)), + pending, + fail: !pending, + }; + let result = tokio::time::timeout( + Duration::from_millis(150), + tx.run_runtime_before_commit(&hook, &effect), + ) + .await; + assert!(result.is_err() || result.unwrap().is_err()); + assert!(hook.staged.load(Ordering::SeqCst)); + Ok(()) + }) + }) + .await; + assert!(matches!(result, Err(TransactionError::Rejected(_)))); + assert_eq!(f.count().await, 0); + assert_eq!(Journal::count_all(&f.db).await.unwrap(), 0); + assert!(events.try_recv().is_err()); + } + f.finish().await; +} +#[tokio::test] +async fn denied_intent_preimage_result_and_output_are_fail_closed() { + let f = Fixture::new().await; + for stage in 0..3 { + let mut authority = f.authority(); + authority.deny_intent = stage == 0; + authority.deny_result = stage == 1; + authority.wrong_return = stage == 2; + assert!(f.run(f.create(1), authority).await.is_err()); + assert_eq!(f.count().await, 0); + } + f.run(f.create(1), f.authority()).await.unwrap(); + let request = f + .schema() + .runtime_delete_request(f.key(1), None, None, Default::default()) + .unwrap(); + let mut authority = f.authority(); + authority.deny_preimage = true; + assert_eq!( + f.run(request, authority) + .await + .unwrap_err() + .public_error() + .runtime_mutation_code(), + Some("not_found") + ); + assert_eq!(f.count().await, 1); + f.finish().await; +} +#[tokio::test] +async fn journal_success_and_propagated_failure_are_atomic() { + let f = Fixture::new().await; + for id in [1, 2] { + let environment = f.environment.clone(); + let request = f.create(id); + let authority = f.authority(); + let result = + f.db.transaction(TransactionMode::StateMachine, move |tx| { + Box::pin(async move { + let effect = tx + .mutate_runtime(&environment, &request, &authority) + .await?; + tx.run_runtime_before_commit( + &JournalHook { + staged: Arc::new(AtomicBool::new(false)), + pending: false, + fail: false, + }, + &effect, + ) + .await?; + Ok(effect) + }) + }) + .await; + assert_eq!(result.is_ok(), id == 1); + } + assert_eq!(f.count().await, 1); + assert_eq!(Journal::count_all(&f.db).await.unwrap(), 1); + f.finish().await; +} + +#[tokio::test] +async fn nullable_patch_omission_and_explicit_null_remain_distinct() { + let f = Fixture::new().await; + f.run(f.create(1), f.authority()).await.unwrap(); + for value in [RuntimeValue::String("present".into()), RuntimeValue::Null] { + let request = f + .schema() + .runtime_update_request( + f.key(1), + &[(f.field("optional"), value.clone())], + None, + Some(f.projection()), + Default::default(), + ) + .unwrap(); + let row = f.run(request, f.authority()).await.unwrap(); + let actual = row.record().unwrap().state(&f.field("optional")).unwrap(); + match &value { + RuntimeValue::Null => assert!(matches!(actual, RuntimeFieldState::Null)), + _ => assert!(matches!(actual,RuntimeFieldState::Value(v) if v==&value)), + } + let request = f + .schema() + .runtime_update_request( + f.key(1), + &[(f.field("label"), RuntimeValue::String("changed".into()))], + None, + Some(f.projection()), + Default::default(), + ) + .unwrap(); + let row = f.run(request, f.authority()).await.unwrap(); + let actual = row.record().unwrap().state(&f.field("optional")).unwrap(); + match &value { + RuntimeValue::Null => assert!(matches!(actual, RuntimeFieldState::Null)), + _ => assert!(matches!(actual,RuntimeFieldState::Value(v) if v==&value)), + } + } + f.finish().await; +} +#[tokio::test] +async fn competing_structural_cas_writers_cannot_both_commit() { + let f = Fixture::new().await; + f.run(f.create(1), f.authority()).await.unwrap(); + let expected = f + .schema() + .runtime_compare( + &f.collection(), + &f.field("score"), + RuntimeScalarOperator::Eq, + RuntimeValue::Integer(7), + Default::default(), + ) + .unwrap(); + let request = f + .schema() + .runtime_update_request( + f.key(1), + &[(f.field("score"), RuntimeValue::Integer(8))], + Some(expected), + None, + Default::default(), + ) + .unwrap(); + let (first, second) = tokio::join!( + f.run(request.clone(), f.authority()), + f.run(request.clone(), f.authority()) + ); + assert_eq!(usize::from(first.is_ok()) + usize::from(second.is_ok()), 1); + let failure = if let Err(e) = first { + e + } else { + second.unwrap_err() + }; + assert!( + failure.is_retryable() + || failure.public_error().runtime_mutation_code() == Some("conflict") + ); + let after = f.run(request, f.authority()).await.unwrap_err(); + assert_eq!( + after.public_error().runtime_mutation_code(), + Some("conflict") + ); + f.finish().await; +} +#[tokio::test] +async fn caught_denial_is_monotonic_and_unpolled_future_stages_nothing() { + let f = Fixture::new().await; + let environment = f.environment.clone(); + let request = f.create(1); + let authority = f.authority(); + f.db.transaction(TransactionMode::StateMachine, move |tx| { + Box::pin(async move { + drop(tx.mutate_runtime(&environment, &request, &authority)); + Ok(()) + }) + }) + .await + .unwrap(); + assert_eq!(f.count().await, 0); + let environment = f.environment.clone(); + let request = f.create(1); + let mut authority = f.authority(); + authority.deny_result = true; + let good = f.authority(); + let later = f.create(2); + let result = + f.db.transaction(TransactionMode::StateMachine, move |tx| { + Box::pin(async move { + assert!( + tx.mutate_runtime(&environment, &request, &authority) + .await + .is_err() + ); + // A later successful operation cannot clear the earlier poison. + tx.mutate_runtime(&environment, &later, &good) + .await + .unwrap(); + Ok(()) + }) + }) + .await; + assert!(result.is_err()); + assert_eq!(f.count().await, 0); + f.finish().await; +} +#[tokio::test] +async fn row_visibility_hides_existence_before_cas_conflict_disclosure() { + let f = Fixture::new().await; + f.run(f.create(1), f.authority()).await.unwrap(); + let request = f + .schema() + .runtime_delete_request(f.key(1), None, None, Default::default()) + .unwrap(); + let mut authority = f.authority(); + authority.predicate = Some( + f.schema() + .runtime_compare( + &f.collection(), + &f.field("tenant"), + RuntimeScalarOperator::Eq, + RuntimeValue::String("south".into()), + Default::default(), + ) + .unwrap(), + ); + assert_eq!( + f.run(request, authority) + .await + .unwrap_err() + .public_error() + .runtime_mutation_code(), + Some("not_found") + ); + assert_eq!(f.count().await, 1); + f.finish().await; +} +#[test] +fn requests_reject_missing_keys_wrong_types_stale_handles_append_only_and_limits() { + let schema = definition().validate().unwrap(); + let c = schema + .resolve_collection(&CollectionId::new("records").unwrap()) + .unwrap(); + let id = schema + .resolve_field(&c, &FieldId::new("id").unwrap()) + .unwrap(); + assert!(schema.runtime_key(&c, &[]).is_err()); + assert!( + schema + .runtime_key(&c, &[(id.clone(), RuntimeValue::Null)]) + .is_err() + ); + assert!( + schema + .runtime_key(&c, &[(id.clone(), RuntimeValue::String("1".into()))]) + .is_err() + ); + assert_eq!( + schema + .runtime_create_request(&c, &[], None, Default::default()) + .unwrap_err() + .code(), + RuntimeMutationErrorCode::MissingRequired + ); + let key = schema + .runtime_key(&c, &[(id.clone(), RuntimeValue::Integer(1))]) + .unwrap(); + assert_eq!( + schema + .runtime_update_request( + key.clone(), + &[(id.clone(), RuntimeValue::Integer(2))], + None, + None, + Default::default() + ) + .unwrap_err() + .code(), + RuntimeMutationErrorCode::FieldDenied + ); + assert!( + schema + .runtime_update_request(key.clone(), &[], None, None, Default::default()) + .is_err() + ); + let mut changed = definition(); + changed.collections[0].append_only = true; + let changed = changed.validate().unwrap(); + assert!( + changed + .runtime_delete_request(key, None, None, Default::default()) + .is_err() + ); + let c = changed.resolve_collection(c.id()).unwrap(); + let id = changed.resolve_field(&c, id.id()).unwrap(); + let key = changed + .runtime_key(&c, &[(id, RuntimeValue::Integer(1))]) + .unwrap(); + assert_eq!( + changed + .runtime_delete_request(key, None, None, Default::default()) + .unwrap_err() + .code(), + RuntimeMutationErrorCode::AppendOnly + ); +} + +#[tokio::test] +async fn whole_transaction_cancellation_after_staged_work_rolls_back() { + let f = Fixture::new().await; + let mut authority = f.authority(); + authority.pending_result = true; + let staged = authority.staged.clone(); + assert!( + tokio::time::timeout(Duration::from_millis(150), f.run(f.create(1), authority)) + .await + .is_err() + ); + assert!(staged.load(Ordering::SeqCst)); + assert_eq!(f.count().await, 0); + assert_eq!(Journal::count_all(&f.db).await.unwrap(), 0); + f.run(f.create(2), f.authority()).await.unwrap(); + f.finish().await; +} +#[tokio::test] +async fn denied_fields_are_checked_before_target_io() { + let f = Fixture::new().await; + let mut authority = f.authority(); + authority.deny_intent = true; + ::execute_with_binds(f.db.pool(), "DROP TABLE runtime_records", &[]) + .await + .unwrap(); + // A target query here would fail. Host intent denial wins before target I/O. + assert_eq!( + f.run(f.create(1), authority) + .await + .unwrap_err() + .public_error() + .runtime_mutation_code(), + Some("field_denied") + ); + f.finish().await; +} +#[cfg(feature = "postgres")] +#[tokio::test] +async fn postgres_auth_context_is_pinned_and_rls_is_not_bypassed() { + use std::str::FromStr; + let f = Fixture::new().await; + for ddl in [ + "ALTER TABLE runtime_records ENABLE ROW LEVEL SECURITY", + "ALTER TABLE runtime_records FORCE ROW LEVEL SECURITY", + "CREATE POLICY tenant_rows ON runtime_records USING (tenant = current_setting('app.tenant_id', true)) WITH CHECK (tenant = current_setting('app.tenant_id', true))", + "CREATE ROLE runtime_writer LOGIN PASSWORD 'owned-fixture-only'", + "GRANT USAGE ON SCHEMA public TO runtime_writer", + "GRANT SELECT, INSERT, UPDATE, DELETE ON runtime_records, runtime_journal TO runtime_writer", + ] { + sqlx::query(ddl).execute(f.db.pool()).await.unwrap(); + } + let options = sqlx::postgres::PgConnectOptions::from_str(&f.owned.url) + .unwrap() + .username("runtime_writer") + .password("owned-fixture-only"); + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect_with(options) + .await + .unwrap(); + let db = Database::::new(pool.clone()); + for tenant in ["north", "south"] { + let environment = f.environment.clone(); + let request = f.create(if tenant == "north" { 1 } else { 2 }); + let authority = f.authority(); + let auth = DbAuthContext { + tenant_id: Some(tenant.into()), + ..Default::default() + }; + let result = db + .transaction_with_auth(TransactionMode::StateMachine, Some(&auth), move |tx| { + Box::pin(async move { + tx.mutate_runtime(&environment, &request, &authority) + .await + .map_err(OrmPublicError::from) + }) + }) + .await; + assert_eq!(result.is_ok(), tenant == "north"); + } + assert_eq!(f.count().await, 1); + let setting: Option = + sqlx::query_scalar("SELECT nullif(current_setting('app.tenant_id', true), '')") + .fetch_one(&pool) + .await + .unwrap(); + assert!(setting.is_none()); + pool.close().await; + f.finish().await; +} + +#[tokio::test] +async fn generated_uuid_json_null_bytes_and_exact_values_use_existing_runtime_types() { + let mut schema = definition(); + schema.collections[0].fields[0].value_kind = RuntimeValueKind::Uuid; + schema.collections[0].fields[0].generated = true; + for (name, kind) in [ + ("json", RuntimeValueKind::Json), + ("bytes", RuntimeValueKind::Bytes), + ] { + schema.collections[0].fields.push(RuntimeField { + id: FieldId::new(name).unwrap(), + api_name: name.into(), + physical_column: name.into(), + value_kind: kind, + nullable: true, + unique: false, + filterable: false, + sortable: false, + generated: false, + default: None, + }); + } + let f = Fixture::with_definition(schema).await; + let projection = f + .schema() + .resolve_projection( + &f.collection(), + &[ + f.field("id"), + f.field("json"), + f.field("bytes"), + f.field("occurred"), + f.field("score"), + ], + ) + .unwrap(); + let datetime = RuntimeDateTime::parse("2026-10-07T01:02:03.123456Z").unwrap(); + let request = f + .schema() + .runtime_create_request( + &f.collection(), + &[ + (f.field("tenant"), RuntimeValue::String("north".into())), + (f.field("label"), RuntimeValue::String("exact".into())), + (f.field("json"), RuntimeValue::Json(serde_json::Value::Null)), + (f.field("bytes"), RuntimeValue::Bytes(vec![0, 255, 128])), + (f.field("score"), RuntimeValue::Integer(i64::MIN)), + ( + f.field("occurred"), + RuntimeValue::DateTime(datetime.clone()), + ), + ], + Some(projection.clone()), + Default::default(), + ) + .unwrap(); + let row = f.run(request, f.authority()).await.unwrap(); + let record = row.record().unwrap(); + assert!(record.uuid(&f.field("id")).is_ok()); + assert!(matches!( + record.value(&f.field("json")).unwrap(), + RuntimeValue::Json(serde_json::Value::Null) + )); + assert_eq!(record.bytes(&f.field("bytes")).unwrap(), &[0, 255, 128]); + assert_eq!(record.datetime(&f.field("occurred")).unwrap(), &datetime); + assert_eq!(record.integer(&f.field("score")).unwrap(), i64::MIN); + let key = f + .schema() + .runtime_key( + &f.collection(), + &[(f.field("id"), record.value(&f.field("id")).unwrap().clone())], + ) + .unwrap(); + let request = f + .schema() + .runtime_update_request( + key, + &[ + (f.field("json"), RuntimeValue::Null), + (f.field("score"), RuntimeValue::Integer(i64::MAX)), + ], + None, + Some(projection), + Default::default(), + ) + .unwrap(); + let row = f.run(request, f.authority()).await.unwrap(); + assert!(matches!( + row.record().unwrap().state(&f.field("json")).unwrap(), + RuntimeFieldState::Null + )); + assert_eq!( + row.record().unwrap().integer(&f.field("score")).unwrap(), + i64::MAX + ); + f.finish().await; +} +#[tokio::test] +async fn complete_composite_keys_and_required_create_members() { + let mut schema = definition(); + schema.collections[0] + .primary_key + .push(FieldId::new("tenant").unwrap()); + let f = Fixture::with_definition(schema).await; + assert!( + f.schema() + .runtime_key( + &f.collection(), + &[(f.field("id"), RuntimeValue::Integer(1))] + ) + .is_err() + ); + f.run(f.create(1), f.authority()).await.unwrap(); + let key = f + .schema() + .runtime_key( + &f.collection(), + &[ + (f.field("tenant"), RuntimeValue::String("north".into())), + (f.field("id"), RuntimeValue::Integer(1)), + ], + ) + .unwrap(); + assert_eq!(key.fields()[0].0.id(), f.field("id").id()); + let request = f + .schema() + .runtime_delete_request(key, None, None, Default::default()) + .unwrap(); + f.run(request, f.authority()).await.unwrap(); + assert_eq!(f.count().await, 0); + f.finish().await; +} +#[tokio::test] +async fn incoming_modifying_dependencies_fail_closed_while_restrict_remains_usable() { + for action in [ + DeletePolicy::Cascade, + DeletePolicy::SetNull, + DeletePolicy::Restrict, + ] { + let mut schema = definition(); + schema.collections[0].fields.push(RuntimeField { + id: FieldId::new("parent").unwrap(), + api_name: "parent".into(), + physical_column: "parent".into(), + value_kind: RuntimeValueKind::Integer, + nullable: true, + unique: false, + filterable: false, + sortable: false, + generated: false, + default: None, + }); + schema.collections[0].relations.push(RuntimeRelation { + id: RelationId::new("parent_link").unwrap(), + api_name: "parentLink".into(), + target: CollectionId::new("records").unwrap(), + key_pairs: vec![RelationKeyPair { + source: FieldId::new("parent").unwrap(), + target: FieldId::new("id").unwrap(), + }], + cardinality: RelationCardinality::One, + enforce_foreign_key: true, + on_delete: Some(action.clone()), + }); + let f = Fixture::with_definition(schema).await; + f.run(f.create(1), f.authority()).await.unwrap(); + let request = f + .schema() + .runtime_delete_request(f.key(1), None, None, Default::default()) + .unwrap(); + let result = f.run(request, f.authority()).await; + if action == DeletePolicy::Restrict { + assert!(result.is_ok()); + } else { + assert_eq!( + result.unwrap_err().public_error().runtime_mutation_code(), + Some("unsupported_cascade") + ); + assert_eq!(f.count().await, 1); + } + f.finish().await; + } +} + +struct PendingIntent; +impl RuntimeWriteAuthority for PendingIntent { + fn authorize_intent<'a>( + &'a self, + _: RuntimeWriteIntent<'a>, + _: &'a mut MutationContext<'_, Backend>, + ) -> BoxFuture<'a, Result> { + Box::pin(std::future::pending()) + } + fn authorize_preimage<'a>( + &'a self, + _: RuntimePreimageCheck<'a>, + _: &'a mut MutationContext<'_, Backend>, + ) -> BoxFuture<'a, Result<(), RuntimeMutationError>> { + Box::pin(async { panic!("preimage must not be reached") }) + } + fn authorize_result<'a>( + &'a self, + _: RuntimeResultCheck<'a>, + _: &'a mut MutationContext<'_, Backend>, + ) -> BoxFuture<'a, Result> { + Box::pin(async { panic!("result must not be reached") }) + } +} +#[tokio::test] +async fn cancellation_before_dml_also_poisons_preceding_host_journal_work() { + let f = Fixture::new().await; + let environment = f.environment.clone(); + let request = f.create(1); + let result = + f.db.transaction(TransactionMode::StateMachine, move |tx| { + Box::pin(async move { + tx.insert::(CreateJournalInput { + id: "before".into(), + message: "staged".into(), + }) + .await?; + assert!( + tokio::time::timeout( + Duration::from_millis(50), + tx.mutate_runtime(&environment, &request, &PendingIntent) + ) + .await + .is_err() + ); + Ok(()) + }) + }) + .await; + assert!(matches!(result, Err(TransactionError::Rejected(_)))); + assert_eq!(f.count().await, 0); + assert_eq!(Journal::count_all(&f.db).await.unwrap(), 0); + f.finish().await; +} +#[test] +fn bounded_inputs_and_predicate_values_reject_without_database_access() { + let schema = definition().validate().unwrap(); + let c = schema + .resolve_collection(&CollectionId::new("records").unwrap()) + .unwrap(); + let id = schema + .resolve_field(&c, &FieldId::new("id").unwrap()) + .unwrap(); + let label = schema + .resolve_field(&c, &FieldId::new("label").unwrap()) + .unwrap(); + let key = schema + .runtime_key(&c, &[(id, RuntimeValue::Integer(1))]) + .unwrap(); + let limits = RuntimeMutationLimits { + max_value_bytes: 64, + ..Default::default() + }; + let value = RuntimeValue::String("fixture-private-value".repeat(10)); + let error = schema + .runtime_update_request( + key.clone(), + &[(label.clone(), value.clone())], + None, + None, + limits, + ) + .unwrap_err(); + assert_eq!(error.code(), RuntimeMutationErrorCode::LimitExceeded); + assert!(!format!("{error:?}").contains("fixture-private")); + let expected = schema + .runtime_compare( + &c, + &label, + RuntimeScalarOperator::Eq, + value, + Default::default(), + ) + .unwrap(); + assert_eq!( + schema + .runtime_update_request( + key, + &[(label, RuntimeValue::String("x".into()))], + Some(expected), + None, + limits + ) + .unwrap_err() + .code(), + RuntimeMutationErrorCode::LimitExceeded + ); +} + +#[tokio::test] +async fn combined_key_cas_and_authority_bind_budget_rejects_before_target_reads() { + let f = Fixture::new().await; + let expected = f + .schema() + .runtime_compare( + &f.collection(), + &f.field("score"), + RuntimeScalarOperator::Eq, + RuntimeValue::Integer(7), + Default::default(), + ) + .unwrap(); + let limits = RuntimeMutationLimits { + query: RuntimeQueryLimits { + max_bind_parameters: 3, + ..Default::default() + }, + ..Default::default() + }; + let request = f + .schema() + .runtime_update_request( + f.key(1), + &[(f.field("label"), RuntimeValue::String("changed".into()))], + Some(expected), + None, + limits, + ) + .unwrap(); + ::execute_with_binds(f.db.pool(), "DROP TABLE runtime_records", &[]) + .await + .unwrap(); + assert_eq!( + f.run(request, f.authority()) + .await + .unwrap_err() + .public_error() + .runtime_mutation_code(), + Some("limit_exceeded") + ); + f.finish().await; +} + +#[tokio::test] +async fn host_policy_only_revision_changes_reject_unchanged_schema_requests() { + let f = Fixture::new().await; + let fingerprint = f.schema().fingerprint(); + let request = f.create(1); + let mut authority = f.authority(); + // Host obtains these values from its pinned policy generation, not the ORM schema. + authority.pinned_policy_revision = 2; + let error = f.run(request, authority).await.unwrap_err(); + assert_eq!( + error.public_error().runtime_mutation_code(), + Some("schema_mismatch") + ); + assert_eq!(f.schema().fingerprint(), fingerprint); + assert_eq!(f.count().await, 0); + let mut fresh = f.authority(); + fresh.pinned_policy_revision = 2; + fresh.expected_policy_revision = 2; + f.run(f.create(1), fresh).await.unwrap(); + f.finish().await; +} + +// PostgreSQL executes only rejection cases until the reviewed #112/A/B reconciliation. +// The renderer's native-slot accounting is independently covered by unit tests. +#[tokio::test] +async fn emitted_suffix_bind_budgets_precede_target_reads_for_cas_and_authority() { + let f = Fixture::new().await; + let suffix = f + .schema() + .runtime_compare( + &f.collection(), + &f.field("label"), + RuntimeScalarOperator::EndsWith, + RuntimeValue::String("tial".into()), + Default::default(), + ) + .unwrap(); + let score = f + .schema() + .runtime_compare( + &f.collection(), + &f.field("score"), + RuntimeScalarOperator::Eq, + RuntimeValue::Integer(7), + Default::default(), + ) + .unwrap(); + let blocked = f + .schema() + .runtime_compare( + &f.collection(), + &f.field("label"), + RuntimeScalarOperator::Eq, + RuntimeValue::String("blocked".into()), + Default::default(), + ) + .unwrap(); + let not_blocked = f + .schema() + .runtime_not(&f.collection(), blocked, Default::default()) + .unwrap(); + let either = f + .schema() + .runtime_or( + &f.collection(), + vec![suffix.clone(), not_blocked], + Default::default(), + ) + .unwrap(); + let nested_cas = f + .schema() + .runtime_and( + &f.collection(), + vec![ + suffix.clone(), + f.schema() + .runtime_or( + &f.collection(), + vec![score.clone(), score.clone()], + Default::default(), + ) + .unwrap(), + ], + Default::default(), + ) + .unwrap(); + let tenant = f.authority().predicate.unwrap(); + let nested_authority = f + .schema() + .runtime_and( + &f.collection(), + vec![tenant.clone(), either], + Default::default(), + ) + .unwrap(); + for (case, (expected, predicate, sqlite_slots, postgres_slots)) in [ + (suffix, tenant.clone(), 5, 4), + (nested_cas.clone(), tenant, 7, 6), + (score, nested_authority.clone(), 7, 6), + (nested_cas, nested_authority, 10, 8), + ] + .into_iter() + .enumerate() + { + let required = if Backend::DIALECT == DatabaseBackend::Sqlite { + sqlite_slots + } else { + postgres_slots + }; + for budget in [required - 1, required, required + 1] { + if Backend::DIALECT == DatabaseBackend::Postgres && budget >= required { + continue; + } + let id = (case * 3 + budget + 2 - required) as i64; + f.run(f.create(id), f.authority()).await.unwrap(); + let request = f + .schema() + .runtime_update_request( + f.key(id), + &[(f.field("score"), RuntimeValue::Integer(8))], + Some(expected.clone()), + None, + RuntimeMutationLimits { + query: RuntimeQueryLimits { + max_bind_parameters: budget, + ..Default::default() + }, + ..Default::default() + }, + ) + .unwrap(); + let mut authority = f.authority(); + authority.predicate = Some(predicate.clone()); + let calls = authority.preimage_calls.clone(); + let result = f.run(request, authority).await; + if budget < required { + assert_eq!( + result.unwrap_err().public_error().runtime_mutation_code(), + Some("limit_exceeded") + ); + assert_eq!( + calls.load(Ordering::SeqCst), + 0, + "case {case}: preimage authorization ran before rejecting the bind budget" + ); + } else { + result.unwrap(); + assert_eq!(calls.load(Ordering::SeqCst), 1); + } + let sql = format!( + "SELECT score FROM runtime_records WHERE id = {}", + Backend::DIALECT.placeholder(1) + ); + let rows = Backend::fetch_rows(f.db.pool(), &sql, &[SqlValue::Int(id)]) + .await + .unwrap(); + assert_eq!( + Backend::try_get_i64(&rows[0], "score").unwrap(), + if budget < required { 7 } else { 8 } + ); + } + } + f.finish().await; +} diff --git a/docs/plans/active/runtime-platform-capabilities/README.md b/docs/plans/active/runtime-platform-capabilities/README.md index 064271fe..c05d92cb 100644 --- a/docs/plans/active/runtime-platform-capabilities/README.md +++ b/docs/plans/active/runtime-platform-capabilities/README.md @@ -1117,38 +1117,23 @@ is advertised until its contract is committed and reviewed. ## Current checkpoint The revised A–D contract at `c53a5de966ca5089a5e4aa2d727e144bae83e6bd` -is approved. The independent repository aggregate fix is merged and included -in the owner-published 0.33.3 baseline. PR A implements owned targets and -read-only planning, explicit ownership and separate guarded application in -ORM/macros 0.40.0; its unpublished identity is reconciled onto released -0.39.0 main without changing the approved interfaces. Host-managed timestamps and -portable static aggregate capabilities remain intact. Its compiled host example and isolated SQLite/PostgreSQL -tests accompany the implementation. It remains subject to owner review. - -A's live FK capability checks reject unsupported actions/deferral with scoped -`UnsupportedForeignKey` diagnostics, including relevant system/unowned incoming -sources, and recheck them on the pinned apply transaction. The legacy physical -model/static APIs are unchanged. Successful SQLite FK restoration retains its -connection; cancellation or failed restoration discards an unsafe lease. ORM API -reproductions at the reviewed head confirm constraint loss. Regressions cover -structured FK rejection, successful in-memory rebuild/no-op replanning, -cancellation, incoming constraints introduced after planning, and unchanged -PostgreSQL system RLS. The approved legacy datetime rejection and -representable static/runtime equivalence gates remain covered. - -Review correction R3 reproduces PostgreSQL DateTime no-op/certification failure -through owned plan/apply APIs. The shared planner now compares the built-in -`TIMESTAMPTZ` / `timestamp with time zone` alias without rewriting hashes or -storage metadata. Required/nullable DateTime targets with no default and -`CurrentTimestamp`, including static/system composition, have executable -SQLite/PostgreSQL fixed-point gates. Timezone-free timestamps and precision -changes remain distinct. This correction awaits exact-head re-review. Prefer -owner merge/release of independent #112 first, then reconcile A with that exact -merged source; Actions remain paused. - -B–D remain approved proposals with unimplemented interfaces and require reviewed -committed predecessors. Keep the operation/hook cancellation, host-managed AEAD -cursor protection, lossless JSON text, required create keys and host policy-only -revision guarantees during implementation. Independent compiled static -relationships/query/view work is coordinated separately and does not require -B–D or dynamic GraphQL. No releases or consumer repository changes are included. +is approved. Digibase accepts A at `f2665ce68ed7fdbccb272cb8a783b6379ae5fd82`: +FK corrections R1/R2 and PostgreSQL DateTime correction R3 are resolved. A's +reviewed-predecessor gate for B is satisfied. A remains unmerged/unreleased, +with ORM/macros 0.40.0. Owner merge/release of independent #112 precedes A's final +reconciliation; Actions remain paused. + +B's independent implementation is based on that exact committed A revision, +with aligned unpublished ORM/macros 0.41.0. It adds schema-bound requests, +explicit intent/preimage/result authority, pinned structural CAS, exact return +projections, atomic repository journal writes and guarded operation/hook +cancellation. It remains subject to exact-head review. Its reference and compiled +host example describe environment/policy revision pinning, backend gates and +ambiguous commit limitations. The static-only inner-timeout limitation remains. + +C/D remain approved proposals with unimplemented interfaces, dependent on their +respective reviewed predecessors. Preserve host-managed AEAD cursor protection, +lossless JSON text, required create keys and independent post-commit payload +read authorization. Reconcile B onto the final reviewed A after #112 integration +and rerun combined backend contracts before downstream adoption. No product or +sibling repository changes, owner merge or release publication are included. diff --git a/docs/reference/graphql-orm/README.md b/docs/reference/graphql-orm/README.md index 5b699012..ce2f05b2 100644 --- a/docs/reference/graphql-orm/README.md +++ b/docs/reference/graphql-orm/README.md @@ -31,6 +31,7 @@ guidance for the core ORM package. ## Generated and repository operations - [Runtime writes, hooks, subscriptions, and policies](runtime-and-writes.md) +- [Schema-bound transactional runtime mutations](runtime-mutations.md) - [Composite mutations](composite-mutations.md) - [Repository-only entities](repository-only-entities.md) - [Read projections](read-projections.md) diff --git a/docs/reference/graphql-orm/runtime-mutations.md b/docs/reference/graphql-orm/runtime-mutations.md new file mode 100644 index 00000000..62457117 --- /dev/null +++ b/docs/reference/graphql-orm/runtime-mutations.md @@ -0,0 +1,156 @@ +--- +title: Schema-bound transactional runtime mutations +kind: reference +status: active +owner: graphql-orm-maintainers +last_reviewed: 2026-10-07 +review_by: 2027-02-01 +supersedes: [] +--- + +# Schema-bound transactional runtime mutations + +ORM/macros 0.41.0 (unreleased) adds single-record runtime create, patch and +complete-key delete. Select `sqlite` or `postgres`; no dynamic GraphQL feature +is needed. This builds on [owned runtime migrations](runtime-migrations.md). +The [compiled host example](../../../crates/graphql-orm/examples/runtime_mutation_journal.rs) +executes create, CAS update and delete with atomic repository journal inserts: + +```sh +CARGO_BUILD_JOBS=2 cargo run -p graphql-orm --no-default-features --features sqlite --example runtime_mutation_journal +CARGO_BUILD_JOBS=2 cargo run -p graphql-orm --no-default-features --features postgres --example runtime_mutation_journal +``` + +SQLite uses process-owned memory. PostgreSQL starts a labelled disposable Docker +server and verifies its identity before cleanup. Neither example consumes an +ambient database URL. The external repository consumer compiles this example +without a direct async-graphql dependency. + +## Requests and execution + +`ValidatedRuntimeSchema::runtime_key` validates a complete, non-null key and +orders members according to the declared primary key. Construct requests with +`runtime_create_request`, `runtime_update_request` and `runtime_delete_request`. +Requests are immutable, owned and fingerprint-bound. Keys remain immutable after +create; append-only collections reject update/delete. Duplicate, generated, +wrong-kind and stale input fields fail closed. Non-generated required create +members must be supplied unless they have an explicit default. Generated UUIDs +are minted by the ORM; other generators are unsupported in this bounded profile. + +```rust,ignore +let request = schema.runtime_update_request( + schema.runtime_key(&collection, &[(id, RuntimeValue::Integer(7))])?, + &[(status, RuntimeValue::String("published".into()))], + Some(expected_draft), Some(public_projection), RuntimeMutationLimits::default(), +)?; +let committed = database.transaction_with_auth( + TransactionMode::StateMachine, db_auth.as_ref(), + |tx| Box::pin(async move { + let pending = tx.mutate_runtime(&environment, &request, &authority).await?; + tx.run_runtime_before_commit(&journal_hook, &pending).await?; + Ok(pending) + }), +).await?; +// Only now is committed work available for a success response or notification. +``` + +The environment comes from A's `runtime_mutation_environment` after validation +of the composed physical target and managed ownership. It includes incoming +system/unowned dependencies and the backend identity. A host must pin this +certificate with its schema generation and fence external DDL; it is not a +lease that makes arbitrary external schema changes safe. Revalidate after drift. +Incoming delete CASCADE/SET NULL rejects with `unsupported_cascade`, even when +no child currently exists. RESTRICT remains usable; database constraint failures +roll back normally. No migration or pool query occurs inside the mutation engine. + +SQLite acquires the write lock before callback reads; PostgreSQL uses the +existing serializable transaction and locks an existing exact-key row. Intent, +preimage, structural CAS, DML, actual-result reread/authorization and host journal +work share this transaction. PostgreSQL `DbAuthContext` remains transaction-local. +Other backends explicitly lack `RuntimeMutationBackend` support or report its +capability constant false; low-level execution rejects before statements. Adapters +must check this capability before opening a transaction. No live MSSQL support +is implied. + +## Authority and private projections + +Implement every `RuntimeWriteAuthority` method; no default allow exists: + +- `authorize_intent`: validate the pinned identity and complete host public/policy + revision, collection action, every supplied field, CAS field and requested return + field. Policy-only revision changes are host state, separate from ORM fingerprints. +- Return `RuntimeWriteGrant::new(policy_projection, row_predicate)`. The projection + selects the minimum internal policy fields. The engine unions keys, changed/CAS + fields and requested output privately. The row predicate limits preimages and + actual create/update results. A result outside it denies and rolls back. +- `authorize_preimage`: check the authoritative visible row before update/delete. + Missing/invisible rows return `not_found`; conflicts disclose no current value + and are distinguished only after preimage authorization. +- `authorize_result`: approve the actual staged record including server defaults + and trigger effects, or the preimage for delete. Return + `RuntimeReturnGrant::new` with exactly the requested projection. Write approval + is not read approval; narrowing fields or replacing protected values with NULL + is not supported. `returning: None` returns no internal record. + +`RuntimePredicate::referenced_fields` supports independent CAS-field approval. +Predicates retain existing typed operator/filterability restrictions. SQL NULL +checks work; JSON equality is not added. Complete-key DML repeats the expected +condition and must affect exactly one row. Serializable/deadlock/busy errors are +transaction failures, distinct from deterministic `conflict`. + +The example's authority deliberately approves its isolated public fixture. It is +not a production policy implementation. Host authorities and hooks must not send +external notifications or open independent pool transactions for policy decisions. + +## Values and bounds + +Omitted create fields use their declared defaults/generators, otherwise SQL NULL +when nullable, otherwise `missing_required`. Omitted patches preserve the stored +value. Explicit `RuntimeValue::Null` writes SQL NULL and never requests a default. +`RuntimeValue::Json(JSON null)` remains distinct. Integers, bytes, UUIDs and canonical +runtime DateTime values retain existing storage semantics; no epoch-unit conversion +or legacy DateTime reinterpretation is introduced. The engine reselects actual +stored results before authorization, including defaults and trigger effects. + +`RuntimeMutationLimits` bounds input count, key arity, serialized value bytes, +record bytes and JSON depth, plus existing predicate/bind/projection limits. +Predicate bind budgets use the shared backend renderer's emitted value slots, +including duplicated SQLite suffix operands and repeated PostgreSQL references to +one native slot. Combined CAS/authority budgets reject before target reads. +Defaults: 128 inputs, 16 key members, 1 MiB per value, 4 MiB per record and JSON +depth 32. Bounds also apply to internal policy records; oversized decoding fails +and poisons the transaction. Driver row decoding still allocates returned values; +these limits are not a driver-level streaming memory cap. + +## Cancellation, rollback and commit + +An operation guard starts on the first poll before any await. Only complete +success disarms it. Error/drop marks the transaction rollback-only; an outstanding +or forgotten guard independently prevents commit. The outer runner and +`commit_and_emit` both enforce this. A caught denial, conflict, limit error or +inner timeout cannot commit staged runtime work, even if later operations succeed +and the callback returns `Ok`. An unpolled future has staged nothing. + +Use `run_runtime_before_commit` for journal hooks. Its separate guard covers hook +writes and awaits, including cancellation after mutation success. Propagate errors +from direct repository journal writes. Calling a hook directly or catching an +unrelated static repository error is not this guarded hook contract. + +Whole-runner cancellation retains transaction-drop rollback. Commit cancellation +or transport failure can leave an ambiguous outcome. Existing `TransactionError` +variants remain; `commit_outcome()` reports confirmed `RolledBack` only for +`Rejected`, conservatively `Unknown` for `Failed`/`Retryable`. Do not blindly retry +unknown outcomes. `OrmPublicError::runtime_mutation_code()` retains safe runtime +codes through callback error conversion; no caller values, SQL or preimages are +formatted by runtime request/grant/effect/error Debug implementations. + +`RuntimeMutationEffect` is pending, not an event or commit receipt. The engine emits +no events. Static repository events/actions remain deferred until commit. Durable +journal storage and replay are host responsibilities. Static-only transactions +retain their existing limitation: catching an inner static mutation timeout and +returning `Ok` does not guarantee rollback. + +Bulk/nested writes, new upsert semantics, dynamic GraphQL and durable delivery are +outside B. C/D remain separate reviewed successors. Owner merge/release of #112 +and final reconciliation of A/B are required before combined downstream adoption; +this branch does not claim to contain the independent string-binding fix. diff --git a/docs/reference/workspace-packages.md b/docs/reference/workspace-packages.md index 40dfd994..3d2d99d3 100644 --- a/docs/reference/workspace-packages.md +++ b/docs/reference/workspace-packages.md @@ -18,11 +18,11 @@ changes. | Package | Version | Path | Default features | Direct internal dependencies | | --- | --- | --- | --- | --- | -| `graphql-orm` | `0.40.0` | `crates/graphql-orm` | `sqlite` | `graphql-orm-macros`, `graphql-orm-operation-catalog`, `graphql-orm-router-protocol` (dev-only) | +| `graphql-orm` | `0.41.0` | `crates/graphql-orm` | `sqlite` | `graphql-orm-macros`, `graphql-orm-operation-catalog`, `graphql-orm-router-protocol` (dev-only) | | `graphql-orm-ai` | `0.106.1` | `crates/graphql-orm-ai` | `sqlite` | `graphql-orm`, `graphql-orm-ai-tool-profiles`, `graphql-orm-storage` | | `graphql-orm-ai-tool-profiles` | `0.15.0` | `crates/graphql-orm-ai-tool-profiles` | none | `graphql-orm-operation-catalog`, `graphql-orm-router-protocol` (dev-only) | | `graphql-orm-backup` | `0.7.2` | `crates/graphql-orm-backup` | `local` | `graphql-orm` (optional), `graphql-orm-storage` | -| `graphql-orm-macros` | `0.40.0` | `crates/graphql-orm-macros` | `sqlite` | none | +| `graphql-orm-macros` | `0.41.0` | `crates/graphql-orm-macros` | `sqlite` | none | | `graphql-orm-operation-catalog` | `0.4.0` | `crates/graphql-orm-operation-catalog` | none | `graphql-orm-router-protocol` (optional) | | `graphql-orm-router` | `0.5.4` | `crates/graphql-orm-router` | none | `graphql-orm-router-protocol` | | `graphql-orm-router-protocol` | `0.2.1` | `crates/graphql-orm-router-protocol` | none | none | diff --git a/scripts/run-owned-database-lanes.sh b/scripts/run-owned-database-lanes.sh index 13dc0ea0..cc6a292e 100755 --- a/scripts/run-owned-database-lanes.sh +++ b/scripts/run-owned-database-lanes.sh @@ -27,7 +27,7 @@ cd "${repository_root}" run_sqlite() { cargo test -p graphql-orm --locked --no-default-features --features sqlite \ - --test runtime_migrations --test runtime_schema_ir --test runtime_migration_memory \ + --test runtime_mutations --test runtime_migrations --test runtime_schema_ir --test runtime_migration_memory \ --test migration_planner --test migration_apply --test legacy_migration_history \ -- --test-threads=1 cargo test -p graphql-orm --no-default-features --features sqlite \ @@ -39,7 +39,7 @@ run_sqlite() { run_postgres() { cargo test -p graphql-orm --locked --no-default-features --features postgres \ - --test runtime_migrations --test runtime_schema_ir --test runtime_migration_memory \ + --test runtime_mutations --test runtime_migrations --test runtime_schema_ir --test runtime_migration_memory \ --test migration_planner --test migration_apply --test legacy_migration_history \ -- --test-threads=1 cargo test -p graphql-orm --no-default-features --features postgres \