From 96ea0b530b26105bae0f1051d32d7e94ac1c44b5 Mon Sep 17 00:00:00 2001 From: Brian Bartman Date: Tue, 11 Aug 2026 15:37:38 +0000 Subject: [PATCH 1/9] validation, control-plane-api: Snapshot authorization model and retryable staleness error Adds validation::Error::AuthorizationSnapshotStale and is_authz_snapshot_stale() as the retryable signal that an authorization denial was evaluated against a Snapshot which may not yet reflect a committed grant. Adds the in-process authorization surface to Snapshot: the three-way Authorization enum (Authorized / Denied / Stale) with ok_or_stale(), user_authorization / role_authorization over the snapshot's grants, spec_capabilities() for error-message rendering, and STALE_RETRY_WAKE. Also renames App::snapshot to App::snapshot_watch, ahead of follow-ups where executors pin per-poll Snapshot tokens distinct from the watch itself. This PR is purely additive; nothing consumes the new surface yet. Split 1 of 4 from #3155. --- crates/agent/src/integration_tests/harness.rs | 2 +- crates/control-plane-api/src/envelope.rs | 2 +- crates/control-plane-api/src/lib.rs | 2 +- crates/control-plane-api/src/server/mod.rs | 6 +- .../control-plane-api/src/server/snapshot.rs | 293 ++++++++++++++++++ crates/validation/src/errors.rs | 16 + crates/validation/src/lib.rs | 2 +- 7 files changed, 316 insertions(+), 7 deletions(-) diff --git a/crates/agent/src/integration_tests/harness.rs b/crates/agent/src/integration_tests/harness.rs index a78cc34140f..47a9ca2f757 100644 --- a/crates/agent/src/integration_tests/harness.rs +++ b/crates/agent/src/integration_tests/harness.rs @@ -1589,7 +1589,7 @@ impl TestHarness { maybe_claims: control_plane_api::MaybeControlClaims::with_verified(verified), original_uri: axum::http::Uri::from_static("/graphql"), pg_pool: self.pool.clone(), - refresh: app.snapshot.token(), + refresh: app.snapshot_watch.token(), retry_after: tokens::DateTime::UNIX_EPOCH, started: tokens::now(), locale: control_plane_api::Locale::EnUS, diff --git a/crates/control-plane-api/src/envelope.rs b/crates/control-plane-api/src/envelope.rs index 3ca7d720771..ef6b28cc57f 100644 --- a/crates/control-plane-api/src/envelope.rs +++ b/crates/control-plane-api/src/envelope.rs @@ -272,7 +272,7 @@ impl axum::extract::FromRequestParts> for Envelope { Ok(Envelope { maybe_claims, retry_after: retry_after.unwrap_or(tokens::DateTime::UNIX_EPOCH), - refresh: state.snapshot.token(), + refresh: state.snapshot_watch.token(), started: started.unwrap_or_else(|| tokens::now()), pg_pool: state.pg_pool.clone(), original_uri, diff --git a/crates/control-plane-api/src/lib.rs b/crates/control-plane-api/src/lib.rs index 856766a693e..ca16775c095 100644 --- a/crates/control-plane-api/src/lib.rs +++ b/crates/control-plane-api/src/lib.rs @@ -53,7 +53,7 @@ pub use envelope::{Envelope, Locale, MaybeControlClaims}; pub(crate) use server::evaluate_names_authorization; pub use server::{ ApiError, App, AuthZRetry, build_router, - snapshot::{self, Snapshot}, + snapshot::{self, Authorization, Snapshot}, }; // Re-export the GraphQL schema SDL function for flow-client build script diff --git a/crates/control-plane-api/src/server/mod.rs b/crates/control-plane-api/src/server/mod.rs index e3eaef0fe71..406393688d2 100644 --- a/crates/control-plane-api/src/server/mod.rs +++ b/crates/control-plane-api/src/server/mod.rs @@ -39,7 +39,7 @@ pub struct App { pub control_plane_jwt_encode_key: tokens::jwt::EncodingKey, pub pg_pool: sqlx::PgPool, pub publisher: crate::publications::Publisher, - pub snapshot: Arc>, + pub snapshot_watch: Arc>, /// Signing secret for verifying inbound Stripe webhook deliveries. `None` /// when unconfigured, in which case the webhook endpoint fails closed rather /// than trusting any request. See `server::public::stripe_webhooks`. @@ -53,7 +53,7 @@ impl App { jwt_secret: &[u8], pg_pool: sqlx::PgPool, publisher: crate::publications::Publisher, - snapshot: Arc>, + snapshot_watch: Arc>, stripe_webhook_secret: Option, ) -> Self { Self { @@ -63,7 +63,7 @@ impl App { control_plane_jwt_encode_key: tokens::jwt::EncodingKey::from_secret(jwt_secret), pg_pool, publisher, - snapshot, + snapshot_watch, stripe_webhook_secret, } } diff --git a/crates/control-plane-api/src/server/snapshot.rs b/crates/control-plane-api/src/server/snapshot.rs index 9c81400bfc0..2281a60ae3e 100644 --- a/crates/control-plane-api/src/server/snapshot.rs +++ b/crates/control-plane-api/src/server/snapshot.rs @@ -73,6 +73,37 @@ pub struct SnapshotTask { pub data_plane_id: models::Id, } +/// Outcome of an authorization check evaluated against a Snapshot, +/// classified by `Snapshot::resolve_authorization`. +#[must_use] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Authorization { + /// The required grant exists in the Snapshot. + Authorized, + /// The grant is absent and the Snapshot is authoritative for the + /// operation's anchor: the denial is final. + Denied, + /// The grant is absent but the Snapshot predates the anchor: a grant + /// committed before the anchor may not be reflected yet, so the denial is + /// provisional and the operation should retry under a fresher Snapshot. + Stale, +} + +impl Authorization { + /// Collapse to "is authorized?", surfacing a provisional denial as the + /// retryable `AuthorizationSnapshotStale` error which callers + /// (see `validation::is_authz_snapshot_stale`) convert into a retry. + pub fn ok_or_stale(self, catalog_name: &str) -> Result { + match self { + Authorization::Authorized => Ok(true), + Authorization::Denied => Ok(false), + Authorization::Stale => Err(validation::Error::AuthorizationSnapshotStale { + catalog_name: catalog_name.to_string(), + }), + } + } +} + // SnapshotMigration is the state of an underway data-plane migration. #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] pub struct SnapshotMigration { @@ -181,6 +212,68 @@ impl Snapshot { self.taken > (started + Self::TEMPORAL_SKEW) } + /// Classify an already-evaluated authorization check against this + /// Snapshot's freshness: the single three-way policy — authorized / + /// authoritative denial / provisional denial — applied at every snapshot + /// authorization enforcement point. + /// + /// A denial is `Denied` only when this Snapshot was taken after `anchor`, + /// the instant the asking operation started: any grant committed before + /// the anchor is then necessarily reflected. Otherwise it is `Stale` — + /// possibly just unobserved. `None` means the caller has no instant to + /// anchor a staleness claim on, so denials are final. + pub fn resolve_authorization( + &self, + authorized: bool, + anchor: Option, + ) -> Authorization { + if authorized { + Authorization::Authorized + } else if anchor.is_none_or(|anchor| self.taken_after(anchor)) { + Authorization::Denied + } else { + Authorization::Stale + } + } + + /// Evaluate whether `user_id` holds `capability` to `name` under this + /// Snapshot's grants, classified against `anchor` freshness + /// (see `resolve_authorization`). + pub fn user_authorization( + &self, + user_id: uuid::Uuid, + name: &str, + capability: impl Into, + anchor: Option, + ) -> Authorization { + self.resolve_authorization( + tables::UserGrant::is_authorized( + &self.role_grants, + &self.user_grants, + user_id, + name, + capability, + ), + anchor, + ) + } + + /// Evaluate whether `subject` (a catalog spec acting as a role) holds + /// `capability` to `object` under this Snapshot's role grants, classified + /// against `anchor` freshness (see `resolve_authorization`). + pub fn role_authorization( + &self, + subject: &str, + object: &str, + capability: impl Into, + anchor: Option, + ) -> Authorization { + self.resolve_authorization( + tables::RoleGrant::is_authorized(&self.role_grants, subject, object, capability), + anchor, + ) + } + // Retrieve all tasks whose names start with the given `prefix`. pub fn tasks_by_prefix<'s>( &'s self, @@ -341,9 +434,26 @@ impl Snapshot { }) } + /// Returns the "spec capabilities" of a spec named `catalog_name`: the role + /// grants whose `subject_role` is a prefix of the name — the capabilities the + /// spec holds by virtue of its own name/role. This is only to be used for error + /// reporting to improve error messages. + pub fn spec_capabilities(&self, catalog_name: &str) -> Vec { + self.role_grants + .iter() + .filter(|grant| catalog_name.starts_with(grant.subject_role.as_str())) + .cloned() + .collect() + } + // Minimal interval between Snapshot refreshes. // We will postpone a requested refresh prior to this interval. pub const MIN_REFRESH_INTERVAL: chrono::TimeDelta = chrono::TimeDelta::seconds(20); + /// Re-poll cadence for a queued task which is deferring until its + /// Snapshot is authoritative (see `taken_after`). This equals + /// `MIN_REFRESH_INTERVAL` because that's the soonest a refresh can land: + /// waking sooner burns polls, waking later delays the task. + pub const STALE_RETRY_WAKE: chrono::TimeDelta = Self::MIN_REFRESH_INTERVAL; // Maximum interval between Snapshot refreshes. // We will refresh an older Snapshot in the background. pub const MAX_REFRESH_INTERVAL: chrono::TimeDelta = chrono::TimeDelta::minutes(5); @@ -837,4 +947,187 @@ mod tests { chrono::DateTime::from_timestamp(300_000, 0).unwrap() ); } + + /// `taken_after` is the single definition of "this Snapshot is authoritative + /// for that instant", and every authorization-staleness decision routes + /// through it. The `TEMPORAL_SKEW` allowance and the strictness of the + /// comparison are therefore load-bearing, so pin both. + #[test] + fn test_taken_after_allows_for_temporal_skew() { + let started = chrono::DateTime::from_timestamp(1_000_000, 0).unwrap(); + let at = |offset: chrono::TimeDelta| Snapshot { + taken: started + offset, + ..Snapshot::empty() + }; + + assert!( + !at(chrono::TimeDelta::zero()).taken_after(started), + "a Snapshot taken at the same instant is not authoritative" + ); + assert!( + !at(-Snapshot::TEMPORAL_SKEW).taken_after(started), + "a Snapshot taken before the event is not authoritative" + ); + assert!( + !at(Snapshot::TEMPORAL_SKEW).taken_after(started), + "the skew allowance is exclusive: exactly TEMPORAL_SKEW later is still not authoritative" + ); + assert!( + at(Snapshot::TEMPORAL_SKEW + chrono::TimeDelta::milliseconds(1)).taken_after(started), + "one millisecond past the skew allowance is authoritative" + ); + } + + /// `resolve_authorization` is the shared three-way classifier behind every + /// snapshot authorization enforcement point. Pin its anchor semantics — + /// a denial is authoritative only under a Snapshot postdating the anchor, + /// and a `None` anchor makes denials final — and `ok_or_stale`'s collapse + /// into authorized / dropped / retryable. + #[test] + fn test_resolve_authorization() { + let anchor = chrono::DateTime::from_timestamp(1_000_000, 0).unwrap(); + let stale = Snapshot { + taken: anchor, + ..Snapshot::empty() + }; + let fresh = Snapshot { + taken: anchor + Snapshot::TEMPORAL_SKEW * 2, + ..Snapshot::empty() + }; + + // A held grant is Authorized regardless of freshness. + assert_eq!( + Authorization::Authorized, + stale.resolve_authorization(true, Some(anchor)) + ); + assert_eq!( + Authorization::Authorized, + stale.resolve_authorization(true, None) + ); + + // A denial is authoritative only under a Snapshot postdating the anchor. + assert_eq!( + Authorization::Denied, + fresh.resolve_authorization(false, Some(anchor)) + ); + assert_eq!( + Authorization::Stale, + stale.resolve_authorization(false, Some(anchor)) + ); + + // Without an anchor there is no basis for a staleness claim. + assert_eq!( + Authorization::Denied, + stale.resolve_authorization(false, None) + ); + + assert!(matches!( + Authorization::Authorized.ok_or_stale("acmeCo/task"), + Ok(true) + )); + assert!(matches!( + Authorization::Denied.ok_or_stale("acmeCo/task"), + Ok(false) + )); + assert!(matches!( + Authorization::Stale.ok_or_stale("acmeCo/task"), + Err(validation::Error::AuthorizationSnapshotStale { catalog_name }) + if catalog_name == "acmeCo/task" + )); + } + + /// `spec_capabilities` replaced a SQL-computed `spec_capabilities` column and + /// now renders the "Available grants are:" list in publication authorization + /// errors. It answers "what may a spec named X do, by virtue of its own + /// name?", which is a prefix match on `subject_role` — not on `object_role`, + /// and not scoped to any user. + #[test] + fn test_spec_capabilities() { + let snapshot = Snapshot::build_fixture(None); + let subjects = |name: &str| { + snapshot + .spec_capabilities(name) + .into_iter() + .map(|g| { + ( + g.subject_role.to_string(), + g.object_role.to_string(), + g.capability, + ) + }) + .collect::>() + }; + + // A name under a granted prefix picks up every grant whose subject_role + // is a prefix of it — here both the tenant-wide grants and the more + // specific `bobCo/tires/` one. + insta::assert_debug_snapshot!(subjects("bobCo/tires/source-tread"), @r#" + [ + ( + "bobCo/", + "bobCo/", + Write, + ), + ( + "bobCo/", + "ops/dp/public/", + Read, + ), + ( + "bobCo/tires/", + "acmeCo/shared/", + Read, + ), + ] + "#); + + // The narrower `bobCo/tires/` grant must not leak to a sibling prefix: + // subject matching is by prefix of the *name*, not by shared tenancy. + insta::assert_debug_snapshot!(subjects("bobCo/widgets/source-squash"), @r#" + [ + ( + "bobCo/", + "bobCo/", + Write, + ), + ( + "bobCo/", + "ops/dp/public/", + Read, + ), + ] + "#); + + // `subject_role` is matched as a prefix of the name, so the role itself + // qualifies. + assert_eq!( + vec![( + "bobCo/tires/".to_string(), + "acmeCo/shared/".to_string(), + models::Capability::Read + )], + subjects("bobCo/tires/") + .into_iter() + .filter(|(s, _, _)| s == "bobCo/tires/") + .collect::>(), + ); + + // Grants are not matched by their object_role: `acmeCo/shared/` is + // reachable *from* `bobCo/tires/`, but a spec named `acmeCo/shared/x` + // holds only `acmeCo/`'s own grants. + insta::assert_debug_snapshot!(subjects("acmeCo/shared/thing"), @r#" + [ + ( + "acmeCo/", + "acmeCo/", + Write, + ), + ] + "#); + + assert!( + subjects("unknownCo/thing").is_empty(), + "a name under no granted prefix holds nothing" + ); + } } diff --git a/crates/validation/src/errors.rs b/crates/validation/src/errors.rs index 63fa9b2198f..187a5d17df6 100644 --- a/crates/validation/src/errors.rs +++ b/crates/validation/src/errors.rs @@ -299,6 +299,10 @@ pub enum Error { build_id: models::Id, larger_id: models::Id, }, + #[error( + "authorization for {catalog_name} was evaluated against a control-plane snapshot that is not authoritative for this operation; please retry the operation" + )] + AuthorizationSnapshotStale { catalog_name: String }, #[error( "This spec was updated while you were editing — please refresh and re-apply your changes.\nThis may have been an automated system update. (expected publication ID {expect_id}, actual {actual_id})" )] @@ -413,3 +417,15 @@ impl Error { errors.insert_row(scope.flatten(), anyhow::anyhow!(self)); } } + +/// Returns true if `err` is (or wraps) an [`Error::AuthorizationSnapshotStale`]. +/// This classifies a *retryable* authorization failure: the decision was made +/// against a control-plane snapshot that is not authoritative for the operation, +/// so it should be retried against a fresher snapshot rather than surfaced as a +/// terminal error. +pub fn is_authz_snapshot_stale(err: &anyhow::Error) -> bool { + matches!( + err.downcast_ref::(), + Some(Error::AuthorizationSnapshotStale { .. }) + ) +} diff --git a/crates/validation/src/lib.rs b/crates/validation/src/lib.rs index 1f9c0c31743..ea0b90f0f0c 100644 --- a/crates/validation/src/lib.rs +++ b/crates/validation/src/lib.rs @@ -17,7 +17,7 @@ mod schema; mod storage_mapping; mod test_step; -pub use errors::Error; +pub use errors::{Error, is_authz_snapshot_stale}; pub use noop::NoOpConnectors; /// Portion of the binding namespace reserved for runtime-internal bindings From e8b2a780973a5f6878bf6cf7f546ca9aa1e284ac Mon Sep 17 00:00:00 2001 From: Alex Bair Date: Tue, 11 Aug 2026 22:30:50 -0400 Subject: [PATCH 2/9] mise: pin RocksDB's target ISA instead of inheriting the builder's `build:rocksdb` passed PORTABLE=0, which RocksDB reads as `-march=native`, so librocksdb.a -- and every binary linking it -- took its instruction floor from whichever machine compiled it. One master build drew an AVX-512 runner and produced a flowctl that SIGILLs on any host without AVX-512, breaking most estuary/connectors CI jobs. The same library goes into the reactor and control-plane images, three of which deploy automatically from Platform Build. This commit pins haswell on x86_64, the level we had already been shipping: reactor:v0.6.13-11-gd9006db953c has zero AVX-512 instructions under RocksDB symbols, and the same source line compiles to vinserti128 there and vinserti64x2 after. Note: we should avoid x86-64-v3, which lacks the PCLMUL that RocksDB's crc32c fast path needs, and PORTABLE=1, which sets no -march at all. aarch64 adds no -march under either setting, so it is unchanged. --- mise/tasks/build/rocksdb | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/mise/tasks/build/rocksdb b/mise/tasks/build/rocksdb index 428c4832a9f..26063af33be 100755 --- a/mise/tasks/build/rocksdb +++ b/mise/tasks/build/rocksdb @@ -31,11 +31,21 @@ export ROCKSDB_DISABLE_LZ4=1 export ROCKSDB_DISABLE_ZLIB=1 export ROCKSDB_DISABLE_ZSTD=1 +# PORTABLE=0 means `-march=native`, so librocksdb.a inherited the builder's ISA: +# an AVX-512 runner once produced binaries that SIGILL everywhere else. haswell +# pins the AVX2 level we had been shipping. Avoid x86-64-v3 -- RocksDB falls back +# to a slower software checksum under it, since v3 lacks PCLMUL. aarch64 adds no +# -march under either setting, so 1 leaves it unchanged. +case "$(uname -m)" in + x86_64) PORTABLE_ARCH=haswell ;; + *) PORTABLE_ARCH=1 ;; +esac + make static_lib -j$(nproc 2>/dev/null || sysctl -n hw.ncpu 2>/dev/null || echo 4) \ EXTRA_CXXFLAGS="-fPIC -frtti" \ EXTRA_CFLAGS="-fPIC" \ DEBUG_LEVEL=0 \ - PORTABLE=0 + PORTABLE="${PORTABLE_ARCH}" strip --strip-unneeded "${SOURCE_DIR}/librocksdb.a" From 6ba6ecf67d4fb5e3e921b40d045ce338042cac45 Mon Sep 17 00:00:00 2001 From: James Date: Thu, 13 Aug 2026 07:26:58 +1000 Subject: [PATCH 3/9] docs: correct the default materialization maxTxnDuration (#3347) The default was raised from 5m to 20m in d09cb56d034 (June 2025), but three descriptions of it were left behind: the ShardTemplate doc comment, its generated entry in flow.schema.json, and the hand-written table in the Configuring Task Shards page. No behavior change. The 20 minute default in assemble is intentional and unchanged; only the text describing it is corrected. flow.schema.json is edited in place rather than regenerated, because the released flowctl predates syncSchedule and regenerating with it would drop valid schema. The edited string is byte-identical to what an in-tree regeneration emits. --- crates/models/src/shards.rs | 2 +- flow.schema.json | 2 +- site/docs/features/configuring-task-shards.md | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/models/src/shards.rs b/crates/models/src/shards.rs index a546fe6e348..ab594fa197b 100644 --- a/crates/models/src/shards.rs +++ b/crates/models/src/shards.rs @@ -32,7 +32,7 @@ pub struct ShardTemplate { /// may process documents before it must flush and commit. /// It may run for less time if there aren't additional ready documents for /// it to process. - /// If not set, the maximum duration defaults to five minutes for materializations, + /// If not set, the maximum duration defaults to twenty minutes for materializations, /// and one second for captures and derivations. /// EXPERIMENTAL: this field MAY be removed. #[serde( diff --git a/flow.schema.json b/flow.schema.json index 31b5fc6b359..4379b4f581b 100644 --- a/flow.schema.json +++ b/flow.schema.json @@ -1281,7 +1281,7 @@ }, "maxTxnDuration": { "title": "Maximum duration of task transactions.", - "description": "This duration upper-bounds the amount of time during which a transaction\nmay process documents before it must flush and commit.\nIt may run for less time if there aren't additional ready documents for\nit to process.\nIf not set, the maximum duration defaults to five minutes for materializations,\nand one second for captures and derivations.\nEXPERIMENTAL: this field MAY be removed.", + "description": "This duration upper-bounds the amount of time during which a transaction\nmay process documents before it must flush and commit.\nIt may run for less time if there aren't additional ready documents for\nit to process.\nIf not set, the maximum duration defaults to twenty minutes for materializations,\nand one second for captures and derivations.\nEXPERIMENTAL: this field MAY be removed.", "type": [ "string", "null" diff --git a/site/docs/features/configuring-task-shards.md b/site/docs/features/configuring-task-shards.md index edd27f3342b..aa4e9384ecc 100644 --- a/site/docs/features/configuring-task-shards.md +++ b/site/docs/features/configuring-task-shards.md @@ -15,7 +15,7 @@ You do this by adding the `shards` configuration to the capture or materializati |---|---|---|---| | `/disable` | Disable | Disable processing of the task's shards. | Boolean | | `/logLevel` | Log level | Log levels may currently be \"error\", \"warn\", \"info\", \"debug\", or \"trace\". If not set, the effective log level is \"info\". | String | -| `/maxTxnDuration` | Maximum transaction duration | This duration upper-bounds the amount of time during which a transaction may process documents before it must initiate a commit. Note that it may take some additional time for the commit to complete after it is initiated. The shard may run for less time if there aren't additional ready documents for it to process. If not set, the maximum duration defaults to one second for captures and derivations, and 5 minutes for materializations. | String | +| `/maxTxnDuration` | Maximum transaction duration | This duration upper-bounds the amount of time during which a transaction may process documents before it must initiate a commit. Note that it may take some additional time for the commit to complete after it is initiated. The shard may run for less time if there aren't additional ready documents for it to process. If not set, the maximum duration defaults to one second for captures and derivations, and 20 minutes for materializations. | String | | `/minTxnDuration` | Minimum transaction duration | This duration lower-bounds the amount of time during which a transaction must process documents before it must flush and commit. It may run for more time if additional documents are available. The default value is zero seconds. | String | For more information about these controls and when you might need to use them, see: From a0878f9c9b54551fb1697dbd9aff4cb023bb528d Mon Sep 17 00:00:00 2001 From: Daniel Nelson Date: Fri, 14 Aug 2026 09:58:46 -0700 Subject: [PATCH 4/9] sources: update snapshot for catalog schema (#3360) --- .../snapshots/schema_generation__catalog_schema_snapshot.snap | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/sources/tests/snapshots/schema_generation__catalog_schema_snapshot.snap b/crates/sources/tests/snapshots/schema_generation__catalog_schema_snapshot.snap index de606792003..c2badf7ef95 100644 --- a/crates/sources/tests/snapshots/schema_generation__catalog_schema_snapshot.snap +++ b/crates/sources/tests/snapshots/schema_generation__catalog_schema_snapshot.snap @@ -1285,7 +1285,7 @@ expression: "&schema" }, "maxTxnDuration": { "title": "Maximum duration of task transactions.", - "description": "This duration upper-bounds the amount of time during which a transaction\nmay process documents before it must flush and commit.\nIt may run for less time if there aren't additional ready documents for\nit to process.\nIf not set, the maximum duration defaults to five minutes for materializations,\nand one second for captures and derivations.\nEXPERIMENTAL: this field MAY be removed.", + "description": "This duration upper-bounds the amount of time during which a transaction\nmay process documents before it must flush and commit.\nIt may run for less time if there aren't additional ready documents for\nit to process.\nIf not set, the maximum duration defaults to twenty minutes for materializations,\nand one second for captures and derivations.\nEXPERIMENTAL: this field MAY be removed.", "type": [ "string", "null" From 6b808e70aa0d601aae00e3709ba5dc722c6f5087 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Se=C3=A1n=20Whelan?= Date: Mon, 17 Aug 2026 08:19:53 +0100 Subject: [PATCH 5/9] data-plane-controller: pass through aws_resource_tags stack config field (#3355) BYOC customers enrolled in AWS's Migration Acceleration Program need a cost-allocation tag (e.g. map-migrated) applied to every AWS resource of their data-plane, including hosts est-dry-dock replaces over time via Pulumi's provider default_tags. data-plane-controller re-serializes data_planes.config into the Pulumi stack config with no serde catch-all, so a field added only on the est-dry-dock/Pydantic side would be silently dropped before Pulumi ever sees it. This adds the Rust side of that field so it survives the round trip; it must ship and deploy before the est-dry-dock change. --- .../data-plane-controller/src/shared/stack.rs | 41 +++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/crates/data-plane-controller/src/shared/stack.rs b/crates/data-plane-controller/src/shared/stack.rs index a9642ebb41f..8d9454a7ecc 100644 --- a/crates/data-plane-controller/src/shared/stack.rs +++ b/crates/data-plane-controller/src/shared/stack.rs @@ -152,6 +152,13 @@ pub struct DataPlane { // IP at S3 changes from the NAT EIP to the private-subnet IP. #[serde(default, skip_serializing_if = "is_false")] pub s3_endpoint_on_private_subnet: bool, + // Extra tags applied to every AWS resource of the data-plane, through the + // Pulumi AWS provider's `default_tags`. Used for customer cost-allocation + // schemes which require a tag on resources we create and replace over + // time, such as the `map-migrated` tag of the AWS Migration Acceleration + // Program. AWS only; Azure and GCP deployments ignore it. + #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")] + pub aws_resource_tags: std::collections::BTreeMap, pub deployments: Vec, #[serde(default, skip_serializing_if = "Option::is_none")] pub connector_limits: Option, @@ -729,6 +736,40 @@ mod test { ); } + // We deserialize `data_planes.config` and re-serialize it into the Pulumi + // stack config, so `aws_resource_tags` reaches est-dry-dock only if it + // survives that round trip. An absent map must stay absent, so that + // data-planes which set no tags see no change to their stack config. + #[test] + fn aws_resource_tags_round_trip() { + let State { stack, .. } = serde_json::from_str(include_str!("state_fixture.json")).unwrap(); + + assert!(stack.config.model.aws_resource_tags.is_empty()); + assert_eq!( + serde_json::to_value(&stack.config.model) + .unwrap() + .get("aws_resource_tags"), + None, + ); + + let tagged = serde_json::to_value(DataPlane { + aws_resource_tags: [("map-migrated".to_string(), "migAQWO5XH8V0".to_string())].into(), + ..stack.config.model.clone() + }) + .unwrap(); + + assert_eq!( + tagged.get("aws_resource_tags").unwrap(), + &serde_json::json!({"map-migrated": "migAQWO5XH8V0"}), + ); + assert_eq!( + serde_json::from_value::(tagged) + .unwrap() + .aws_resource_tags, + [("map-migrated".to_string(), "migAQWO5XH8V0".to_string())].into(), + ); + } + // The wire shape est-dry-dock parses: `id` is a sibling of the flattened // link config, omitted when None, and round-trips losslessly. #[test] From cd51dccf869926e4c5999b3cba2a750582e5fa84 Mon Sep 17 00:00:00 2001 From: James Date: Mon, 17 Aug 2026 19:43:44 +1000 Subject: [PATCH 6/9] docs: fix broken anchor, enforce anchor checking, document macOS build gotcha (#3272) The firstWriteWins/lastWriteWins page linked ./#reduction-guarantees (itself) instead of ../#reduction-guarantees (the reduction-strategies overview page, where that heading actually lives). With the only broken anchor site-wide now fixed, flip onBrokenAnchors from warn to throw so future broken anchors fail CI instead of shipping silently. Also documents the pre-existing hubspot-real-time redirect collision: it only reproduces on macOS's case-insensitive filesystem and doesn't affect the Linux-based production build, so it's a local-dev note rather than a bug fix. --- site/README.md | 9 +++++++++ site/docusaurus.config.js | 2 +- 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/site/README.md b/site/README.md index 3e39ea5429f..da53ce6d579 100644 --- a/site/README.md +++ b/site/README.md @@ -26,6 +26,15 @@ $ npm build This command generates static content into the `build` directory and can be served using any static contents hosting service. +**macOS note:** `npm run build` fails locally on macOS's default case-insensitive +filesystem with `The redirect plugin is not supposed to override existing files` +for the `hubspot-real-time` redirect in `docusaurus.config.js`. That redirect's +`from` path differs from the real `HubSpot-real-time` page only by case, so the +plugin sees a collision when writing the redirect stub. This does not affect +production: CI builds on Linux (case-sensitive), so the two paths are genuinely +distinct there. To build locally on macOS, temporarily comment out that redirect +entry, or build inside a case-sensitive volume/Docker. + ### Deployment Using SSH: diff --git a/site/docusaurus.config.js b/site/docusaurus.config.js index 61d85b9c638..183be88ec7d 100644 --- a/site/docusaurus.config.js +++ b/site/docusaurus.config.js @@ -46,7 +46,7 @@ const config = { tagline: 'Fast, configurable ETL your way', url: URL, baseUrl: BASE_URL, - onBrokenAnchors: 'warn', // TODO(johnny): Fix broken links and make this 'throw'. + onBrokenAnchors: 'throw', onBrokenLinks: 'throw', favicon: 'img/favicon-2.ico', organizationName: 'estuary', From 7466d95517656b0879f5ca10a215034884f03332 Mon Sep 17 00:00:00 2001 From: James Date: Mon, 17 Aug 2026 19:44:11 +1000 Subject: [PATCH 7/9] docs: reference gcloud storage instead of gsutil for GCS bucket IAM (#3220) Google is removing gsutil from the default gcloud CLI bundle after March 2027; gcloud storage is the supported replacement. --- site/docs/getting-started/installation.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/site/docs/getting-started/installation.md b/site/docs/getting-started/installation.md index 411d976b132..683ca60f88d 100644 --- a/site/docs/getting-started/installation.md +++ b/site/docs/getting-started/installation.md @@ -97,7 +97,7 @@ For a [GCS bucket](https://cloud.google.com/storage/docs/creating-buckets), upda * [`roles/storage.admin`](https://cloud.google.com/storage/docs/access-control/iam-roles) as the **role** You can configure this through the [Cloud Console](https://console.cloud.google.com/) or -the `gsutil` CLI. +the `gcloud storage` CLI. ### Amazon S3 From cd8d2713c8565c21124a0194c33739f9dc7bcbab Mon Sep 17 00:00:00 2001 From: James Date: Mon, 17 Aug 2026 19:45:06 +1000 Subject: [PATCH 8/9] docs: clarify sync schedule connector scope and backfill heuristic (#3233) * docs: clarify sync schedule connector scope and backfill heuristic Sync schedules are used by warehouse connectors AND file/object-store connectors (S3/GCS/Azure Blob file sinks, Iceberg), not just warehouses. Transactional databases (Postgres, MySQL, SQL Server) and streaming/API destinations do not use them. Also document how a connector distinguishes catching-up from backfilling (recent transaction sizes) so the delay-skip behavior during backfills is explicit. * modify language on upload interval connectors --------- Co-authored-by: aeluce --- .../materialization-sync-schedule.md | 23 +++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/site/docs/guides/customize-dataflows/materialization-sync-schedule.md b/site/docs/guides/customize-dataflows/materialization-sync-schedule.md index ae36c72eb0a..2816d72bcc8 100644 --- a/site/docs/guides/customize-dataflows/materialization-sync-schedule.md +++ b/site/docs/guides/customize-dataflows/materialization-sync-schedule.md @@ -18,8 +18,18 @@ reduce the costs incurred in the destination from the actions the connector takes to load data to it. ::: -Most warehouse materialization connectors (such as [Snowflake](/reference/Connectors/materialization-connectors/Snowflake), [Databricks](/reference/Connectors/materialization-connectors/databricks), and [BigQuery](/reference/Connectors/materialization-connectors/BigQuery)) support configuring a sync schedule. -Check the connector reference docs to determine if a specific connector supports sync schedules. +Sync schedules are supported by **warehouse** materialization connectors, such as [Snowflake](/reference/Connectors/materialization-connectors/Snowflake), [Databricks](/reference/Connectors/materialization-connectors/databricks), [BigQuery](/reference/Connectors/materialization-connectors/BigQuery), Amazon Redshift, MotherDuck, ClickHouse, and Azure Fabric Warehouse. + +In contrast, transactional database materializations (such as PostgreSQL, MySQL, and SQL Server) and streaming or API destinations (such as Elasticsearch, MongoDB, DynamoDB, and Pinecone) do _not_ use a sync schedule; they apply updates as they arrive. + +The connector reference docs will note whether a specific connector supports sync schedules. + +:::note +File and object-store connectors support a similar concept to sync schedules. +These connectors (such as the Amazon S3, Google Cloud Storage, and Azure Blob Storage file materializations) use an `uploadInterval` property or similar to manage data load frequency. + +They do not support other sync schedule functionality like fast sync times. +::: ## How transactions are used to sync data to a destination @@ -43,6 +53,15 @@ fully caught up - backfills always run as fast as possible. And while a transaction is delayed, Estuary will continue batching and combining new documents so that the next transaction contains all of the latest data. +The connector decides whether it is caught up or still backfilling by looking at +the sizes of its recent transactions. If any of the last several transactions +stored a large number of documents (on the order of a million), the +materialization is assumed to still be backfilling, and the sync schedule delay +is skipped so that it can catch up as fast as possible. Once several consecutive +transactions are all below that size, the materialization is treated as caught +up and the sync schedule delay applies. This behavior is the same for every +connector that supports a sync schedule or upload interval. + You can read about [how continuous materialization works](/concepts/materialization/#how-continuous-materialization-works) for more background information. From c674aabfe29990f8fd757a90c40d4240c563af1c Mon Sep 17 00:00:00 2001 From: Brian Bartman Date: Tue, 11 Aug 2026 15:37:38 +0000 Subject: [PATCH 9/9] validation, control-plane-api: Snapshot authorization model and retryable staleness error Adds validation::Error::AuthorizationSnapshotStale and is_authz_snapshot_stale() as the retryable signal that an authorization denial was evaluated against a Snapshot which may not yet reflect a committed grant. Adds the in-process authorization surface to Snapshot: the three-way Authorization enum (Authorized / Denied / Stale) with ok_or_stale(), user_authorization / role_authorization over the snapshot's grants, spec_capabilities() for error-message rendering, and STALE_RETRY_WAKE. Also renames App::snapshot to App::snapshot_watch, ahead of follow-ups where executors pin per-poll Snapshot tokens distinct from the watch itself. This PR is purely additive; nothing consumes the new surface yet. Split 1 of 4 from #3155. --- crates/agent/src/integration_tests/harness.rs | 2 +- crates/control-plane-api/src/envelope.rs | 2 +- crates/control-plane-api/src/lib.rs | 2 +- crates/control-plane-api/src/server/mod.rs | 6 +- .../control-plane-api/src/server/snapshot.rs | 293 ++++++++++++++++++ crates/validation/src/errors.rs | 16 + crates/validation/src/lib.rs | 2 +- 7 files changed, 316 insertions(+), 7 deletions(-) diff --git a/crates/agent/src/integration_tests/harness.rs b/crates/agent/src/integration_tests/harness.rs index a78cc34140f..47a9ca2f757 100644 --- a/crates/agent/src/integration_tests/harness.rs +++ b/crates/agent/src/integration_tests/harness.rs @@ -1589,7 +1589,7 @@ impl TestHarness { maybe_claims: control_plane_api::MaybeControlClaims::with_verified(verified), original_uri: axum::http::Uri::from_static("/graphql"), pg_pool: self.pool.clone(), - refresh: app.snapshot.token(), + refresh: app.snapshot_watch.token(), retry_after: tokens::DateTime::UNIX_EPOCH, started: tokens::now(), locale: control_plane_api::Locale::EnUS, diff --git a/crates/control-plane-api/src/envelope.rs b/crates/control-plane-api/src/envelope.rs index 3ca7d720771..ef6b28cc57f 100644 --- a/crates/control-plane-api/src/envelope.rs +++ b/crates/control-plane-api/src/envelope.rs @@ -272,7 +272,7 @@ impl axum::extract::FromRequestParts> for Envelope { Ok(Envelope { maybe_claims, retry_after: retry_after.unwrap_or(tokens::DateTime::UNIX_EPOCH), - refresh: state.snapshot.token(), + refresh: state.snapshot_watch.token(), started: started.unwrap_or_else(|| tokens::now()), pg_pool: state.pg_pool.clone(), original_uri, diff --git a/crates/control-plane-api/src/lib.rs b/crates/control-plane-api/src/lib.rs index 856766a693e..ca16775c095 100644 --- a/crates/control-plane-api/src/lib.rs +++ b/crates/control-plane-api/src/lib.rs @@ -53,7 +53,7 @@ pub use envelope::{Envelope, Locale, MaybeControlClaims}; pub(crate) use server::evaluate_names_authorization; pub use server::{ ApiError, App, AuthZRetry, build_router, - snapshot::{self, Snapshot}, + snapshot::{self, Authorization, Snapshot}, }; // Re-export the GraphQL schema SDL function for flow-client build script diff --git a/crates/control-plane-api/src/server/mod.rs b/crates/control-plane-api/src/server/mod.rs index e3eaef0fe71..406393688d2 100644 --- a/crates/control-plane-api/src/server/mod.rs +++ b/crates/control-plane-api/src/server/mod.rs @@ -39,7 +39,7 @@ pub struct App { pub control_plane_jwt_encode_key: tokens::jwt::EncodingKey, pub pg_pool: sqlx::PgPool, pub publisher: crate::publications::Publisher, - pub snapshot: Arc>, + pub snapshot_watch: Arc>, /// Signing secret for verifying inbound Stripe webhook deliveries. `None` /// when unconfigured, in which case the webhook endpoint fails closed rather /// than trusting any request. See `server::public::stripe_webhooks`. @@ -53,7 +53,7 @@ impl App { jwt_secret: &[u8], pg_pool: sqlx::PgPool, publisher: crate::publications::Publisher, - snapshot: Arc>, + snapshot_watch: Arc>, stripe_webhook_secret: Option, ) -> Self { Self { @@ -63,7 +63,7 @@ impl App { control_plane_jwt_encode_key: tokens::jwt::EncodingKey::from_secret(jwt_secret), pg_pool, publisher, - snapshot, + snapshot_watch, stripe_webhook_secret, } } diff --git a/crates/control-plane-api/src/server/snapshot.rs b/crates/control-plane-api/src/server/snapshot.rs index 9c81400bfc0..2281a60ae3e 100644 --- a/crates/control-plane-api/src/server/snapshot.rs +++ b/crates/control-plane-api/src/server/snapshot.rs @@ -73,6 +73,37 @@ pub struct SnapshotTask { pub data_plane_id: models::Id, } +/// Outcome of an authorization check evaluated against a Snapshot, +/// classified by `Snapshot::resolve_authorization`. +#[must_use] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Authorization { + /// The required grant exists in the Snapshot. + Authorized, + /// The grant is absent and the Snapshot is authoritative for the + /// operation's anchor: the denial is final. + Denied, + /// The grant is absent but the Snapshot predates the anchor: a grant + /// committed before the anchor may not be reflected yet, so the denial is + /// provisional and the operation should retry under a fresher Snapshot. + Stale, +} + +impl Authorization { + /// Collapse to "is authorized?", surfacing a provisional denial as the + /// retryable `AuthorizationSnapshotStale` error which callers + /// (see `validation::is_authz_snapshot_stale`) convert into a retry. + pub fn ok_or_stale(self, catalog_name: &str) -> Result { + match self { + Authorization::Authorized => Ok(true), + Authorization::Denied => Ok(false), + Authorization::Stale => Err(validation::Error::AuthorizationSnapshotStale { + catalog_name: catalog_name.to_string(), + }), + } + } +} + // SnapshotMigration is the state of an underway data-plane migration. #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] pub struct SnapshotMigration { @@ -181,6 +212,68 @@ impl Snapshot { self.taken > (started + Self::TEMPORAL_SKEW) } + /// Classify an already-evaluated authorization check against this + /// Snapshot's freshness: the single three-way policy — authorized / + /// authoritative denial / provisional denial — applied at every snapshot + /// authorization enforcement point. + /// + /// A denial is `Denied` only when this Snapshot was taken after `anchor`, + /// the instant the asking operation started: any grant committed before + /// the anchor is then necessarily reflected. Otherwise it is `Stale` — + /// possibly just unobserved. `None` means the caller has no instant to + /// anchor a staleness claim on, so denials are final. + pub fn resolve_authorization( + &self, + authorized: bool, + anchor: Option, + ) -> Authorization { + if authorized { + Authorization::Authorized + } else if anchor.is_none_or(|anchor| self.taken_after(anchor)) { + Authorization::Denied + } else { + Authorization::Stale + } + } + + /// Evaluate whether `user_id` holds `capability` to `name` under this + /// Snapshot's grants, classified against `anchor` freshness + /// (see `resolve_authorization`). + pub fn user_authorization( + &self, + user_id: uuid::Uuid, + name: &str, + capability: impl Into, + anchor: Option, + ) -> Authorization { + self.resolve_authorization( + tables::UserGrant::is_authorized( + &self.role_grants, + &self.user_grants, + user_id, + name, + capability, + ), + anchor, + ) + } + + /// Evaluate whether `subject` (a catalog spec acting as a role) holds + /// `capability` to `object` under this Snapshot's role grants, classified + /// against `anchor` freshness (see `resolve_authorization`). + pub fn role_authorization( + &self, + subject: &str, + object: &str, + capability: impl Into, + anchor: Option, + ) -> Authorization { + self.resolve_authorization( + tables::RoleGrant::is_authorized(&self.role_grants, subject, object, capability), + anchor, + ) + } + // Retrieve all tasks whose names start with the given `prefix`. pub fn tasks_by_prefix<'s>( &'s self, @@ -341,9 +434,26 @@ impl Snapshot { }) } + /// Returns the "spec capabilities" of a spec named `catalog_name`: the role + /// grants whose `subject_role` is a prefix of the name — the capabilities the + /// spec holds by virtue of its own name/role. This is only to be used for error + /// reporting to improve error messages. + pub fn spec_capabilities(&self, catalog_name: &str) -> Vec { + self.role_grants + .iter() + .filter(|grant| catalog_name.starts_with(grant.subject_role.as_str())) + .cloned() + .collect() + } + // Minimal interval between Snapshot refreshes. // We will postpone a requested refresh prior to this interval. pub const MIN_REFRESH_INTERVAL: chrono::TimeDelta = chrono::TimeDelta::seconds(20); + /// Re-poll cadence for a queued task which is deferring until its + /// Snapshot is authoritative (see `taken_after`). This equals + /// `MIN_REFRESH_INTERVAL` because that's the soonest a refresh can land: + /// waking sooner burns polls, waking later delays the task. + pub const STALE_RETRY_WAKE: chrono::TimeDelta = Self::MIN_REFRESH_INTERVAL; // Maximum interval between Snapshot refreshes. // We will refresh an older Snapshot in the background. pub const MAX_REFRESH_INTERVAL: chrono::TimeDelta = chrono::TimeDelta::minutes(5); @@ -837,4 +947,187 @@ mod tests { chrono::DateTime::from_timestamp(300_000, 0).unwrap() ); } + + /// `taken_after` is the single definition of "this Snapshot is authoritative + /// for that instant", and every authorization-staleness decision routes + /// through it. The `TEMPORAL_SKEW` allowance and the strictness of the + /// comparison are therefore load-bearing, so pin both. + #[test] + fn test_taken_after_allows_for_temporal_skew() { + let started = chrono::DateTime::from_timestamp(1_000_000, 0).unwrap(); + let at = |offset: chrono::TimeDelta| Snapshot { + taken: started + offset, + ..Snapshot::empty() + }; + + assert!( + !at(chrono::TimeDelta::zero()).taken_after(started), + "a Snapshot taken at the same instant is not authoritative" + ); + assert!( + !at(-Snapshot::TEMPORAL_SKEW).taken_after(started), + "a Snapshot taken before the event is not authoritative" + ); + assert!( + !at(Snapshot::TEMPORAL_SKEW).taken_after(started), + "the skew allowance is exclusive: exactly TEMPORAL_SKEW later is still not authoritative" + ); + assert!( + at(Snapshot::TEMPORAL_SKEW + chrono::TimeDelta::milliseconds(1)).taken_after(started), + "one millisecond past the skew allowance is authoritative" + ); + } + + /// `resolve_authorization` is the shared three-way classifier behind every + /// snapshot authorization enforcement point. Pin its anchor semantics — + /// a denial is authoritative only under a Snapshot postdating the anchor, + /// and a `None` anchor makes denials final — and `ok_or_stale`'s collapse + /// into authorized / dropped / retryable. + #[test] + fn test_resolve_authorization() { + let anchor = chrono::DateTime::from_timestamp(1_000_000, 0).unwrap(); + let stale = Snapshot { + taken: anchor, + ..Snapshot::empty() + }; + let fresh = Snapshot { + taken: anchor + Snapshot::TEMPORAL_SKEW * 2, + ..Snapshot::empty() + }; + + // A held grant is Authorized regardless of freshness. + assert_eq!( + Authorization::Authorized, + stale.resolve_authorization(true, Some(anchor)) + ); + assert_eq!( + Authorization::Authorized, + stale.resolve_authorization(true, None) + ); + + // A denial is authoritative only under a Snapshot postdating the anchor. + assert_eq!( + Authorization::Denied, + fresh.resolve_authorization(false, Some(anchor)) + ); + assert_eq!( + Authorization::Stale, + stale.resolve_authorization(false, Some(anchor)) + ); + + // Without an anchor there is no basis for a staleness claim. + assert_eq!( + Authorization::Denied, + stale.resolve_authorization(false, None) + ); + + assert!(matches!( + Authorization::Authorized.ok_or_stale("acmeCo/task"), + Ok(true) + )); + assert!(matches!( + Authorization::Denied.ok_or_stale("acmeCo/task"), + Ok(false) + )); + assert!(matches!( + Authorization::Stale.ok_or_stale("acmeCo/task"), + Err(validation::Error::AuthorizationSnapshotStale { catalog_name }) + if catalog_name == "acmeCo/task" + )); + } + + /// `spec_capabilities` replaced a SQL-computed `spec_capabilities` column and + /// now renders the "Available grants are:" list in publication authorization + /// errors. It answers "what may a spec named X do, by virtue of its own + /// name?", which is a prefix match on `subject_role` — not on `object_role`, + /// and not scoped to any user. + #[test] + fn test_spec_capabilities() { + let snapshot = Snapshot::build_fixture(None); + let subjects = |name: &str| { + snapshot + .spec_capabilities(name) + .into_iter() + .map(|g| { + ( + g.subject_role.to_string(), + g.object_role.to_string(), + g.capability, + ) + }) + .collect::>() + }; + + // A name under a granted prefix picks up every grant whose subject_role + // is a prefix of it — here both the tenant-wide grants and the more + // specific `bobCo/tires/` one. + insta::assert_debug_snapshot!(subjects("bobCo/tires/source-tread"), @r#" + [ + ( + "bobCo/", + "bobCo/", + Write, + ), + ( + "bobCo/", + "ops/dp/public/", + Read, + ), + ( + "bobCo/tires/", + "acmeCo/shared/", + Read, + ), + ] + "#); + + // The narrower `bobCo/tires/` grant must not leak to a sibling prefix: + // subject matching is by prefix of the *name*, not by shared tenancy. + insta::assert_debug_snapshot!(subjects("bobCo/widgets/source-squash"), @r#" + [ + ( + "bobCo/", + "bobCo/", + Write, + ), + ( + "bobCo/", + "ops/dp/public/", + Read, + ), + ] + "#); + + // `subject_role` is matched as a prefix of the name, so the role itself + // qualifies. + assert_eq!( + vec![( + "bobCo/tires/".to_string(), + "acmeCo/shared/".to_string(), + models::Capability::Read + )], + subjects("bobCo/tires/") + .into_iter() + .filter(|(s, _, _)| s == "bobCo/tires/") + .collect::>(), + ); + + // Grants are not matched by their object_role: `acmeCo/shared/` is + // reachable *from* `bobCo/tires/`, but a spec named `acmeCo/shared/x` + // holds only `acmeCo/`'s own grants. + insta::assert_debug_snapshot!(subjects("acmeCo/shared/thing"), @r#" + [ + ( + "acmeCo/", + "acmeCo/", + Write, + ), + ] + "#); + + assert!( + subjects("unknownCo/thing").is_empty(), + "a name under no granted prefix holds nothing" + ); + } } diff --git a/crates/validation/src/errors.rs b/crates/validation/src/errors.rs index 63fa9b2198f..187a5d17df6 100644 --- a/crates/validation/src/errors.rs +++ b/crates/validation/src/errors.rs @@ -299,6 +299,10 @@ pub enum Error { build_id: models::Id, larger_id: models::Id, }, + #[error( + "authorization for {catalog_name} was evaluated against a control-plane snapshot that is not authoritative for this operation; please retry the operation" + )] + AuthorizationSnapshotStale { catalog_name: String }, #[error( "This spec was updated while you were editing — please refresh and re-apply your changes.\nThis may have been an automated system update. (expected publication ID {expect_id}, actual {actual_id})" )] @@ -413,3 +417,15 @@ impl Error { errors.insert_row(scope.flatten(), anyhow::anyhow!(self)); } } + +/// Returns true if `err` is (or wraps) an [`Error::AuthorizationSnapshotStale`]. +/// This classifies a *retryable* authorization failure: the decision was made +/// against a control-plane snapshot that is not authoritative for the operation, +/// so it should be retried against a fresher snapshot rather than surfaced as a +/// terminal error. +pub fn is_authz_snapshot_stale(err: &anyhow::Error) -> bool { + matches!( + err.downcast_ref::(), + Some(Error::AuthorizationSnapshotStale { .. }) + ) +} diff --git a/crates/validation/src/lib.rs b/crates/validation/src/lib.rs index 1f9c0c31743..ea0b90f0f0c 100644 --- a/crates/validation/src/lib.rs +++ b/crates/validation/src/lib.rs @@ -17,7 +17,7 @@ mod schema; mod storage_mapping; mod test_step; -pub use errors::Error; +pub use errors::{Error, is_authz_snapshot_stale}; pub use noop::NoOpConnectors; /// Portion of the binding namespace reserved for runtime-internal bindings