diff --git a/crates/agent/src/integration_tests/harness.rs b/crates/agent/src/integration_tests/harness.rs index c3f90ed5482..4f143c367f0 100644 --- a/crates/agent/src/integration_tests/harness.rs +++ b/crates/agent/src/integration_tests/harness.rs @@ -589,10 +589,24 @@ impl TestHarness { // // Production refreshes the authorization Snapshot through // `PgSnapshotSource`'s timer-gated polling loop. The harness backs the - // same watch with a manual writer (`set_snapshot`) instead: refreshes are - // explicit, nothing refreshes on a timer, and `MIN_REFRESH_INTERVAL` - // never gates a test. Grant-mutating helpers refresh the watch after - // writing, so authorization observes what a test just set up. + // same watch with a manual writer (`set_snapshot`) instead: + // + // - Refreshes are explicit. Nothing refreshes on a timer, and + // `MIN_REFRESH_INTERVAL` never gates a test. Grant-mutating helpers + // either refresh the watch (`add_role_grant`) or deliberately leave it + // holding the pre-grant world (`add_role_grant_unobserved`). + // + // - Observed *state* and the authoritative *timestamp* are controlled + // separately. A refresh always fetches current Postgres state, but + // stamps it with a caller-chosen `taken`. Staleness compares `taken` + // against an operation's freshness anchor (`Snapshot::taken_after`, + // allowing `TEMPORAL_SKEW`), and tests compress wall-clock time into + // milliseconds — a `taken = now()` Snapshot still reads as stale for a + // row written moments earlier. `refresh_snapshot_authoritative` / + // `refresh_snapshot_stale` push `taken` clear of the skew in either + // direction. + // + // Individual helpers below document only how they differ. // Current Postgres state, stamped `taken = now()`. async fn fetch_snapshot(pool: &sqlx::PgPool) -> control_plane_api::Snapshot { @@ -618,7 +632,8 @@ impl TestHarness { self.refresh_snapshot_at(tokens::now()).await } - /// Refreshes with an exact `taken`. + /// Refreshes with an exact `taken`. Prefer `refresh_snapshot_authoritative` + /// / `refresh_snapshot_stale` unless a test needs a precise instant. pub async fn refresh_snapshot_at(&self, taken: tokens::DateTime) { let snapshot = Self::fetch_snapshot_at(&self.pool, taken).await; (self.set_snapshot)(snapshot); @@ -632,6 +647,14 @@ impl TestHarness { .await } + /// The inverse: current grant state stamped in the past, so any denial it + /// produces reads as provisional and retries. Models production's window + /// where a write has landed in Postgres but the Snapshot predates it. + pub async fn refresh_snapshot_stale(&self) { + self.refresh_snapshot_at(tokens::now() - Self::snapshot_settle()) + .await + } + // Margin pushing `taken` clear of `TEMPORAL_SKEW` in either direction. // Any multiple > 1 works; 4 leaves obvious headroom. fn snapshot_settle() -> chrono::TimeDelta { @@ -660,6 +683,22 @@ impl TestHarness { } pub async fn add_role_grant(&mut self, subject: &str, object: &str, capability: Capability) { + self.add_role_grant_unobserved(subject, object, capability) + .await; + // Re-sync the authorization Snapshot with the new grant. + self.refresh_snapshot().await; + } + + /// Writes a role grant to Postgres *without* re-syncing the authorization + /// Snapshot, modelling production's window between a grant landing in the + /// database and the next Snapshot refresh observing it. Authorization run + /// during that window sees the pre-grant world. + pub async fn add_role_grant_unobserved( + &mut self, + subject: &str, + object: &str, + capability: Capability, + ) { sqlx::query!( r#" insert into role_grants (subject_role, object_role, capability) @@ -672,11 +711,22 @@ impl TestHarness { .execute(&self.pool) .await .unwrap(); + } + + pub async fn add_user_grant(&mut self, user_id: Uuid, role: &str, capability: Capability) { + self.add_user_grant_unobserved(user_id, role, capability) + .await; // Re-sync the authorization Snapshot with the new grant. self.refresh_snapshot().await; } - pub async fn add_user_grant(&mut self, user_id: Uuid, role: &str, capability: Capability) { + /// The `add_role_grant_unobserved` counterpart for user grants. + pub async fn add_user_grant_unobserved( + &mut self, + user_id: Uuid, + role: &str, + capability: Capability, + ) { let mut txn = self.pool.begin().await.unwrap(); control_plane_api::grants::upsert_user_grant( user_id, @@ -688,8 +738,28 @@ impl TestHarness { .await .unwrap(); txn.commit().await.unwrap(); - // Re-sync the authorization Snapshot with the new grant. - self.refresh_snapshot().await; + } + + /// Rewrites `catalog_name`'s `last_pub_id` so the spec reads as published + /// long before any event in the current test. Compressed test time means + /// everything is otherwise "just published", which sidesteps the common + /// production shape of an old spec whose *authorization* changes now. The + /// id sits a few days past the Estuary epoch — old, but non-zero, because + /// a zero id means "never published". + pub async fn age_live_spec(&self, catalog_name: &str) { + let updated = sqlx::query( + "update live_specs set last_pub_id = '00:08:00:00:00:00:00:00'::flowid + where catalog_name = $1", + ) + .bind(catalog_name) + .execute(&self.pool) + .await + .unwrap(); + assert_eq!( + 1, + updated.rows_affected(), + "expected to age exactly one live spec named {catalog_name}" + ); } pub async fn assert_specs_touched_since(&mut self, prev_specs: &tables::LiveCatalog) { @@ -1389,16 +1459,16 @@ impl TestHarness { .await } - /// Runs a publication by inserting into the `publications` table and - /// waiting for the publications handler to process it. Returns - /// a `ScenarioResult` (a hold over from the old publications tests, which - /// were ported over) describing the results of the publication. - async fn async_publication( + /// Inserts a queued `publications` row (creating the draft if one wasn't + /// supplied) and returns its id, *without* running it. `async_publication` + /// runs the task to completion; tests that need to control what the + /// authorization Snapshot looks like between polls drive it themselves. + pub async fn queue_publication( &mut self, user_id: Uuid, detail: impl Into, draft: Either, - ) -> ScenarioResult { + ) -> Id { let detail = detail.into(); let draft_id = match draft { Either::L(catalog) => self.create_draft(user_id, detail.clone(), catalog).await, @@ -1413,30 +1483,75 @@ impl TestHarness { &mut txn, user_id, draft_id, - detail.clone(), + detail, "ops/dp/public/test".to_string(), ) .await .expect("failed to create publication"); txn.commit().await.expect("failed to commit transaction"); + pub_id + } - // The publication's pinned Snapshot must be authoritative for specs - // committed earlier in this test, or an authorization denial reads as - // provisional. Compressed test time never advances past the skew on - // its own, so model production's elapsed wait explicitly. - self.refresh_snapshot_authoritative().await; - + /// Runs exactly one poll of the publications task `pub_id` and returns the + /// resulting `ScenarioResult`. A result whose status is still `Queued` means + /// the executor rescheduled rather than resolving — today that happens only + /// for a stale authorization Snapshot. + pub async fn poll_publication_once(&mut self, pub_id: Id) -> ScenarioResult { let task_id = self .run_automation_task(automations::task_types::PUBLICATIONS) .await .expect("expected a publication task to have run"); - assert_eq!( - task_id, pub_id, - "automations task id should match the publication that was just created" - ); + assert_eq!(task_id, pub_id, "an unexpected publication task ran"); + self.get_publication_result(pub_id.into()).await + } - let pub_result = self.get_publication_result(pub_id.into()).await; - assert_ne!(publications::StatusType::Queued, pub_result.status.r#type); + /// Runs a publication by inserting into the `publications` table and + /// waiting for the publications handler to process it. Returns a + /// `ScenarioResult` (a hold over from the old publications tests, which + /// were ported over) describing the results of the publication. + async fn async_publication( + &mut self, + user_id: Uuid, + detail: impl Into, + draft: Either, + ) -> ScenarioResult { + let detail = detail.into(); + let pub_id = self.queue_publication(user_id, detail, draft).await; + + // A stale-Snapshot publication reschedules (Action::Sleep) rather + // than resolving. Mimic production's re-poll-after-refresh loop, + // bounded so a genuine failure to converge still surfaces. + let mut attempts = 0; + let pub_result = loop { + let task_id = self + .run_automation_task(automations::task_types::PUBLICATIONS) + .await + .expect("expected a publication task to have run"); + assert_eq!( + task_id, pub_id, + "automations task id should match the publication that was just created" + ); + + let pub_result = self.get_publication_result(pub_id.into()).await; + if pub_result.status.r#type != publications::StatusType::Queued { + break pub_result; + } + + attempts += 1; + assert!( + attempts < 5, + "publication kept rescheduling on a stale authorization snapshot" + ); + // Compressed test time never advances past the skew on its own, + // so model production's elapsed wait explicitly (see the Snapshot + // testing model above `fetch_snapshot`). + self.refresh_snapshot_authoritative().await; + self.set_min_task_wake_at(pub_id).await; + }; + assert!( + attempts == 0 || !pub_result.status.is_success(), + "an authorized publication resolved only after {attempts} deferral(s)" + ); pub_result } @@ -2330,7 +2445,7 @@ impl ControlPlane for TestControlPlane { } } -enum Either { +pub enum Either { L(L), R(R), } diff --git a/crates/agent/src/integration_tests/user_publications.rs b/crates/agent/src/integration_tests/user_publications.rs index e31394f4707..b9d4e6852ba 100644 --- a/crates/agent/src/integration_tests/user_publications.rs +++ b/crates/agent/src/integration_tests/user_publications.rs @@ -1,9 +1,10 @@ use super::harness::{ - TestHarness, draft_catalog, get_collection_generation_id, mock_inferred_schema, set_of, + Either, TestHarness, draft_catalog, get_collection_generation_id, mock_inferred_schema, set_of, }; use crate::{ ControlPlane, controllers::ControllerState, integration_tests::harness::InjectBuildError, }; +use control_plane_api::publications; use models::{Capability, CatalogType, Id, status::AlertType}; #[tokio::test] @@ -428,6 +429,506 @@ async fn successful_user_publication_clears_background_publication_failed_alert( harness.assert_alert_resolved(fired_alert.alert.id).await; } +/// A draft that materializes `cats/noms`, which `dogs` may only publish once it +/// holds both a user grant and a role grant to `cats/`. +fn dogs_materialize_cats_draft() -> tables::DraftCatalog { + draft_catalog(serde_json::json!({ + "materializations": { + "dogs/materialize": { + "endpoint": { + "connector": { + "image": "materialize/test:test", + "config": {} + } + }, + "bindings": [ + { + "resource": { "table": "dog_noms" }, + "source": "cats/noms" + } + ] + } + } + })) +} + +/// Publishes `cats/noms` and returns the `dogs` user id. Shared setup for the +/// stale-authorization publication tests below. +async fn setup_cross_tenant_publication(harness: &mut TestHarness) -> uuid::Uuid { + let cats_user = harness.setup_tenant("cats").await; + + // The capture isn't incidental: a draft holding only a collection with no + // writer builds to zero specs and is reported as an empty draft. + let result = harness + .user_publication( + cats_user, + "publish cats/noms", + draft_catalog(serde_json::json!({ + "collections": { + "cats/noms": { + "schema": { + "type": "object", + "properties": { "id": { "type": "string" } } + }, + "key": ["/id"] + } + }, + "captures": { + "cats/capture": { + "endpoint": { + "connector": { + "image": "source/test:test", + "config": {} + } + }, + "bindings": [ + { + "resource": { "id": "noms" }, + "target": "cats/noms" + } + ] + } + } + })), + ) + .await; + assert!( + result.status.is_success(), + "setup publication failed: {:?} {:?}", + result.status, + result.errors + ); + + harness.setup_tenant("dogs").await +} + +/// A publication must reschedule when its selected data plane is denied by a +/// Snapshot which predates the queued publication. This is the concrete race: +/// the grant is restored in Postgres before the publication is queued, but the +/// in-memory Snapshot still reflects the brief revocation. +#[tokio::test] +async fn test_publication_reschedules_on_stale_data_plane_authz() { + let mut harness = TestHarness::init("test_publication_stale_data_plane_authz").await; + let cats_user = harness.setup_tenant("cats").await; + + let deleted = sqlx::query( + "delete from role_grants + where subject_role = 'cats/' and object_role = 'ops/dp/public/'", + ) + .execute(&harness.pool) + .await + .expect("failed to remove the tenant's public-plane grant"); + assert_eq!(1, deleted.rows_affected()); + + // Snapshot A observes the revocation. Restore the grant without refreshing, + // then queue the publication so A is not authoritative for its denial. + harness.refresh_snapshot().await; + harness + .add_role_grant_unobserved("cats/", "ops/dp/public/", Capability::Read) + .await; + let pub_id = harness + .queue_publication( + cats_user, + "public-plane grant awaiting Snapshot refresh", + Either::L(draft_catalog(serde_json::json!({ + "collections": { + "cats/noms": { + "schema": { + "type": "object", + "properties": { "id": { "type": "string" } } + }, + "key": ["/id"] + } + }, + "captures": { + "cats/capture": { + "endpoint": { + "connector": { "image": "source/test:test", "config": {} } + }, + "bindings": [ + { "resource": { "id": "noms" }, "target": "cats/noms" } + ] + } + } + }))), + ) + .await; + + let first = harness.poll_publication_once(pub_id).await; + assert_eq!( + publications::StatusType::Queued, + first.status.r#type, + "publication should reschedule while the restored plane grant is unobserved, got: {:?}", + first.errors + ); + + harness.refresh_snapshot_authoritative().await; + harness.set_min_task_wake_at(pub_id).await; + + let second = harness.poll_publication_once(pub_id).await; + assert!( + second.status.is_success(), + "publication should succeed once the plane grant is observed, got: {:?}", + second.errors + ); +} + +/// After a stale-Snapshot denial, the executor persists the instant an +/// authoritative Snapshot must postdate (in `internal.tasks`, so whichever +/// agent instance dequeues the next poll applies the same criterion) and +/// defers re-polls without loading or building the draft. Once the local +/// Snapshot postdates that instant, the retry proceeds and succeeds. +#[tokio::test] +async fn test_publication_defers_polls_until_authoritative_snapshot() { + let mut harness = TestHarness::init("test_publication_defers_polls").await; + let dogs_user = setup_cross_tenant_publication(&mut harness).await; + + harness.refresh_snapshot_stale().await; + harness + .add_user_grant_unobserved(dogs_user, "cats/", Capability::Read) + .await; + harness + .add_role_grant_unobserved("dogs/", "cats/", Capability::Read) + .await; + + let pub_id = harness + .queue_publication( + dogs_user, + "deferred until authoritative", + Either::L(dogs_materialize_cats_draft()), + ) + .await; + + let first = harness.poll_publication_once(pub_id).await; + assert_eq!( + publications::StatusType::Queued, + first.status.r#type, + "publication should reschedule while the grants are unobserved, got: {:?}", + first.errors + ); + + let state: serde_json::Value = harness.get_task_state(pub_id).await; + assert!( + state + .get("awaiting_snapshot_after") + .is_some_and(|v| v.is_string()), + "the executor should record the instant a Snapshot must postdate, got: {state}" + ); + + // Because the anchor is persisted, the re-poll may be dequeued by a + // *different* agent instance whose own local Snapshot is stale — one whose + // revoke token the original attempt never cancelled. Model that handoff by + // replacing the watch with another stale Snapshot bearing a fresh token. + harness.refresh_snapshot_stale().await; + let handoff_revoke = harness + .snapshot_watch + .token() + .result() + .unwrap() + .revoke + .clone(); + assert!(!handoff_revoke.is_cancelled()); + + // A re-poll under the still-stale Snapshot defers, leaving the row queued. + harness.set_min_task_wake_at(pub_id).await; + let deferred = harness.poll_publication_once(pub_id).await; + assert_eq!( + publications::StatusType::Queued, + deferred.status.r#type, + "a re-poll under a still-stale Snapshot should defer, got: {:?}", + deferred.errors + ); + + // The deferring poll must request a refresh of the Snapshot it observed: + // no prior cancellation covers this instance's Snapshot, and without one + // the task would idle until the watch's ordinary refresh interval. + assert!( + handoff_revoke.is_cancelled(), + "a deferring poll should cancel the stale Snapshot it observed" + ); + + harness.refresh_snapshot_authoritative().await; + harness.set_min_task_wake_at(pub_id).await; + let resolved = harness.poll_publication_once(pub_id).await; + assert!( + resolved.status.is_success(), + "publication should succeed once the Snapshot postdates the anchor, got: {:?}", + resolved.errors + ); +} + +/// The variant of the late-grant race that the test above cannot catch: the +/// referenced spec is *old*. A Snapshot taken after the spec's publication but +/// before the new grants is inconclusive for a publication queued after those +/// grants — staleness is a property of the publication's queued time, not of +/// the referenced spec's age. The publication must remain queued under that +/// Snapshot and succeed once a refresh observes the grants. +#[tokio::test] +async fn test_old_spec_publication_succeeds_after_late_grant() { + let mut harness = + TestHarness::init("test_old_spec_publication_succeeds_after_late_grant").await; + let dogs_user = setup_cross_tenant_publication(&mut harness).await; + + // `cats/noms` was published long before any of the events below. + harness.age_live_spec("cats/noms").await; + + // Snapshot A: taken after the (old) spec but before the grants and the + // publication, so it holds the pre-grant world. + harness.refresh_snapshot_stale().await; + harness + .add_user_grant_unobserved(dogs_user, "cats/", Capability::Read) + .await; + harness + .add_role_grant_unobserved("dogs/", "cats/", Capability::Read) + .await; + + let pub_id = harness + .queue_publication( + dogs_user, + "late grant, old spec", + Either::L(dogs_materialize_cats_draft()), + ) + .await; + + let first = harness.poll_publication_once(pub_id).await; + assert_eq!( + publications::StatusType::Queued, + first.status.r#type, + "a publication evaluated against a Snapshot older than its queued time \ + must reschedule regardless of the referenced spec's age, got: {:?}", + first.errors + ); + + harness.refresh_snapshot_authoritative().await; + harness.set_min_task_wake_at(pub_id).await; + + let second = harness.poll_publication_once(pub_id).await; + assert!( + second.status.is_success(), + "publication should succeed once the grants are observed, got: {:?}", + second.errors + ); +} + +/// An `Initialize` stage which revokes the grants that authorize the test's +/// publication and pushes a refreshed, authoritative Snapshot into the watch — +/// exactly what a background refresh landing between draft initialization and +/// live-spec resolution does in production. Composed as the final Initialize +/// stage, it runs after `ExpandDraft` and before `build`, squarely on the +/// phase boundary that snapshot pinning exists to protect. +struct RevokeMidPublication<'h> { + harness: &'h TestHarness, + dogs_user: uuid::Uuid, +} + +impl publications::Initialize for RevokeMidPublication<'_> { + async fn initialize( + &self, + db: &sqlx::PgPool, + _user_id: uuid::Uuid, + _draft: &mut tables::DraftCatalog, + _snapshot: &control_plane_api::Snapshot, + ) -> anyhow::Result<()> { + sqlx::query( + "delete from role_grants where subject_role = 'dogs/' and object_role = 'cats/'", + ) + .execute(db) + .await?; + sqlx::query("delete from user_grants where user_id = $1 and object_role = 'cats/'") + .bind(self.dogs_user) + .execute(db) + .await?; + self.harness.refresh_snapshot_authoritative().await; + Ok(()) + } +} + +/// One publication must evaluate authorization against exactly one Snapshot: +/// `try_publish` resolves the watch once and threads that Snapshot through +/// both draft initialization and live-spec resolution. A refresh landing +/// between those phases must not swap the view mid-flight. +/// +/// `RevokeMidPublication` deletes the authorizing grants and refreshes the +/// watch after expansion. Resolution still authorizes under the pinned +/// pre-revocation Snapshot, so the publication succeeds; if it consulted the +/// watch anew it would see the revoked world and deny. The guard publication +/// then proves the refreshed watch really does deny the same draft, so the +/// first result is attributable to pinning alone. +#[tokio::test] +async fn test_publication_uses_one_snapshot_across_phases() { + let mut harness = TestHarness::init("test_publication_one_snapshot_across_phases").await; + let dogs_user = setup_cross_tenant_publication(&mut harness).await; + + // Snapshot A: grants written and observed, stamped authoritative. + harness + .add_user_grant(dogs_user, "cats/", Capability::Read) + .await; + harness + .add_role_grant("dogs/", "cats/", Capability::Read) + .await; + harness.refresh_snapshot_authoritative().await; + + // Pin the pre-revocation Snapshot which the whole publication evaluates + // against; the mid-publication refresh below must not displace it. + let refresh = harness.snapshot_watch.token(); + let publication = publications::DraftPublication { + user_id: dogs_user, + logs_token: uuid::Uuid::new_v4(), + dry_run: true, + detail: Some("one snapshot across phases".to_string()), + draft: dogs_materialize_cats_draft(), + started_at: Some(tokens::now()), + snapshot: refresh + .result() + .expect("authorization snapshot is not ready"), + verify_user_authz: true, + default_data_plane_name: Some("ops/dp/public/test".to_string()), + initialize: ( + publications::ExpandDraft { + filter_user_authz: true, + }, + RevokeMidPublication { + harness: &harness, + dogs_user, + }, + ), + finalize: publications::PruneUnboundCollections, + retry: publications::DoNotRetry, + with_commit: publications::NoopWithCommit, + }; + let result = harness + .publisher + .publish(publication) + .await + .expect("publish should not error"); + assert!( + result.status.is_success(), + "the pinned pre-revocation Snapshot should authorize both phases, got: {:?} draft: {:?} live: {:?} built: {:?}", + result.status, + result.draft.errors, + result.live.errors, + result.built.errors, + ); + + // Guard: the same draft judged against the refreshed watch is denied — + // the revocation above is real, and the success was due to pinning. The + // extra refresh stamps the Snapshot authoritative for this publication's + // `started_at`, making the denial terminal rather than a stale retry. + let started_at = tokens::now(); + harness.refresh_snapshot_authoritative().await; + let guard_refresh = harness.snapshot_watch.token(); + let guard = publications::DraftPublication { + user_id: dogs_user, + logs_token: uuid::Uuid::new_v4(), + dry_run: true, + detail: Some("post-revocation guard".to_string()), + draft: dogs_materialize_cats_draft(), + started_at: Some(started_at), + snapshot: guard_refresh + .result() + .expect("authorization snapshot is not ready"), + verify_user_authz: true, + default_data_plane_name: Some("ops/dp/public/test".to_string()), + initialize: publications::ExpandDraft { + filter_user_authz: true, + }, + finalize: publications::PruneUnboundCollections, + retry: publications::DoNotRetry, + with_commit: publications::NoopWithCommit, + }; + let denied = harness + .publisher + .publish(guard) + .await + .expect("guard publish should not error"); + assert!( + !denied.status.is_success(), + "the revoked, authoritative Snapshot must deny the same draft, got: {:?}", + denied.status, + ); +} + +/// The guard on the test above: a genuinely unauthorized publication must not be +/// hidden by the reschedule path. It reschedules only while the Snapshot is +/// inconclusive, then fails with the same authorization errors as before. +/// This also pins the anchor's other boundary: changes committed after the +/// queued publication carry no observation guarantee, so an authoritative +/// denial is terminal regardless of what commits later. +#[tokio::test] +async fn test_publication_stale_then_authoritative_denial() { + let mut harness = TestHarness::init("test_publication_stale_denial").await; + let dogs_user = setup_cross_tenant_publication(&mut harness).await; + + // No grants are ever added — only the Snapshot's age changes. + harness.refresh_snapshot_stale().await; + let pub_id = harness + .queue_publication( + dogs_user, + "never authorized", + Either::L(dogs_materialize_cats_draft()), + ) + .await; + + let first = harness.poll_publication_once(pub_id).await; + assert_eq!( + publications::StatusType::Queued, + first.status.r#type, + "an inconclusive denial should reschedule, got: {:?}", + first.errors + ); + + harness.refresh_snapshot_authoritative().await; + harness.set_min_task_wake_at(pub_id).await; + + let second = harness.poll_publication_once(pub_id).await; + assert!(!second.status.is_success()); + insta::assert_debug_snapshot!(second.errors, @r#" + [ + ( + "flow://unauthorized/cats/noms", + "User is not authorized to read this catalog name", + ), + ( + "flow://materialization/dogs/materialize", + "Specification 'dogs/materialize' is not read-authorized to 'cats/noms'.\nAvailable grants are: [\n {\n \"subject_role\": \"dogs/\",\n \"object_role\": \"dogs/\",\n \"capability\": \"write\",\n \"bundles\": []\n },\n {\n \"subject_role\": \"dogs/\",\n \"object_role\": \"ops/dp/public/\",\n \"capability\": \"read\",\n \"bundles\": []\n }\n]", + ), + ] + "#); +} + +/// Rescheduling alone isn't enough: the raising site must also cancel the +/// Snapshot's `revoke` token, which is what asks the background watch to refresh +/// ahead of its normal interval. Without it a stale publication would sleep +/// against an unchanged Snapshot until the next scheduled refresh. +#[tokio::test] +async fn test_publication_requests_snapshot_refresh() { + let mut harness = TestHarness::init("test_publication_requests_refresh").await; + let dogs_user = setup_cross_tenant_publication(&mut harness).await; + + harness.refresh_snapshot_stale().await; + let token = harness.snapshot_watch.token(); + let snapshot = token.result().expect("snapshot should be ready"); + assert!( + !snapshot.revoke.is_cancelled(), + "a freshly-published Snapshot should not already be revoked" + ); + + let pub_id = harness + .queue_publication( + dogs_user, + "requests refresh", + Either::L(dogs_materialize_cats_draft()), + ) + .await; + let result = harness.poll_publication_once(pub_id).await; + assert_eq!(publications::StatusType::Queued, result.status.r#type); + + assert!( + snapshot.revoke.is_cancelled(), + "a stale-snapshot publication must request an early Snapshot refresh" + ); +} + async fn assert_publication_included( publication_id: Id, catalog_names: &[&str], diff --git a/crates/agent/src/publications.rs b/crates/agent/src/publications.rs index 4810786101b..a77b7da94f1 100644 --- a/crates/agent/src/publications.rs +++ b/crates/agent/src/publications.rs @@ -17,7 +17,9 @@ pub struct PublicationsExecutor { pub publisher: Publisher, pub pg_pool: sqlx::PgPool, /// Authorization Snapshot watch. Each poll pins one Snapshot from this - /// watch, which serves every authorization decision of the publication. + /// watch: first to cheaply defer while it remains stale for a queued + /// publication (see `Snapshot::taken_after`), and then to serve + /// every authorization decision of the publication itself. pub snapshot_watch: std::sync::Arc>, /// When true, newly-created captures are published onto runtime v2; see [`RuntimeV2Rollout`]. pub runtime_v2_new_captures: bool, @@ -29,16 +31,14 @@ pub struct PublicationsExecutor { /// Poll state persisted to `internal.tasks` between polls, and therefore /// shared with whichever agent instance dequeues the next poll. -/// -/// This deploy doesn't yet read or write this state: it's carried so that -/// state persisted by the upcoming stale-snapshot deferral changes remains -/// decodable by this version during a deploy or rollback. -#[allow(dead_code)] #[derive(Debug, Default, serde::Serialize, serde::Deserialize)] pub struct PublicationState { /// The instant a Snapshot must postdate (per `Snapshot::taken_after`) /// before this publication is retried: the queued time its prior attempt - /// anchored authorization staleness on. + /// anchored authorization staleness on. While set, polls defer — without + /// loading or building the draft — until the local Snapshot satisfies it. + /// Optional so that reschedules for other, future reasons aren't bound to + /// this check. #[serde(default)] pub awaiting_snapshot_after: Option, } @@ -60,23 +60,29 @@ impl automations::Executor for PublicationsExecutor { pool: &'s sqlx::PgPool, task_id: models::Id, _parent_id: Option, - _state: &'s mut Self::State, + state: &'s mut Self::State, inbox: &'s mut std::collections::VecDeque<(models::Id, Option)>, ) -> anyhow::Result { tracing::debug!(?inbox, "starting publication task"); let row = fetch_publication(task_id, pool).await?; - self.handle_task(row).await?; + let action = self.handle_task(row, state).await?; // Always clear inbox, or else we'll get re-polled. inbox.clear(); - // Publication tasks are always done at the end. We don't retry because there is likely - // a user waiting for the result, who could easily retry the operation themselves. - Ok(automations::Action::Done) + // A publication is normally `Done` at the end — we don't retry failures + // because a user is likely waiting and can retry themselves. The one + // exception is a stale authorization snapshot, where `handle_task` + // returns a `Sleep` and we re-poll until a fresher snapshot decides it. + Ok(action) } } impl PublicationsExecutor { - async fn handle_task(&self, row: Row) -> anyhow::Result<()> { + async fn handle_task( + &self, + row: Row, + state: &mut Option, + ) -> anyhow::Result { let id = row.id; // First ensure that the publication status is queued. Otherwise, @@ -85,7 +91,7 @@ impl PublicationsExecutor { Ok(status) if status.r#type == StatusType::Queued => { /* continue to publish */ } Ok(other) => { tracing::warn!(?other, "skipping publication which is no longer queued"); - return Ok(()); + return Ok(automations::Action::Done); } Err(error) => { // Weird edge case, but we don't update the status so that we @@ -93,17 +99,33 @@ impl PublicationsExecutor { // the task completed so that the user can update the status // back to queued if they want. tracing::error!(?error, "failed to parse publication job status"); - return Ok(()); + return Ok(automations::Action::Done); } } - // Pin one Snapshot for this poll: every authorization decision of the - // publication observes the same view. + // Pin one Snapshot for this poll: the deferral decision and every + // authorization decision of the publication observe the same view. let snapshot = self.snapshot_watch.token(); let snapshot = snapshot.result().unwrap(); + // A prior attempt was denied under a Snapshot that was not + // authoritative for this publication. Defer — without loading or + // building the draft — until this instance's Snapshot is, at which + // point the retry is guaranteed to classify deterministically. + if let Some(anchor) = state.as_ref().and_then(|s| s.awaiting_snapshot_after) { + if !snapshot.taken_after(anchor) { + snapshot.revoke.cancel(); + return Ok(automations::Action::Sleep( + Snapshot::STALE_RETRY_WAKE + .to_std() + .expect("wake interval is positive"), + )); + } + } + let dry_run = row.dry_run; let draft_id = row.draft_id; + let queued_at = row.updated_at; let time_queued = chrono::Utc::now().signed_duration_since(row.updated_at); @@ -123,6 +145,24 @@ impl PublicationsExecutor { }; (result.status, errors, final_id) } + Err(error) if validation::is_authz_snapshot_stale(&error) => { + // An authorization denial was evaluated against a Snapshot that + // isn't authoritative for this publication. `Publisher::publish` + // already requested an early refresh; record the instant an + // authoritative Snapshot must postdate and reschedule, so that + // re-polls defer cheaply until one lands rather than reporting + // a failure. + tracing::info!( + pub_id = %id, %time_queued, + "publication authorization snapshot is stale; rescheduling" + ); + state.get_or_insert_default().awaiting_snapshot_after = Some(queued_at); + return Ok(automations::Action::Sleep( + Snapshot::STALE_RETRY_WAKE + .to_std() + .expect("wake interval is positive"), + )); + } Err(error) => { tracing::warn!(?error, pub_id = %id, "build finished with error"); let errors = vec![draft_error::Error { @@ -155,7 +195,7 @@ impl PublicationsExecutor { if status.is_success() && !dry_run { delete_draft(draft_id, &self.pg_pool).await?; } - Ok(()) + Ok(automations::Action::Done) } #[tracing::instrument(skip_all, fields( @@ -201,12 +241,11 @@ impl PublicationsExecutor { dry_run: row.dry_run, detail: row.detail.clone(), draft, - // `None` anchors authorization staleness to each spec's own - // `last_pub_id`, preserving the pre-Snapshot semantics where a - // denial is always evaluated against current-enough state. A - // follow-up anchors this to the queued publication row and defers - // on staleness instead. - started_at: None, + // `updated_at` is the instant this row entered `queued`, and is + // stable across our reschedules. Authorization denials evaluated + // against a snapshot older than it are treated as not-yet-observed + // and retried rather than reported. + started_at: Some(row.updated_at), snapshot, verify_user_authz: true, default_data_plane_name: row.data_plane_name.clone().filter(|s| !s.is_empty()), diff --git a/crates/control-plane-api/src/fixtures/attenuated_grants.sql b/crates/control-plane-api/src/fixtures/attenuated_grants.sql new file mode 100644 index 00000000000..432a5035251 --- /dev/null +++ b/crates/control-plane-api/src/fixtures/attenuated_grants.sql @@ -0,0 +1,36 @@ +-- Grant paths whose *raw* legacy capability reaches a data-plane prefix with +-- `admin`, but whose *effective* (attenuated) authority differs. Used to pin +-- that data-plane visibility is decided by effective authority — the exact +-- regression where a filter consults the raw legacy capability of the edge +-- which reached the prefix. +-- +-- Both users traverse the same 2-hop shape through `sharedCo/`: +-- +-- user_grant(user, 'sharedCo/', C, B) -> role_grant('sharedCo/' -> 'ops/dp/public/', 'admin') +-- +-- The role_grant node's effective bits are `admin`'s bits intersected with +-- what the parent may delegate: +-- +-- * erin: 'none' + '{editor}' delegates CatalogRead|JournalRead|SpecEdit|Delegate, +-- which misses ViewDataPlanePrivateNetworking — so she fails the Viewer +-- requirement of `Capability::Read` despite the raw `admin` edge. +-- * frank: 'read' + '{delegate}' delegates the full Viewer set, so the same +-- path *does* authorize him: the positive control proving the traversal +-- works and only attenuation blocks erin. +-- +-- Their own tenants deliberately hold no role_grants: a second path to +-- `ops/dp/public/` would union its bits into the plane node and mask the +-- attenuation under test. +insert into auth.users (id, email) values + ('55555555-5555-5555-5555-555555555555', 'erin@example.com'), + ('66666666-6666-6666-6666-666666666666', 'frank@example.com') +; +insert into public.user_grants (user_id, object_role, capability, bundles) values + ('55555555-5555-5555-5555-555555555555', 'erinCo/', 'admin', '{}'), + ('55555555-5555-5555-5555-555555555555', 'sharedCo/', 'none', '{editor}'), + ('66666666-6666-6666-6666-666666666666', 'frankCo/', 'admin', '{}'), + ('66666666-6666-6666-6666-666666666666', 'sharedCo/', 'read', '{delegate}') +; +insert into public.role_grants (subject_role, object_role, capability, bundles) values + ('sharedCo/', 'ops/dp/public/', 'admin', '{}') +; diff --git a/crates/control-plane-api/src/publications/specs.rs b/crates/control-plane-api/src/publications/specs.rs index b0c27faae0f..8252cfecf2f 100644 --- a/crates/control-plane-api/src/publications/specs.rs +++ b/crates/control-plane-api/src/publications/specs.rs @@ -1228,6 +1228,9 @@ mod resolve_tests { // From `fixtures/authz_specs.sql`. const CAROL: uuid::Uuid = uuid::uuid!("33333333-3333-3333-3333-333333333333"); const DAN: uuid::Uuid = uuid::uuid!("44444444-4444-4444-4444-444444444444"); + // From `fixtures/attenuated_grants.sql`. + const ERIN: uuid::Uuid = uuid::uuid!("55555555-5555-5555-5555-555555555555"); + const FRANK: uuid::Uuid = uuid::uuid!("66666666-6666-6666-6666-666666666666"); const COLLECTION: &str = "carolCo/data/foo"; const CAPTURE: &str = "carolCo/in/capture-foo"; const MATERIALIZATION: &str = "carolCo/out/materialize-bar"; @@ -1573,4 +1576,319 @@ mod resolve_tests { .expect_err("spec authorization is checked regardless of verify_user_authz"); assert_stale_for(err, CAPTURE); } + + /// Named data planes use the publication's durable `started` timestamp as + /// their freshness anchor. Grants win regardless of Snapshot age, while a + /// denial is retryable only until the Snapshot becomes authoritative. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_data_plane_name_authorization_freshness(pool: sqlx::PgPool) { + let dan_draft = draft_of(serde_json::json!({ + "collections": { + "danCo/thing": { + "schema": { "type": "object", "properties": { "id": { "type": "string" } } }, + "key": ["/id"] + } + } + })); + let started = published_at(&pool).await; + let stale_snapshot = stale(&pool).await; + + // Dan admins `danCo/` but was granted nothing on `ops/dp/public/`. + // Because this Snapshot is not authoritative for `started`, its denial + // is provisional and names the plane which triggered it. + let err = resolve_live_specs( + DAN, + &dan_draft, + &pool, + true, + Some(PLANE), + &stale_snapshot, + Some(started), + ) + .await + .expect_err("a stale data-plane denial should be retryable"); + assert_stale_for(err, PLANE); + + // Once the Snapshot is authoritative, the same denial preserves the + // existing non-disclosure behavior and silently omits the plane. + let live = resolve_live_specs( + DAN, + &dan_draft, + &pool, + true, + Some(PLANE), + &authoritative(&pool).await, + Some(started), + ) + .await + .expect("an authoritative data-plane denial is terminal omission"); + assert!( + live.errors.is_empty(), + "unexpected errors: {:?}", + error_pairs(&live) + ); + assert!( + live.data_planes.is_empty(), + "an authoritatively denied data-plane should be omitted" + ); + + // Carol holds `carolCo/ -> ops/dp/public/ read`, so the same plane is + // included even though the Snapshot is too old to make denials final. + let carol_draft = draft_of(serde_json::json!({ + "collections": { + "carolCo/thing": { + "schema": { "type": "object", "properties": { "id": { "type": "string" } } }, + "key": ["/id"] + } + } + })); + let live = resolve_live_specs( + CAROL, + &carol_draft, + &pool, + true, + Some(PLANE), + &stale_snapshot, + Some(started), + ) + .await + .expect("an observed grant wins regardless of Snapshot age"); + assert_eq!(1, live.data_planes.len()); + + // System publications skip user authorization for named planes as well + // as catalog specs. Spec-to-spec RoleGrant checks remain mandatory. + let live = resolve_live_specs( + DAN, + &dan_draft, + &pool, + false, + Some(PLANE), + &stale_snapshot, + Some(started), + ) + .await + .expect("verify_user_authz=false should include the named plane"); + assert_eq!(1, live.data_planes.len()); + + // Callers without a durable operation timestamp must not invent one: + // their denials preserve the prior terminal omission behavior. + let live = resolve_live_specs( + DAN, + &dan_draft, + &pool, + true, + Some(PLANE), + &stale_snapshot, + None, + ) + .await + .expect("a plane denial without a freshness anchor is terminal"); + assert!(live.data_planes.is_empty()); + } + + /// Storage-mapping plane names follow the same freshness policy even when + /// there is no explicit/default plane name in the publication. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_storage_mapping_data_plane_authorization_freshness(pool: sqlx::PgPool) { + let mapping = crate::TextJson(models::StorageDef { + data_planes: vec![PLANE.to_string()], + stores: vec![models::Store::example()], + }); + sqlx::query("insert into storage_mappings (catalog_prefix, spec) values ($1, $2)") + .bind("danCo/") + .bind(&mapping) + .execute(&pool) + .await + .expect("failed to insert test storage mapping"); + + let draft = draft_of(serde_json::json!({ + "collections": { + "danCo/thing": { + "schema": { "type": "object", "properties": { "id": { "type": "string" } } }, + "key": ["/id"] + } + } + })); + let err = resolve_live_specs( + DAN, + &draft, + &pool, + true, + None, + &stale(&pool).await, + Some(published_at(&pool).await), + ) + .await + .expect_err("a stale storage-mapping plane denial should be retryable"); + assert_stale_for(err, PLANE); + } + + /// The data-plane name filter must be decided by *effective* (attenuated) + /// authority, not the raw legacy capability of the edge which reached the + /// prefix. Erin and frank traverse the identical 2-hop path through + /// `sharedCo/` to a raw-`admin` grant on `ops/dp/public/`; only frank's + /// root grant delegates the Viewer bits, so only frank sees the plane. A + /// regression to raw-capability filtering makes the plane visible to erin. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures( + path = "../fixtures", + scripts("data_planes", "authz_specs", "attenuated_grants") + ) + )] + async fn test_attenuated_data_plane_grant_is_not_visible(pool: sqlx::PgPool) { + let snapshot = authoritative(&pool).await; + + // The premise that makes this attenuation rather than simple absence: + // erin's raw reachable capability at the plane is Admin, and yet her + // effective authority does not satisfy Read. + assert_eq!( + Some(models::Capability::Admin), + tables::UserGrant::get_user_capability( + &snapshot.role_grants, + &snapshot.user_grants, + ERIN, + PLANE, + ), + ); + assert!(!tables::UserGrant::is_authorized( + &snapshot.role_grants, + &snapshot.user_grants, + ERIN, + PLANE, + models::Capability::Read, + )); + + let erin_draft = draft_of(serde_json::json!({ + "collections": { + "erinCo/thing": { + "schema": { "type": "object", "properties": { "id": { "type": "string" } } }, + "key": ["/id"] + } + } + })); + let live = resolve_live_specs(ERIN, &erin_draft, &pool, true, Some(PLANE), &snapshot, None) + .await + .expect("an unauthorized data-plane name is not an error"); + assert!( + live.errors.is_empty(), + "unexpected errors: {:?}", + error_pairs(&live) + ); + assert!( + live.data_planes.is_empty(), + "a plane reached with raw admin but attenuated effective authority must not be visible" + ); + + let frank_draft = draft_of(serde_json::json!({ + "collections": { + "frankCo/thing": { + "schema": { "type": "object", "properties": { "id": { "type": "string" } } }, + "key": ["/id"] + } + } + })); + let live = resolve_live_specs( + FRANK, + &frank_draft, + &pool, + true, + Some(PLANE), + &snapshot, + None, + ) + .await + .expect("frank is authorized to the plane"); + assert!( + live.errors.is_empty(), + "unexpected errors: {:?}", + error_pairs(&live) + ); + assert_eq!(1, live.data_planes.len()); + } + + /// Scenario 2: Request-relative staleness anchoring allows retries when the + /// snapshot predates the request, even if the spec is old. This is the + /// "old-spec late-grant" case: a grant might exist but arrive in the system + /// after the snapshot was taken but before the request was queued. + /// + /// This test shows that with request-relative anchoring, a denial is: + /// - Retried if snapshot.taken_before(request_start) (grant might exist but not in snapshot) + /// - Terminal if snapshot.taken_after(request_start) (grant would be in snapshot if it existed) + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("authz_specs")) + )] + async fn test_old_spec_stale_snapshot_relative_to_request(pool: sqlx::PgPool) { + let draft = capture_draft(&[CAPTURE]); + + // A snapshot taken well before "now" is stale relative to any request + // queued around "now". This should trigger a retry even though the spec + // itself is old. + let stale_snapshot = stale(&pool).await; + let now = published_at(&pool).await + chrono::TimeDelta::seconds(3600); + + let err = resolve_live_specs( + uuid::Uuid::nil(), + &draft, + &pool, + false, + None, + &stale_snapshot, + // Request was queued at `now`, well after the stale snapshot. + Some(now), + ) + .await + .expect_err("spec authorization required even without user authz"); + + // The denial should be stale relative to the request time, so retryable. + assert_stale_for(err, CAPTURE); + } + + /// When the snapshot is authoritative relative to the request start time, + /// an authorization denial is terminal (not retried), even for an old spec. + /// This shows the request-relative anchor is properly applied. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("authz_specs")) + )] + async fn test_old_spec_authoritative_snapshot_relative_to_request(pool: sqlx::PgPool) { + let draft = capture_draft(&[CAPTURE]); + + // An authoritative snapshot is taken at published_at + TEMPORAL_SKEW * 4. + let authoritative_snapshot = authoritative(&pool).await; + let pub_time = published_at(&pool).await; + // Request queued just before the snapshot. Since snapshot is at pub_time + 1s, + // queuing at pub_time means snapshot.taken_after(now) is true (snapshot is authoritative). + let now = pub_time; + + let live = resolve_live_specs( + uuid::Uuid::nil(), + &draft, + &pool, + false, + None, + &authoritative_snapshot, + // Request was queued at `now`, before the authoritative snapshot. + Some(now), + ) + .await + .expect("resolve should not error with authoritative snapshot"); + + // The denial should be terminal (not stale) because the snapshot is + // authoritative relative to the request start time. The capture spec + // lacks authorization, so we get a hard error, not a retry. + assert!(!live.errors.is_empty(), "expected authorization denial"); + let error = &live.errors.iter().next().unwrap().error; + assert!( + !validation::is_authz_snapshot_stale(error), + "error should not be stale-snapshot error" + ); + } }