diff --git a/.sqlx/query-f133c223a0df27245c413ef1ef8d94773a3c9ce12c6c34161874455e206e35dc.json b/.sqlx/query-f133c223a0df27245c413ef1ef8d94773a3c9ce12c6c34161874455e206e35dc.json deleted file mode 100644 index 858e7943170..00000000000 --- a/.sqlx/query-f133c223a0df27245c413ef1ef8d94773a3c9ce12c6c34161874455e206e35dc.json +++ /dev/null @@ -1,89 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n SELECT\n d.id AS \"control_id: Id\",\n d.data_plane_name,\n d.closed,\n d.hmac_keys,\n d.encrypted_hmac_keys AS \"encrypted_hmac_keys: models::RawValue\",\n d.data_plane_fqdn,\n d.broker_address,\n d.reactor_address,\n d.dekaf_address,\n d.dekaf_registry_address,\n d.ops_logs_name AS \"ops_logs_name: models::Collection\",\n d.ops_stats_name AS \"ops_stats_name: models::Collection\"\n FROM data_planes d\n WHERE data_plane_name = $1\n AND EXISTS (\n SELECT 1 FROM internal.user_roles($2, 'read') r\n WHERE starts_with($1, r.role_prefix)\n )\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "control_id: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 1, - "name": "data_plane_name", - "type_info": "Text" - }, - { - "ordinal": 2, - "name": "closed", - "type_info": "Bool" - }, - { - "ordinal": 3, - "name": "hmac_keys", - "type_info": "TextArray" - }, - { - "ordinal": 4, - "name": "encrypted_hmac_keys: models::RawValue", - "type_info": "Json" - }, - { - "ordinal": 5, - "name": "data_plane_fqdn", - "type_info": "Text" - }, - { - "ordinal": 6, - "name": "broker_address", - "type_info": "Text" - }, - { - "ordinal": 7, - "name": "reactor_address", - "type_info": "Text" - }, - { - "ordinal": 8, - "name": "dekaf_address", - "type_info": "Text" - }, - { - "ordinal": 9, - "name": "dekaf_registry_address", - "type_info": "Text" - }, - { - "ordinal": 10, - "name": "ops_logs_name: models::Collection", - "type_info": "Text" - }, - { - "ordinal": 11, - "name": "ops_stats_name: models::Collection", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Text", - "Uuid" - ] - }, - "nullable": [ - false, - false, - false, - false, - false, - false, - false, - false, - true, - true, - false, - false - ] - }, - "hash": "f133c223a0df27245c413ef1ef8d94773a3c9ce12c6c34161874455e206e35dc" -} diff --git a/crates/agent/src/discovers.rs b/crates/agent/src/discovers.rs index 23d7e8e423b..6f5e76877a2 100644 --- a/crates/agent/src/discovers.rs +++ b/crates/agent/src/discovers.rs @@ -1,6 +1,6 @@ use anyhow::Context; use control_plane_api::{ - Snapshot, connector_tags, + Authorization, Snapshot, connector_tags, discovers::{Discover, DiscoverHandler, Row, fetch_discover}, draft, live_specs, proxy_connectors::DiscoverConnectors, @@ -19,6 +19,7 @@ pub enum JobStatus { PullFailed, DiscoverFailed, MergeFailed, + NotAuthorized, Success { #[serde(default, skip_serializing_if = "Option::is_none")] publication_id: Option, @@ -43,25 +44,36 @@ type ProcessResult = Result, } -pub struct DiscoverOutcome { - id: Id, - draft_id: Id, - result: ProcessResult, - status: JobStatus, +/// Outcome of evaluating a discover in `DiscoverExecutor::process`. +enum Processed { + Resolved(JobStatus, ProcessResult), + /// Control-plane state is not authoritative because the snapshot predates + /// the discover row. A refresh has been requested; retry after a short delay. + RetryStale, +} + +pub enum DiscoverOutcome { + Resolved { + id: Id, + draft_id: Id, + result: ProcessResult, + status: JobStatus, + }, + /// The control-plane snapshot was stale; the discover is left queued and + /// re-polled once refreshed state should be authoritative. + RetryStale, } impl automations::Outcome for DiscoverOutcome { @@ -69,12 +81,22 @@ impl automations::Outcome for DiscoverOutcome { self, txn: &'s mut sqlx::PgConnection, ) -> anyhow::Result { - let DiscoverOutcome { + let DiscoverOutcome::Resolved { id, draft_id, result, status, - } = self; + } = self + else { + // Leave the discover unresolved and re-poll once a refresh could + // have landed. If the Snapshot is still not authoritative by + // then, the poll defers again (see `Snapshot::taken_after`). + return Ok(automations::Action::Sleep( + Snapshot::STALE_RETRY_WAKE + .to_std() + .expect("wake interval is positive"), + )); + }; control_plane_api::draft::delete_errors(draft_id, txn) .await @@ -94,14 +116,12 @@ impl automations::Outcome for DiscoverOutcome { } } -fn precheck_failed(status: JobStatus) -> (JobStatus, ProcessResult) { - (status, Err(Vec::new())) +fn precheck_failed(status: JobStatus) -> Processed { + Processed::Resolved(status, Err(Vec::new())) } pub struct DiscoverExecutor { pub handler: DiscoverHandler, - /// Authorization Snapshot watch. Each poll pins one Snapshot from this - /// watch, which serves every authorization decision of the discover. pub snapshot_watch: std::sync::Arc>, } @@ -122,29 +142,53 @@ impl automations::Executor for DiscoverExecutor { 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, %task_id, "executing discover task"); let row = fetch_discover(task_id, pool).await?; let draft_id = row.draft_id; assert_eq!(row.id, task_id); + let queued_at = row.updated_at; let time_queued = chrono::Utc::now().signed_duration_since(row.updated_at); - // Pin one Snapshot for this poll: every authorization decision of the - // discover observes the same view. + // Pin one Snapshot for this poll: the deferral decision and every + // authorization decision of the discover observe the same view. let snapshot = self.snapshot_watch.token(); let snapshot = snapshot.result().unwrap(); - let (status, result) = self.process(row, pool, snapshot).await?; - tracing::info!(id=%task_id, %time_queued, ?status, "finished"); + // A prior attempt could not classify under a stale Snapshot. Defer — + // without pre-flight checks or connector work — until this instance's + // Snapshot is authoritative for the recorded instant. + if let Some(anchor) = state.as_ref().and_then(|s| s.awaiting_snapshot_after) { + if !snapshot.taken_after(anchor) { + snapshot.revoke.cancel(); + inbox.clear(); + return Ok(DiscoverOutcome::RetryStale); + } + } + + let processed = self.process(row, pool, &snapshot).await?; inbox.clear(); - Ok(DiscoverOutcome { - id: task_id, - draft_id, - result, - status, - }) + match processed { + Processed::Resolved(status, result) => { + tracing::info!(id=%task_id, %time_queued, ?status, "finished"); + Ok(DiscoverOutcome::Resolved { + id: task_id, + draft_id, + result, + status, + }) + } + Processed::RetryStale => { + tracing::info!( + id=%task_id, %time_queued, + "control-plane snapshot is stale; rescheduling discover after refresh" + ); + state.get_or_insert_default().awaiting_snapshot_after = Some(queued_at); + Ok(DiscoverOutcome::RetryStale) + } + } } } @@ -155,7 +199,7 @@ impl DiscoverExecutor { row: Row, pool: &sqlx::PgPool, snapshot: &Snapshot, - ) -> anyhow::Result<(JobStatus, ProcessResult)> { + ) -> anyhow::Result { tracing::info!( %row.capture_name, %row.connector_tag_id, @@ -180,38 +224,42 @@ impl DiscoverExecutor { } else if !connector_tags::does_connector_exist(&row.image_name, pool).await? { return Ok(precheck_failed(JobStatus::ImageForbidden)); } - let maybe_data_plane = sqlx::query_as!( - tables::DataPlane, - r#" - SELECT - d.id AS "control_id: Id", - d.data_plane_name, - d.closed, - d.hmac_keys, - d.encrypted_hmac_keys AS "encrypted_hmac_keys: models::RawValue", - d.data_plane_fqdn, - d.broker_address, - d.reactor_address, - d.dekaf_address, - d.dekaf_registry_address, - d.ops_logs_name AS "ops_logs_name: models::Collection", - d.ops_stats_name AS "ops_stats_name: models::Collection" - FROM data_planes d - WHERE data_plane_name = $1 - AND EXISTS ( - SELECT 1 FROM internal.user_roles($2, 'read') r - WHERE starts_with($1, r.role_prefix) - ) - "#, - row.data_plane_name, - row.user_id, - ) - .fetch_optional(pool) - .await - .context("fetching data-plane")?; - let Some(data_plane) = maybe_data_plane else { - tracing::warn!(data_plane_name = ?row.data_plane_name, "data-plane not found or user may not be authorized"); + // Legacy `read` is what conveys deploy-level trust in a data plane; + // no narrow capability bit expresses that trust yet. + match snapshot.user_authorization( + row.user_id, + &row.data_plane_name, + models::Capability::Read, + Some(row.updated_at), + ) { + Authorization::Authorized => (), + Authorization::Denied => { + // The snapshot reflects the world after this discover was + // queued, so the denial is authoritative. + tracing::warn!(data_plane_name = ?row.data_plane_name, "user is not authorized to read data plane"); + return Ok(precheck_failed(JobStatus::NotAuthorized)); + } + Authorization::Stale => { + // The snapshot predates this discover's row, so a grant that + // would authorize the read may not be reflected yet. Request an + // early refresh and retry, rather than emitting a spurious + // NotAuthorized/NoDataPlane. + tracing::warn!(data_plane_name = ?row.data_plane_name, "data-plane read denied under a stale snapshot"); + snapshot.revoke.cancel(); + return Ok(Processed::RetryStale); + } + } + let data_plane = snapshot.data_plane_by_catalog_name(&row.data_plane_name); + + let Some(data_plane) = data_plane else { + tracing::warn!(data_plane_name = ?row.data_plane_name, "data-plane not found in control-plane snapshot"); + if !snapshot.taken_after(row.updated_at) { + // A plane registered after this Snapshot may already exist. + // Request an early refresh before making the absence terminal. + snapshot.revoke.cancel(); + return Ok(Processed::RetryStale); + } return Ok(precheck_failed(JobStatus::NoDataPlane)); }; @@ -224,14 +272,10 @@ impl DiscoverExecutor { row.update_only, row.logs_token, image_composed, - data_plane, + data_plane.clone(), pool, &snapshot, - // `None` anchors authorization staleness to each spec's own - // `last_pub_id`, preserving the pre-Snapshot semantics. A - // follow-up anchors this to the queued discover row and defers - // on staleness instead. - None, + Some(row.updated_at), ) .await; @@ -241,7 +285,7 @@ impl DiscoverExecutor { }; match result { - Ok(output) if output.is_success() => Ok(( + Ok(output) if output.is_success() => Ok(Processed::Resolved( JobStatus::Success { publication_id: None, specs_unchanged: false, @@ -255,7 +299,17 @@ impl DiscoverExecutor { .iter() .map(tables::Error::to_draft_error) .collect::>(); - Ok((JobStatus::DiscoverFailed, Err(draft_errs))) + Ok(Processed::Resolved( + JobStatus::DiscoverFailed, + Err(draft_errs), + )) + } + Err(err) if validation::is_authz_snapshot_stale(&err) => { + // An authorization denial was evaluated against a Snapshot that + // isn't authoritative for this discover. Request an early refresh + // and retry, rather than reporting a spurious DiscoverFailed. + snapshot.revoke.cancel(); + Ok(Processed::RetryStale) } Err(err) => { let draft_errors = vec![models::draft_error::Error { @@ -266,7 +320,10 @@ impl DiscoverExecutor { catalog_name: row.capture_name.clone(), detail: format!("{:#}", err), }]; - Ok((JobStatus::DiscoverFailed, Err(draft_errors))) + Ok(Processed::Resolved( + JobStatus::DiscoverFailed, + Err(draft_errors), + )) } } } @@ -307,6 +364,9 @@ async fn prepare_discover<'a>( // running task does. It's empty for a task which doesn't exist yet. // Filter to only specs that the user can view. If they can't edit, then // wait until they try to publish to surface that error. + // Use request-relative staleness: a denial from a Snapshot older than the + // queued discover is provisional, because a grant committed before queuing + // may be missing from it. let name = &[capture_name.to_string()]; let live = live_specs::get_live_specs( user_id, diff --git a/crates/agent/src/integration_tests/harness.rs b/crates/agent/src/integration_tests/harness.rs index 4f143c367f0..7343ec1c819 100644 --- a/crates/agent/src/integration_tests/harness.rs +++ b/crates/agent/src/integration_tests/harness.rs @@ -129,7 +129,7 @@ pub struct UserDiscoverResult { } impl UserDiscoverResult { - async fn load(discover_id: Id, db: &sqlx::PgPool) -> UserDiscoverResult { + pub async fn load(discover_id: Id, db: &sqlx::PgPool) -> UserDiscoverResult { let discover = sqlx::query!( r#"select draft_id as "draft_id: Id", @@ -167,6 +167,24 @@ async fn load_draft_errors(draft_id: Id, db: &sqlx::PgPool) -> Vec<(String, Stri .collect::>() } +/// Owned counterpart of `TestHarness::refresh_snapshot_authoritative`, handed +/// out by `TestHarness::snapshot_refresher` to `'static` test fixtures which +/// cannot borrow the harness. +#[derive(Clone)] +pub struct SnapshotRefresher { + pool: sqlx::PgPool, + set_snapshot: Arc, +} + +impl SnapshotRefresher { + /// See `TestHarness::refresh_snapshot_authoritative`. + pub async fn refresh_authoritative(&self) { + let taken = tokens::now() + TestHarness::snapshot_settle(); + let snapshot = TestHarness::fetch_snapshot_at(&self.pool, taken).await; + (self.set_snapshot)(snapshot); + } +} + /// Facilitates writing integration tests. /// **Note:** integration tests require exclusive access to the database, /// so it's required to serialize test runs (see .config/nextest.toml). @@ -639,6 +657,16 @@ impl TestHarness { (self.set_snapshot)(snapshot); } + /// An owned handle which performs `refresh_snapshot_authoritative`, for + /// injecting a refresh from within `'static` test fixtures — such as a + /// `DiscoverConnectors` impl — which cannot borrow the harness. + pub fn snapshot_refresher(&self) -> SnapshotRefresher { + SnapshotRefresher { + pool: self.pool.clone(), + set_snapshot: self.set_snapshot.clone(), + } + } + /// Refreshes with `taken` pushed far enough forward to be authoritative /// for everything written up to now: any denial it produces is definitive /// rather than retryable. @@ -1405,6 +1433,110 @@ impl TestHarness { UserDiscoverResult::load(disco_id, &self.pool).await } + /// Registers an additional data-plane beyond the `ops/dp/public/test` one + /// that `setup_test_connectors` creates. Tests use this when they need a + /// plane that *exists* but that a given user has no grant to read, which is + /// the only way to distinguish an authorization denial from a missing plane. + pub async fn add_data_plane(&self, data_plane_name: &str) { + sqlx::query!( + r##"insert into data_planes ( + data_plane_name, + data_plane_fqdn, + ops_logs_name, + ops_stats_name, + ops_l1_inferred_name, + ops_l1_stats_name, + ops_l1_events_name, + ops_l2_inferred_transform, + ops_l2_stats_transform, + ops_l2_events_transform, + broker_address, + reactor_address, + hmac_keys, + enable_l2 + ) values ( + $1, + $2, + 'ops/logs', + 'ops/stats', + 'ops/L1/inferred', + 'ops/L1/stats', + 'ops/L1/events', + 'from-L1-inferred', + 'from-L1-stats', + 'from-L1-events', + 'broker:address', + 'reactor:address', + '{secret-key}', + false + );"##, + data_plane_name as &str, + format!("{}.dp.estuary-data.com", data_plane_name.replace('/', "-")) as String, + ) + .execute(&self.pool) + .await + .expect("failed to insert data-plane"); + } + + /// Inserts a queued `discovers` row for a caller-chosen `data_plane_name` + /// and returns its id, without running it. Unlike `user_discover`, the + /// data-plane is a parameter so tests can drive authorization outcomes + /// (which are evaluated against `data_plane_name`). + pub async fn queue_discover( + &self, + image_name: &str, + image_tag: &str, + capture_name: &str, + draft_id: Id, + data_plane_name: &str, + ) -> Id { + let connector_tag = sqlx::query!( + r##"select ct.id as "id: Id" + from connectors c + join connector_tags ct on c.id = ct.connector_id + where c.image_name = $1 and ct.image_tag = $2;"##, + image_name, + image_tag + ) + .fetch_one(&self.pool) + .await + .expect("querying for connector_tags id"); + + let config_json = TextJson(models::RawValue::from_str("{}").unwrap()); + let disco = sqlx::query!( + r##"insert into discovers ( + capture_name, + connector_tag_id, + draft_id, + endpoint_config, + update_only, + data_plane_name + ) values ($1, $2, $3, $4, false, $5) + returning id as "id: Id";"##, + capture_name as &str, + connector_tag.id as Id, + draft_id as Id, + config_json as TextJson, + data_plane_name as &str, + ) + .fetch_one(&self.pool) + .await + .unwrap(); + disco.id + } + + pub async fn discover_job_status(&self, discover_id: Id) -> crate::discovers::JobStatus { + let row = sqlx::query!( + r#"select job_status as "job_status: TextJson" + from discovers where id = $1;"#, + discover_id as Id, + ) + .fetch_one(&self.pool) + .await + .expect("failed to query discover"); + row.job_status.0 + } + pub async fn fail_shard(&mut self, shard: &ShardRef) { let fields = serde_json::from_value(serde_json::json!({ "eventType": "shardFailure", diff --git a/crates/agent/src/integration_tests/user_discovers.rs b/crates/agent/src/integration_tests/user_discovers.rs index a6c8b9f9ccb..8a2559da004 100644 --- a/crates/agent/src/integration_tests/user_discovers.rs +++ b/crates/agent/src/integration_tests/user_discovers.rs @@ -1,9 +1,20 @@ use super::{spec_fixture, wrap_connector_schema}; use crate::{ ControlPlane, - integration_tests::harness::{TestHarness, UserDiscoverResult, draft_catalog, set_of}, + discovers::JobStatus, + integration_tests::harness::{ + SnapshotRefresher, TestHarness, UserDiscoverResult, connectors::MockDiscoverConnectors, + draft_catalog, set_of, + }, }; +use control_plane_api::{ + discovers::{Discover, DiscoverHandler}, + proxy_connectors::DiscoverConnectors, +}; +use models::Id; +use proto_flow::capture; use proto_flow::capture::response::{Discovered, discovered::Binding}; +use uuid::Uuid; #[tokio::test] async fn test_user_discovers() { @@ -333,6 +344,942 @@ async fn test_user_discovers() { } } +// A data-plane that exists but that no tenant in these tests is granted to +// read. Registering it (rather than using a bogus name) is what separates an +// authorization denial from `NoDataPlane`. +const FOREIGN_DATA_PLANE: &str = "dogs/dp/private/test"; + +/// Queues a discover for `capture_name` against `FOREIGN_DATA_PLANE`, which the +/// caller's tenant has no grant to read. +async fn queue_foreign_dp_discover(harness: &mut TestHarness, user_id: Uuid, name: &str) -> Id { + let draft_id = harness + .create_draft(user_id, name, Default::default()) + .await; + harness + .queue_discover("source/test", ":test", name, draft_id, FOREIGN_DATA_PLANE) + .await +} + +/// A discover whose data-plane authorization is denied by a Snapshot that +/// predates the discover row must be retried, not failed: a grant that would +/// authorize it may simply not be reflected in that Snapshot yet. The row stays +/// queued and the task reschedules. +#[tokio::test] +async fn test_discover_reschedules_on_stale_data_plane_authz() { + let mut harness = TestHarness::init("test_discover_reschedules_on_stale_data_plane").await; + let user_id = harness.setup_tenant("cats").await; + harness.add_data_plane(FOREIGN_DATA_PLANE).await; + + let disco_id = queue_foreign_dp_discover(&mut harness, user_id, "cats/capture-stale").await; + harness.refresh_snapshot_stale().await; + + let ran = harness + .run_automation_task(automations::task_types::DISCOVERS) + .await; + assert_eq!(Some(disco_id), ran, "expected the stale discover to run"); + assert!( + matches!( + harness.discover_job_status(disco_id).await, + JobStatus::Queued + ), + "stale-snapshot discover should stay queued (rescheduled), got: {:?}", + harness.discover_job_status(disco_id).await, + ); +} + +/// The converse: once the Snapshot is authoritative for the discover row, the +/// same denial is definitive and resolves terminally rather than looping. +/// This is the anchor's other boundary: changes committed after the queued +/// discover carry no observation guarantee, so an authoritative denial is +/// terminal regardless of what commits later. +#[tokio::test] +async fn test_discover_unauthorized_data_plane_is_terminal() { + let mut harness = TestHarness::init("test_discover_unauthorized_data_plane").await; + let user_id = harness.setup_tenant("cats").await; + harness.add_data_plane(FOREIGN_DATA_PLANE).await; + + let disco_id = queue_foreign_dp_discover(&mut harness, user_id, "cats/capture-authz").await; + harness.refresh_snapshot_authoritative().await; + + let ran = harness + .run_automation_task(automations::task_types::DISCOVERS) + .await; + assert_eq!( + Some(disco_id), + ran, + "expected the authoritative discover to run" + ); + assert!( + matches!( + harness.discover_job_status(disco_id).await, + JobStatus::NotAuthorized + ), + "authoritative unauthorized discover should resolve NotAuthorized, got: {:?}", + harness.discover_job_status(disco_id).await, + ); +} + +/// After a stale-Snapshot denial, the discover 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 pre-flight checks or connector work. Once the +/// local Snapshot postdates that instant, the retry proceeds and succeeds. +#[tokio::test] +async fn test_discover_defers_polls_until_authoritative_snapshot() { + let mut harness = TestHarness::init("test_discover_defers_polls").await; + let user_id = harness.setup_tenant("cats").await; + harness.add_data_plane(FOREIGN_DATA_PLANE).await; + + harness.refresh_snapshot_stale().await; + harness + .add_role_grant_unobserved("cats/", "dogs/dp/private/", models::Capability::Read) + .await; + + let capture_name = "cats/capture-deferred"; + let disco_id = queue_foreign_dp_discover(&mut harness, user_id, capture_name).await; + harness.discover_handler.connectors.mock_discover( + capture_name, + Ok((spec_fixture(), single_binding_response("acorns"))), + ); + + let ran = harness + .run_automation_task(automations::task_types::DISCOVERS) + .await; + assert_eq!(Some(disco_id), ran); + assert!( + matches!( + harness.discover_job_status(disco_id).await, + JobStatus::Queued + ), + "discover should reschedule while the grant is unobserved, got: {:?}", + harness.discover_job_status(disco_id).await, + ); + + let state: serde_json::Value = harness.get_task_state(disco_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(disco_id).await; + let ran = harness + .run_automation_task(automations::task_types::DISCOVERS) + .await; + assert_eq!(Some(disco_id), ran); + assert!( + matches!( + harness.discover_job_status(disco_id).await, + JobStatus::Queued + ), + "a re-poll under a still-stale Snapshot should defer, got: {:?}", + harness.discover_job_status(disco_id).await, + ); + + // 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(disco_id).await; + let ran = harness + .run_automation_task(automations::task_types::DISCOVERS) + .await; + assert_eq!(Some(disco_id), ran); + let status = harness.discover_job_status(disco_id).await; + assert!( + matches!(status, JobStatus::Success { .. }), + "discover should succeed once the Snapshot postdates the anchor, got: {status:?}", + ); +} + +/// The second, independent stale path through `DiscoverExecutor::process`: the +/// data-plane check passes, but `prepare_discover`'s `get_live_specs` denies the +/// discover's own capture against a Snapshot older than that capture. The error +/// arrives as `AuthorizationSnapshotStale` from `control-plane-api` and must be +/// mapped to a reschedule rather than a `DiscoverFailed` status. +#[tokio::test] +async fn test_discover_reschedules_on_stale_live_spec_authz() { + let mut harness = TestHarness::init("test_discover_stale_live_spec_authz").await; + let cats_user = harness.setup_tenant("cats").await; + let dogs_user = harness.setup_tenant("dogs").await; + + // Publish a capture owned by `cats`. + let capture_name = "cats/capture-owned"; + let pub_result = harness + .user_publication( + cats_user, + "publish cats capture", + draft_catalog(serde_json::json!({ + "captures": { capture_name: minimal_capture() }, + })), + ) + .await; + assert!( + pub_result.status.is_success(), + "setup publication failed: {:?}", + pub_result.errors + ); + + // `dogs` may read the shared data-plane (every tenant is granted + // `ops/dp/public/`), so the data-plane precheck passes and we reach the + // live-spec authorization inside `prepare_discover`. `dogs` has no grant to + // `cats/`, and the Snapshot predates the capture's publication. + let draft_id = harness + .create_draft(dogs_user, "cross-tenant discover", Default::default()) + .await; + let disco_id = harness + .queue_discover( + "source/test", + ":test", + capture_name, + draft_id, + "ops/dp/public/test", + ) + .await; + harness.refresh_snapshot_stale().await; + + let ran = harness + .run_automation_task(automations::task_types::DISCOVERS) + .await; + assert_eq!(Some(disco_id), ran); + assert!( + matches!( + harness.discover_job_status(disco_id).await, + JobStatus::Queued + ), + "stale live-spec authorization should reschedule, got: {:?}", + harness.discover_job_status(disco_id).await, + ); + + // With an authoritative Snapshot the denial stops being retryable and the + // discover reaches a terminal status instead of looping forever. + harness.refresh_snapshot_authoritative().await; + harness.discover_handler.connectors.mock_discover( + capture_name, + Ok((spec_fixture(), single_binding_response("kibble"))), + ); + harness.set_min_task_wake_at(disco_id).await; + + let ran = harness + .run_automation_task(automations::task_types::DISCOVERS) + .await; + assert_eq!(Some(disco_id), ran); + // Once authoritative, the denial stops being retryable. `get_live_specs` + // falls back to its pre-existing behavior of silently omitting the + // unreadable spec, so the discover proceeds as if the capture were new + // rather than looping. Documenting the concrete status (not merely + // "terminal") keeps that silent drop visible. + let status = harness.discover_job_status(disco_id).await; + assert!( + matches!(status, JobStatus::Success { .. }), + "discover should reach a terminal status once the Snapshot is authoritative, got: {status:?}", + ); +} + +/// A collection which a capture binding targets across tenant lines. Its owner +/// (`dogs`) publishes it with a user-set projection; the discovering user +/// (`cats`) needs a read grant to it for the merge to see it. +const SHARED_COLLECTION: &str = "dogs/shared/data"; + +/// Publishes `SHARED_COLLECTION` under `dogs`, ages it so its publication time +/// cannot mask request-relative staleness, and drafts a `cats` capture whose +/// binding targets it. Returns the draft id for `queue_discover`. +async fn setup_shared_collection_discover( + harness: &mut TestHarness, + cats_user: Uuid, + dogs_user: Uuid, + capture_name: &str, +) -> Id { + // The writer 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 pub_result = harness + .user_publication( + dogs_user, + "publish shared collection", + draft_catalog(serde_json::json!({ + "collections": { + SHARED_COLLECTION: { + // Wrapped as a connector-managed schema: the merge + // refuses to update collections whose schemas + // auto-discover doesn't manage. + "schema": wrap_connector_schema(serde_json::json!({ + "type": "object", + "properties": { "id": { "type": "string" } }, + "required": ["id"] + })), + "key": ["/id"], + "projections": { "id_projection": "/id" } + } + }, + "captures": { + "dogs/shared/writer": { + "endpoint": { + "connector": { "image": "source/test:test", "config": {} } + }, + "bindings": [ + { "resource": { "id": "data" }, "target": SHARED_COLLECTION } + ] + } + }, + })), + ) + .await; + assert!( + pub_result.status.is_success(), + "setup publication failed: {:?} {:?}", + pub_result.status, + pub_result.errors + ); + harness.age_live_spec(SHARED_COLLECTION).await; + + harness + .create_draft( + cats_user, + "shared collection discover", + draft_catalog(serde_json::json!({ + "captures": { + capture_name: { + "endpoint": { + "connector": { "image": "source/test:test", "config": {} } + }, + "bindings": [ + { "resource": { "id": "data" }, "target": SHARED_COLLECTION } + ], + } + }, + })), + ) + .await +} + +/// Asserts the drafted `SHARED_COLLECTION` was merged *from the live +/// collection*: it expects the live, nonzero publication id and keeps the +/// owner's projection. A collection drafted from scratch — what a silently +/// dropped authorization produces — has `expect_pub_id: zero` and no +/// projections, so each assertion is discriminating on its own. +fn assert_live_collection_preserved(draft: &tables::DraftCatalog) { + let drafted = draft + .collections + .get_by_key(&models::Collection::new(SHARED_COLLECTION)) + .expect("the target collection should be drafted"); + assert!( + drafted.expect_pub_id.is_some_and(|id| !id.is_zero()), + "the drafted collection should expect the live publication id, got: {:?}", + drafted.expect_pub_id, + ); + let model = drafted.model.as_ref().expect("drafted collection model"); + assert!( + model + .projections + .contains_key(&models::Field::new("id_projection")), + "the live collection's projection should be preserved, got: {:?}", + model.projections, + ); +} + +/// The merge phase fetches the capture's target collections with the user's +/// `CatalogRead` capability, and its staleness anchor must be the discover request — +/// not the target collection's own age. This is the late-observation race for +/// a *collection*: the grant to `dogs/shared/` commits before the discover is +/// queued, but the Snapshot predates both. Judged spec-relatively the (aged) +/// collection makes the denial authoritative, and it is silently dropped: the +/// discover "succeeds", re-drafting the collection from scratch with a zeroed +/// publication id. Judged request-relatively the denial is provisional, the +/// discover reschedules, and a refreshed Snapshot — guaranteed to include the +/// pre-queue grant — preserves the live collection. +#[tokio::test] +async fn test_discover_reschedules_on_stale_collection_authz() { + let mut harness = TestHarness::init("test_discover_stale_collection_authz").await; + let cats_user = harness.setup_tenant("cats").await; + let dogs_user = harness.setup_tenant("dogs").await; + + let capture_name = "cats/capture-shared"; + let draft_id = + setup_shared_collection_discover(&mut harness, cats_user, dogs_user, capture_name).await; + + // The Snapshot holds the pre-grant world; the grant and the discover row + // both come after it, in that order. + harness.refresh_snapshot_stale().await; + harness + .add_role_grant_unobserved("cats/", "dogs/shared/", models::Capability::Read) + .await; + + let disco_id = harness + .queue_discover( + "source/test", + ":test", + capture_name, + draft_id, + "ops/dp/public/test", + ) + .await; + harness.discover_handler.connectors.mock_discover( + capture_name, + Ok((spec_fixture(), single_binding_response("data"))), + ); + + let ran = harness + .run_automation_task(automations::task_types::DISCOVERS) + .await; + assert_eq!(Some(disco_id), ran); + assert!( + matches!( + harness.discover_job_status(disco_id).await, + JobStatus::Queued + ), + "a stale denial of the binding's target collection should reschedule, got: {:?}", + harness.discover_job_status(disco_id).await, + ); + + // The Snapshot catches up, observing the grant: the discover completes and + // the merge is based on the live collection. + harness.refresh_snapshot_authoritative().await; + harness.set_min_task_wake_at(disco_id).await; + + let ran = harness + .run_automation_task(automations::task_types::DISCOVERS) + .await; + assert_eq!(Some(disco_id), ran); + + let result = UserDiscoverResult::load(disco_id, &harness.pool).await; + assert!( + result.job_status.is_success(), + "discover should succeed once the grant is observed, got: {:?} with errors: {:?}", + result.job_status, + result.errors, + ); + assert_live_collection_preserved(&result.draft); +} + +/// The complement of `test_discover_reschedules_on_stale_live_spec_authz`, +/// and the reported scenario end to end: an *existing* capture with +/// non-default bindings and settings, whose reader is granted access just +/// before queuing a re-discover — after the Snapshot was taken. The capture +/// is aged, so a spec-relative staleness anchor would call the stale denial +/// authoritative, silently filter the live capture, and "succeed" with a +/// starter baseline — `expect_pub_id: 0`, no bindings, default settings. +/// Anchored to the discover request the denial is provisional: the first +/// poll reschedules, and a refreshed Snapshot — guaranteed to include the +/// pre-queue grant — preserves the live capture: its nonzero publication id, +/// its binding, and its non-default interval. Each assertion is +/// discriminating on its own, because the starter baseline has none of them. +#[tokio::test] +async fn test_discover_preserves_live_capture_after_late_grant() { + let mut harness = TestHarness::init("test_discover_capture_late_grant").await; + let cats_user = harness.setup_tenant("cats").await; + let dogs_user = harness.setup_tenant("dogs").await; + + // Publish a capture owned by `cats` with distinctive, non-default + // properties: a bound collection and a 42-minute interval. The collection + // schema is connector-managed so a re-discover may merge into it. The + // capture and collection live under *different* sub-prefixes: `dogs` gets + // an observed grant to the collection below, so that only the capture's + // own authorization rides on the late grant — otherwise the (correctly + // request-anchored) collection check would also reschedule and mask a + // regression of the capture anchor. + let capture_name = "cats/in/capture-owned"; + let pub_result = harness + .user_publication( + cats_user, + "publish cats capture", + draft_catalog(serde_json::json!({ + "collections": { + "cats/data/noms": { + "schema": wrap_connector_schema(serde_json::json!({ + "type": "object", + "properties": { "id": { "type": "string" } }, + "required": ["id"] + })), + "key": ["/id"] + } + }, + "captures": { + capture_name: { + "endpoint": { + "connector": { "image": "source/test:test", "config": {} } + }, + "bindings": [ + { "resource": { "id": "noms" }, "target": "cats/data/noms" } + ], + "interval": "42m" + } + }, + })), + ) + .await; + assert!( + pub_result.status.is_success(), + "setup publication failed: {:?} {:?}", + pub_result.status, + pub_result.errors + ); + // Age the capture: a spec-relative anchor would now judge any recent + // Snapshot authoritative for it, which is exactly the regression this + // test discriminates against. + harness.age_live_spec(capture_name).await; + // The collection grant is observed from the start. + harness + .add_role_grant("dogs/", "cats/data/", models::Capability::Read) + .await; + + let draft_id = harness + .create_draft(dogs_user, "late-grant re-discover", Default::default()) + .await; + + // The Snapshot observes the collection grant but not the capture's, + // which commits after it and just before the discover is queued. + harness.refresh_snapshot_stale().await; + harness + .add_role_grant_unobserved("dogs/", "cats/in/", models::Capability::Read) + .await; + + let disco_id = harness + .queue_discover( + "source/test", + ":test", + capture_name, + draft_id, + "ops/dp/public/test", + ) + .await; + harness.discover_handler.connectors.mock_discover( + capture_name, + Ok((spec_fixture(), single_binding_response("noms"))), + ); + + let ran = harness + .run_automation_task(automations::task_types::DISCOVERS) + .await; + assert_eq!(Some(disco_id), ran); + assert!( + matches!( + harness.discover_job_status(disco_id).await, + JobStatus::Queued + ), + "a stale denial of the existing capture should reschedule, got: {:?}", + harness.discover_job_status(disco_id).await, + ); + + // The Snapshot observes the grant; the discover completes against the + // live capture rather than a starter baseline. + harness.refresh_snapshot_authoritative().await; + harness.set_min_task_wake_at(disco_id).await; + + let ran = harness + .run_automation_task(automations::task_types::DISCOVERS) + .await; + assert_eq!(Some(disco_id), ran); + + let result = UserDiscoverResult::load(disco_id, &harness.pool).await; + assert!( + result.job_status.is_success(), + "discover should succeed once the grant is observed, got: {:?} with errors: {:?}", + result.job_status, + result.errors, + ); + + let drafted = result + .draft + .captures + .get_by_key(&models::Capture::new(capture_name)) + .expect("the capture should be drafted"); + assert!( + drafted.expect_pub_id.is_some_and(|id| !id.is_zero()), + "the drafted capture should expect the live publication id, got: {:?}", + drafted.expect_pub_id, + ); + let model = drafted.model.as_ref().expect("drafted capture model"); + assert_eq!( + vec!["cats/data/noms"], + model + .bindings + .iter() + .map(|b| b.target.as_str()) + .collect::>(), + "the live capture's binding should be preserved", + ); + assert_eq!( + std::time::Duration::from_secs(42 * 60), + model.interval, + "the live capture's non-default interval should be preserved", + ); + assert!( + model.auto_discover.is_none(), + "a preserved live capture must not gain the starter's auto_discover", + ); +} + +/// A `DiscoverConnectors` which revokes the grant authorizing the discover's +/// target collection — and pushes a refreshed, authoritative Snapshot into +/// the watch — while the connector RPC is in flight, before answering with +/// the underlying mock. This is production's shape: connector RPCs run for +/// seconds, and grant changes and Snapshot refreshes land freely within them. +#[derive(Clone)] +struct RevokeMidRpc { + pool: sqlx::PgPool, + refresher: SnapshotRefresher, + inner: MockDiscoverConnectors, +} + +impl DiscoverConnectors for RevokeMidRpc { + async fn discover<'a>( + &'a self, + data_plane: &'a tables::DataPlane, + task: &'a models::Capture, + logs_token: Uuid, + request: capture::Request, + ) -> anyhow::Result<(capture::response::Spec, capture::response::Discovered)> { + sqlx::query( + "delete from role_grants where subject_role = 'cats/' and object_role = 'dogs/shared/'", + ) + .execute(&self.pool) + .await?; + self.refresher.refresh_authoritative().await; + self.inner + .discover(data_plane, task, logs_token, request) + .await + } +} + +/// One discover must evaluate authorization against exactly one Snapshot: the +/// executor pins the watch's Snapshot once, and it rides the `Discover` +/// through the connector RPC into the merge. A grant change landing while the +/// RPC is in flight must not cause the capture baseline (resolved before the +/// RPC) and the collection baselines (resolved after it) to come from +/// different views. +/// +/// `RevokeMidRpc` deletes the authorizing grant and refreshes the watch from +/// inside the RPC. The merge still resolves `SHARED_COLLECTION` under the +/// pinned pre-revocation Snapshot, so the discover succeeds and preserves the +/// live collection. The guard discover then pins the post-revocation Snapshot +/// and shows the same merge silently drops the collection — so the first +/// result is attributable to pinning alone. +#[tokio::test] +async fn test_discover_uses_one_snapshot_across_connector_rpc() { + let mut harness = TestHarness::init("test_discover_one_snapshot_across_rpc").await; + let cats_user = harness.setup_tenant("cats").await; + let dogs_user = harness.setup_tenant("dogs").await; + + let capture_name = "cats/capture-shared"; + let draft_id = + setup_shared_collection_discover(&mut harness, cats_user, dogs_user, capture_name).await; + // Snapshot A: the grant written and observed, stamped authoritative. + harness + .add_role_grant("cats/", "dogs/shared/", models::Capability::Read) + .await; + harness.refresh_snapshot_authoritative().await; + + let mut mock = MockDiscoverConnectors::default(); + mock.mock_discover( + capture_name, + Ok((spec_fixture(), single_binding_response("data"))), + ); + let handler = DiscoverHandler::new(RevokeMidRpc { + pool: harness.pool.clone(), + refresher: harness.snapshot_refresher(), + inner: mock, + }); + + // Pin Snapshot A and assemble the request, as `DiscoverExecutor::process` + // and `prepare_discover` do. + let draft = control_plane_api::draft::load_draft(draft_id, &harness.pool) + .await + .unwrap(); + let snapshot = harness.snapshot_watch.token(); + let snapshot = snapshot.result().unwrap(); + let data_plane = snapshot + .data_plane_by_catalog_name("ops/dp/public/test") + .expect("test data-plane exists") + .clone(); + let output = handler + .discover( + &harness.pool, + Discover { + capture_name: models::Capture::new(capture_name), + data_plane, + logs_token: Uuid::new_v4(), + user_id: cats_user, + filter_user_authz: true, + update_only: false, + reset_on_key_change: false, + draft, + created_at: String::new(), + snapshot, + started_at: Some(tokens::now()), + }, + ) + .await + .expect("discover should not error"); + assert!( + output.is_success(), + "the pinned pre-revocation Snapshot should authorize the merge, got: {:?}", + output.draft.errors, + ); + assert_live_collection_preserved(&output.draft); + + // Guard: the same discover pinning the post-revocation Snapshot cannot see + // the live collection: the merge silently drops it and re-drafts it from + // scratch. The extra refresh stamps the Snapshot authoritative for this + // discover's `started_at`, making the denial terminal rather than stale. + let started_at = tokens::now(); + harness.refresh_snapshot_authoritative().await; + + let mut mock = MockDiscoverConnectors::default(); + mock.mock_discover( + capture_name, + Ok((spec_fixture(), single_binding_response("data"))), + ); + let handler = DiscoverHandler::new(mock); + let draft = control_plane_api::draft::load_draft(draft_id, &harness.pool) + .await + .unwrap(); + let snapshot = harness.snapshot_watch.token(); + let snapshot = snapshot.result().unwrap(); + let data_plane = snapshot + .data_plane_by_catalog_name("ops/dp/public/test") + .expect("test data-plane exists") + .clone(); + let output = handler + .discover( + &harness.pool, + Discover { + capture_name: models::Capture::new(capture_name), + data_plane, + logs_token: Uuid::new_v4(), + user_id: cats_user, + filter_user_authz: true, + update_only: false, + reset_on_key_change: false, + draft, + created_at: String::new(), + snapshot, + started_at: Some(started_at), + }, + ) + .await + .expect("guard discover should not error"); + assert!( + output.is_success(), + "an authoritative denial silently drops the collection, got: {:?}", + output.draft.errors, + ); + let drafted = output + .draft + .collections + .get_by_key(&models::Collection::new(SHARED_COLLECTION)) + .expect("the target collection should be drafted"); + assert_eq!( + Some(models::Id::zero()), + drafted.expect_pub_id, + "under the revoked view the live collection is invisible and re-drafted from scratch", + ); +} + +/// A data-plane registration that lands after the current Snapshot must be +/// observable to a discover queued afterward. Until the Snapshot catches up, +/// the missing plane is provisional: the discover stays queued and requests an +/// early refresh. It proceeds once an authoritative Snapshot includes the +/// registration. +#[tokio::test] +async fn test_discover_succeeds_after_late_data_plane_registration() { + let mut harness = TestHarness::init("test_discover_late_data_plane_registration").await; + let user_id = harness.setup_tenant("cats").await; + let data_plane_name = "ops/dp/public/late-registration"; + let capture_name = "cats/capture-late-data-plane"; + + // Snapshot A includes the tenant's grant to `ops/dp/public/`, but not the + // concrete plane which is registered immediately afterward. + 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" + ); + harness.add_data_plane(data_plane_name).await; + + let draft_id = harness + .create_draft(user_id, "late data-plane discover", Default::default()) + .await; + let disco_id = harness + .queue_discover( + "source/test", + ":test", + capture_name, + draft_id, + data_plane_name, + ) + .await; + harness.discover_handler.connectors.mock_discover( + capture_name, + Ok((spec_fixture(), single_binding_response("acorns"))), + ); + + let ran = harness + .run_automation_task(automations::task_types::DISCOVERS) + .await; + assert_eq!(Some(disco_id), ran); + assert!( + matches!( + harness.discover_job_status(disco_id).await, + JobStatus::Queued + ), + "a plane missing from a stale Snapshot should reschedule, got: {:?}", + harness.discover_job_status(disco_id).await, + ); + assert!( + snapshot.revoke.is_cancelled(), + "a stale missing-plane decision must request an early Snapshot refresh" + ); + + harness.refresh_snapshot_authoritative().await; + harness.set_min_task_wake_at(disco_id).await; + + let ran = harness + .run_automation_task(automations::task_types::DISCOVERS) + .await; + assert_eq!(Some(disco_id), ran); + let status = harness.discover_job_status(disco_id).await; + assert!( + matches!(status, JobStatus::Success { .. }), + "discover should succeed once the plane is observed, got: {status:?}", + ); +} + +/// An authorized plane that remains absent after an authoritative refresh must +/// resolve as `NoDataPlane`. The stale first poll is provisional, while the +/// authoritative second poll is terminal; neither may invoke the connector. +#[tokio::test] +async fn test_discover_missing_data_plane_is_terminal_after_refresh() { + let mut harness = TestHarness::init("test_discover_missing_data_plane").await; + let user_id = harness.setup_tenant("cats").await; + let capture_name = "cats/capture-missing-dp"; + let data_plane_name = "ops/dp/public/does-not-exist"; + + // `setup_tenant` grants `cats/ -> ops/dp/public/ read`, so this name passes + // authorization; Snapshot A and Postgres both lack the concrete plane. + harness.refresh_snapshot_stale().await; + let draft_id = harness + .create_draft(user_id, "missing dp discover", Default::default()) + .await; + let disco_id = harness + .queue_discover( + "source/test", + ":test", + capture_name, + draft_id, + data_plane_name, + ) + .await; + + let ran = harness + .run_automation_task(automations::task_types::DISCOVERS) + .await; + assert_eq!(Some(disco_id), ran); + assert!( + matches!( + harness.discover_job_status(disco_id).await, + JobStatus::Queued + ), + "a plane missing from a stale Snapshot should reschedule, got: {:?}", + harness.discover_job_status(disco_id).await, + ); + assert!( + harness + .discover_handler + .connectors + .last_discover_request(capture_name) + .is_none(), + "the connector must not be invoked while plane existence is unknown" + ); + + harness.refresh_snapshot_authoritative().await; + harness.set_min_task_wake_at(disco_id).await; + + let ran = harness + .run_automation_task(automations::task_types::DISCOVERS) + .await; + assert_eq!(Some(disco_id), ran); + assert!( + matches!( + harness.discover_job_status(disco_id).await, + JobStatus::NoDataPlane + ), + "a plane missing from an authoritative Snapshot should be NoDataPlane, got: {:?}", + harness.discover_job_status(disco_id).await, + ); + assert!( + harness + .discover_handler + .connectors + .last_discover_request(capture_name) + .is_none(), + "the connector must not be invoked for a missing plane" + ); +} + +/// `JobStatus::NotAuthorized` is new, and job statuses round-trip through a JSON +/// column. Pin its serialized form so a rename can't silently orphan rows that +/// were written with the old spelling. +#[test] +fn test_job_status_not_authorized_serde() { + let encoded = serde_json::to_value(&JobStatus::NotAuthorized).unwrap(); + assert_eq!(serde_json::json!({"type": "notAuthorized"}), encoded); + assert!(matches!( + serde_json::from_value::(encoded).unwrap(), + JobStatus::NotAuthorized + )); +} + +/// A discovered response with a single enabled binding, sufficient for a +/// discover to merge and succeed. +fn single_binding_response(name: &str) -> Discovered { + Discovered { + bindings: vec![Binding { + recommended_name: name.to_string(), + document_schema_json: document_schema(1), + resource_config_json: format!(r#"{{"id": "{name}"}}"#).into(), + key: vec!["/id".to_string()], + disable: false, + resource_path: Vec::new(), + is_fallback_key: false, + }], + } +} + +fn minimal_capture() -> serde_json::Value { + serde_json::json!({ + "endpoint": { + "connector": { + "image": "source/test:test", + "config": {} + } + }, + "bindings": [], + }) +} + fn document_schema(version: usize) -> bytes::Bytes { serde_json::to_string(&serde_json::json!({ "type": "object",