diff --git a/.sqlx/query-04b899d07ba206e427e2881f72d7c46c8e32f4d09f87acd20b102c7db14d8fe5.json b/.sqlx/query-04b899d07ba206e427e2881f72d7c46c8e32f4d09f87acd20b102c7db14d8fe5.json new file mode 100644 index 00000000000..2f5f8f62e5f --- /dev/null +++ b/.sqlx/query-04b899d07ba206e427e2881f72d7c46c8e32f4d09f87acd20b102c7db14d8fe5.json @@ -0,0 +1,48 @@ +{ + "db_name": "PostgreSQL", + "query": "\n WITH pinned AS (\n SELECT id, generation\n FROM unnest($2::flowid[], $3::bigint[]) AS p(id, generation)\n ),\n results AS (\n SELECT id, status, error, detail\n FROM unnest($4::flowid[], $5::text[], $6::text[], $7::jsonb[])\n AS r(id, status, error, detail)\n )\n UPDATE internal.data_plane_private_links l SET\n status = r.status,\n details = r.detail,\n error = r.error,\n observed_at = now(),\n updated_at = now()\n FROM internal.data_plane_private_links l2\n JOIN pinned p ON p.id = l2.id AND p.generation = l2.generation\n JOIN results r ON r.id = l2.id\n WHERE l.id = l2.id\n AND l2.data_plane_id = $1\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Macaddr8", + { + "Custom": { + "name": "flowid[]", + "kind": { + "Array": { + "Custom": { + "name": "flowid", + "kind": { + "Domain": "Macaddr8" + } + } + } + } + } + }, + "Int8Array", + { + "Custom": { + "name": "flowid[]", + "kind": { + "Array": { + "Custom": { + "name": "flowid", + "kind": { + "Domain": "Macaddr8" + } + } + } + } + } + }, + "TextArray", + "TextArray", + "JsonbArray" + ] + }, + "nullable": [] + }, + "hash": "04b899d07ba206e427e2881f72d7c46c8e32f4d09f87acd20b102c7db14d8fe5" +} diff --git a/.sqlx/query-05346a208af70f3ae464564dd11b40cc8038bb3358af182b081e94af6322e500.json b/.sqlx/query-05346a208af70f3ae464564dd11b40cc8038bb3358af182b081e94af6322e500.json new file mode 100644 index 00000000000..982f0aff6c4 --- /dev/null +++ b/.sqlx/query-05346a208af70f3ae464564dd11b40cc8038bb3358af182b081e94af6322e500.json @@ -0,0 +1,34 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT\n id AS \"id: models::Id\",\n generation,\n config AS \"config!: sqlx::types::Json\"\n FROM internal.data_plane_private_links\n WHERE data_plane_id = $1\n ORDER BY created_at, id\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id: models::Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 1, + "name": "generation", + "type_info": "Int8" + }, + { + "ordinal": 2, + "name": "config!: sqlx::types::Json", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Macaddr8" + ] + }, + "nullable": [ + false, + false, + false + ] + }, + "hash": "05346a208af70f3ae464564dd11b40cc8038bb3358af182b081e94af6322e500" +} diff --git a/.sqlx/query-31a9225ee20553ae34d112d0b70bfa85e1000cf48342e9c8eeb159de5b9c0e9c.json b/.sqlx/query-31a9225ee20553ae34d112d0b70bfa85e1000cf48342e9c8eeb159de5b9c0e9c.json new file mode 100644 index 00000000000..43b3cbce92e --- /dev/null +++ b/.sqlx/query-31a9225ee20553ae34d112d0b70bfa85e1000cf48342e9c8eeb159de5b9c0e9c.json @@ -0,0 +1,41 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE internal.data_plane_private_links SET\n config = $2\n WHERE id = $1\n RETURNING\n status,\n details as \"details: sqlx::types::Json\",\n error,\n observed_at as \"observed_at: chrono::DateTime\"\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "status", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "details: sqlx::types::Json", + "type_info": "Jsonb" + }, + { + "ordinal": 2, + "name": "error", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "observed_at: chrono::DateTime", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Macaddr8", + "Jsonb" + ] + }, + "nullable": [ + false, + true, + true, + true + ] + }, + "hash": "31a9225ee20553ae34d112d0b70bfa85e1000cf48342e9c8eeb159de5b9c0e9c" +} diff --git a/.sqlx/query-569dccb049de8992f60d541d4fa5cc93466162e2f3e084d8dcc3a3e30acf7aee.json b/.sqlx/query-569dccb049de8992f60d541d4fa5cc93466162e2f3e084d8dcc3a3e30acf7aee.json new file mode 100644 index 00000000000..c09c542c96a --- /dev/null +++ b/.sqlx/query-569dccb049de8992f60d541d4fa5cc93466162e2f3e084d8dcc3a3e30acf7aee.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM internal.data_plane_private_links WHERE id = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Macaddr8" + ] + }, + "nullable": [] + }, + "hash": "569dccb049de8992f60d541d4fa5cc93466162e2f3e084d8dcc3a3e30acf7aee" +} diff --git a/.sqlx/query-7fab372a5bba76a7e743c58e40674728b86e6f63caaebc59aefac93a43c6a0fa.json b/.sqlx/query-7fab372a5bba76a7e743c58e40674728b86e6f63caaebc59aefac93a43c6a0fa.json new file mode 100644 index 00000000000..112c0a9c9a3 --- /dev/null +++ b/.sqlx/query-7fab372a5bba76a7e743c58e40674728b86e6f63caaebc59aefac93a43c6a0fa.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT dp.data_plane_name\n FROM internal.data_plane_private_links l\n JOIN data_planes dp ON dp.id = l.data_plane_id\n WHERE l.id = $1\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "data_plane_name", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Macaddr8" + ] + }, + "nullable": [ + false + ] + }, + "hash": "7fab372a5bba76a7e743c58e40674728b86e6f63caaebc59aefac93a43c6a0fa" +} diff --git a/.sqlx/query-81951888355221339404af3eac1005d173c2d0204b3cfdf07ad52895dd18951b.json b/.sqlx/query-81951888355221339404af3eac1005d173c2d0204b3cfdf07ad52895dd18951b.json new file mode 100644 index 00000000000..7d61f29057f --- /dev/null +++ b/.sqlx/query-81951888355221339404af3eac1005d173c2d0204b3cfdf07ad52895dd18951b.json @@ -0,0 +1,72 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT\n data_plane_id as \"data_plane_id: models::Id\",\n id as \"id: models::Id\",\n config as \"config!: sqlx::types::Json\",\n status,\n details as \"details: sqlx::types::Json\",\n error,\n observed_at as \"observed_at: chrono::DateTime\"\n FROM internal.data_plane_private_links\n WHERE data_plane_id = ANY($1::flowid[])\n ORDER BY data_plane_id, created_at, id\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "data_plane_id: models::Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 1, + "name": "id: models::Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 2, + "name": "config!: sqlx::types::Json", + "type_info": "Jsonb" + }, + { + "ordinal": 3, + "name": "status", + "type_info": "Text" + }, + { + "ordinal": 4, + "name": "details: sqlx::types::Json", + "type_info": "Jsonb" + }, + { + "ordinal": 5, + "name": "error", + "type_info": "Text" + }, + { + "ordinal": 6, + "name": "observed_at: chrono::DateTime", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + { + "Custom": { + "name": "flowid[]", + "kind": { + "Array": { + "Custom": { + "name": "flowid", + "kind": { + "Domain": "Macaddr8" + } + } + } + } + } + } + ] + }, + "nullable": [ + false, + false, + false, + false, + true, + true, + true + ] + }, + "hash": "81951888355221339404af3eac1005d173c2d0204b3cfdf07ad52895dd18951b" +} diff --git a/.sqlx/query-85b146f39fee0ab26b92ff4e37ef0b891a8778542d1d46e39ecc9515e2c16810.json b/.sqlx/query-85b146f39fee0ab26b92ff4e37ef0b891a8778542d1d46e39ecc9515e2c16810.json deleted file mode 100644 index 8ffa1f76aa0..00000000000 --- a/.sqlx/query-85b146f39fee0ab26b92ff4e37ef0b891a8778542d1d46e39ecc9515e2c16810.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE data_planes\n SET private_links = $2, updated_at = now()\n WHERE data_plane_name = $1\n RETURNING private_links as \"private_links!: Vec\"\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "private_links!: Vec", - "type_info": "JsonArray" - } - ], - "parameters": { - "Left": [ - "Text", - "JsonArray" - ] - }, - "nullable": [ - false - ] - }, - "hash": "85b146f39fee0ab26b92ff4e37ef0b891a8778542d1d46e39ecc9515e2c16810" -} diff --git a/.sqlx/query-a93ae7bf9e1f6b70def0680845e783a097cab9a98b20f17a5f125f993c295062.json b/.sqlx/query-9dbbc797f9e651eb717f5430186ae1d15adc381c81c4a832f458bcd3171c4044.json similarity index 69% rename from .sqlx/query-a93ae7bf9e1f6b70def0680845e783a097cab9a98b20f17a5f125f993c295062.json rename to .sqlx/query-9dbbc797f9e651eb717f5430186ae1d15adc381c81c4a832f458bcd3171c4044.json index f7578f37405..ad17dc79186 100644 --- a/.sqlx/query-a93ae7bf9e1f6b70def0680845e783a097cab9a98b20f17a5f125f993c295062.json +++ b/.sqlx/query-9dbbc797f9e651eb717f5430186ae1d15adc381c81c4a832f458bcd3171c4044.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "select\n dp.data_plane_name,\n dp.cidr_blocks::text[] as \"cidr_blocks!: Vec\",\n dp.gcp_service_account_email,\n dp.aws_iam_user_arn,\n dp.azure_application_name,\n dp.azure_application_client_id,\n dp.private_links as \"private_links!: Vec\",\n dp.aws_link_endpoints as \"aws_link_endpoints: Vec\",\n dp.azure_link_endpoints as \"azure_link_endpoints: Vec\",\n dp.gcp_psc_endpoints as \"gcp_psc_endpoints: Vec\"\n from unnest($1::text[]) as input(name)\n join data_planes dp on dp.data_plane_name = input.name\n ", + "query": "select\n dp.data_plane_name,\n dp.cidr_blocks::text[] as \"cidr_blocks!: Vec\",\n dp.gcp_service_account_email,\n dp.aws_iam_user_arn,\n dp.azure_application_name,\n dp.azure_application_client_id,\n dp.aws_link_endpoints as \"aws_link_endpoints: Vec\",\n dp.azure_link_endpoints as \"azure_link_endpoints: Vec\",\n dp.gcp_psc_endpoints as \"gcp_psc_endpoints: Vec\"\n from unnest($1::text[]) as input(name)\n join data_planes dp on dp.data_plane_name = input.name\n ", "describe": { "columns": [ { @@ -35,21 +35,16 @@ }, { "ordinal": 6, - "name": "private_links!: Vec", - "type_info": "JsonArray" - }, - { - "ordinal": 7, "name": "aws_link_endpoints: Vec", "type_info": "JsonArray" }, { - "ordinal": 8, + "ordinal": 7, "name": "azure_link_endpoints: Vec", "type_info": "JsonArray" }, { - "ordinal": 9, + "ordinal": 8, "name": "gcp_psc_endpoints: Vec", "type_info": "JsonArray" } @@ -66,11 +61,10 @@ true, true, true, - false, true, true, true ] }, - "hash": "a93ae7bf9e1f6b70def0680845e783a097cab9a98b20f17a5f125f993c295062" + "hash": "9dbbc797f9e651eb717f5430186ae1d15adc381c81c4a832f458bcd3171c4044" } diff --git a/.sqlx/query-a0c421220700d7e356553f5fbdceb74cae7803adeed28c851bd6a1c697d78561.json b/.sqlx/query-a0c421220700d7e356553f5fbdceb74cae7803adeed28c851bd6a1c697d78561.json new file mode 100644 index 00000000000..ab41a2226a6 --- /dev/null +++ b/.sqlx/query-a0c421220700d7e356553f5fbdceb74cae7803adeed28c851bd6a1c697d78561.json @@ -0,0 +1,47 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO internal.data_plane_private_links (data_plane_id, config)\n SELECT dp.id, $2\n FROM data_planes dp WHERE dp.data_plane_name = $1\n RETURNING\n id as \"id: models::Id\",\n status,\n details as \"details: sqlx::types::Json\",\n error,\n observed_at as \"observed_at: chrono::DateTime\"\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id: models::Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 1, + "name": "status", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "details: sqlx::types::Json", + "type_info": "Jsonb" + }, + { + "ordinal": 3, + "name": "error", + "type_info": "Text" + }, + { + "ordinal": 4, + "name": "observed_at: chrono::DateTime", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Text", + "Jsonb" + ] + }, + "nullable": [ + false, + false, + true, + true, + true + ] + }, + "hash": "a0c421220700d7e356553f5fbdceb74cae7803adeed28c851bd6a1c697d78561" +} diff --git a/.sqlx/query-c53448e2f383127421fd3f060629fb2dd02304ce4b51045d729b390497137d75.json b/.sqlx/query-a52cf45e935f28427ba48d2f181134a3bb943053fa46f58c0f1bba820a11fc5a.json similarity index 72% rename from .sqlx/query-c53448e2f383127421fd3f060629fb2dd02304ce4b51045d729b390497137d75.json rename to .sqlx/query-a52cf45e935f28427ba48d2f181134a3bb943053fa46f58c0f1bba820a11fc5a.json index 5724f4c8a74..f7b2dfa9401 100644 --- a/.sqlx/query-c53448e2f383127421fd3f060629fb2dd02304ce4b51045d729b390497137d75.json +++ b/.sqlx/query-a52cf45e935f28427ba48d2f181134a3bb943053fa46f58c0f1bba820a11fc5a.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT\n id as \"data_plane_id: models::Id\",\n config AS \"config: sqlx::types::Json\",\n deploy_branch AS \"deploy_branch!\",\n logs_token,\n data_plane_name,\n data_plane_fqdn,\n private_links AS \"private_links: Vec>\",\n pulumi_key AS \"pulumi_key\",\n pulumi_stack AS \"pulumi_stack!\"\n FROM data_planes\n WHERE controller_task_id = $1\n ", + "query": "\n SELECT\n id as \"data_plane_id: models::Id\",\n config AS \"config: sqlx::types::Json\",\n deploy_branch AS \"deploy_branch!\",\n logs_token,\n data_plane_name,\n data_plane_fqdn,\n pulumi_key AS \"pulumi_key\",\n pulumi_stack AS \"pulumi_stack!\"\n FROM data_planes\n WHERE controller_task_id = $1\n ", "describe": { "columns": [ { @@ -35,16 +35,11 @@ }, { "ordinal": 6, - "name": "private_links: Vec>", - "type_info": "JsonArray" - }, - { - "ordinal": 7, "name": "pulumi_key", "type_info": "Text" }, { - "ordinal": 8, + "ordinal": 7, "name": "pulumi_stack!", "type_info": "Text" } @@ -61,10 +56,9 @@ false, false, false, - false, true, true ] }, - "hash": "c53448e2f383127421fd3f060629fb2dd02304ce4b51045d729b390497137d75" + "hash": "a52cf45e935f28427ba48d2f181134a3bb943053fa46f58c0f1bba820a11fc5a" } diff --git a/crates/control-plane-api/src/fixtures/private_links.sql b/crates/control-plane-api/src/fixtures/private_links.sql index f3edc31be7a..f5eedca8b32 100644 --- a/crates/control-plane-api/src/fixtures/private_links.sql +++ b/crates/control-plane-api/src/fixtures/private_links.sql @@ -1,12 +1,15 @@ --- Adds a private data plane with populated private-link config and one AWS --- provisioning result row, plus a `read` grant for Alice on the private --- prefix so the GraphQL authorization layer can surface it. +-- Adds a private data plane plus three configured private links (one per +-- provider) as rows in `data_plane_private_links`: the AWS link is +-- `provisioned` and carries its endpoint details, the Azure and GCP links are +-- still `pending`. One AWS endpoint result is also set on the data_planes row +-- for the legacy `awsLinkEndpoints` field. Grants Alice `read` (plus the +-- `manage_data_plane` bundle) on the private prefix so the GraphQL +-- authorization layer can surface and mutate it. -- --- Loaded alongside `data_planes` and `alice` in tests that exercise the --- typed `privateLinks` field on the dataPlanes query and the --- `updateDataPlanePrivateLinks` mutation. Kept separate from the shared --- `data_planes.sql` so its presence does not change every other GraphQL --- test. +-- Loaded alongside `data_planes` and `alice` by tests that exercise the +-- `privateLinks` field and the per-link CRUD mutations. Kept separate from the +-- shared `data_planes.sql` so its presence does not change every other GraphQL +-- test. Link ids are explicit so snapshots are deterministic. do $$ declare alice_private_dp_id flowid := '444444444444'; @@ -38,7 +41,6 @@ begin gcp_service_account_email, azure_application_name, azure_application_client_id, - private_links, aws_link_endpoints ) values ( alice_private_dp_id, @@ -63,15 +65,33 @@ begin 'estuary-test-app-private', '44444444-4444-4444-4444-444444444444', array[ - '{"region":"us-east-1","az_ids":["use1-az1","use1-az2"],"service_name":"com.amazonaws.vpce.us-east-1.vpce-svc-abc123"}'::json, - '{"service_name":"/subscriptions/x/resourceGroups/rg/providers/Microsoft.Network/privateLinkServices/svc","location":"eastus","dns_name":"privatelink.database.windows.net"}'::json, - '{"service_attachment":"projects/p/regions/us-central1/serviceAttachments/sa","region":"us-central1","dns_zone_name":"z","dns_record_names":["r1","r2"],"all_ports":true}'::json - ], - array[ - '{"endpoint_id":"vpce-0123456789abcdef0","state":"available"}'::json + '{"service_name":"com.amazonaws.vpce.us-east-1.vpce-svc-abc123","dns_entries":[{"dns_name":"vpce-0123abc.vpce-svc-abc123.us-east-1.vpce.amazonaws.com","hosted_zone_id":"Z7HUB22EVRPK5"}]}'::json ] ); + insert into internal.data_plane_private_links (id, data_plane_id, config, status, details) values + ( + '00:00:00:00:00:00:0a:01', + alice_private_dp_id, + '{"region":"us-east-1","az_ids":["use1-az1","use1-az2"],"service_name":"com.amazonaws.vpce.us-east-1.vpce-svc-abc123"}'::jsonb, + 'provisioned', + '{"service_name":"com.amazonaws.vpce.us-east-1.vpce-svc-abc123","dns_entries":[{"dns_name":"vpce-0123abc.vpce-svc-abc123.us-east-1.vpce.amazonaws.com","hosted_zone_id":"Z7HUB22EVRPK5"}]}'::jsonb + ), + ( + '00:00:00:00:00:00:0a:02', + alice_private_dp_id, + '{"service_name":"/subscriptions/x/resourceGroups/rg/providers/Microsoft.Network/privateLinkServices/svc","location":"eastus","dns_name":"privatelink.database.windows.net"}'::jsonb, + 'pending', + null + ), + ( + '00:00:00:00:00:00:0a:03', + alice_private_dp_id, + '{"service_attachment":"projects/p/regions/us-central1/serviceAttachments/sa","region":"us-central1","dns_zone_name":"z","dns_record_names":["r1","r2"],"all_ports":true}'::jsonb, + 'pending', + null + ); + -- Mirrors what `create_data_plane.rs` installs at provisioning time: -- legacy `read` for RLS/`user_roles()`, and the `ManageDataPlane` bundle -- for the capability bits. diff --git a/crates/control-plane-api/src/server/public/graphql/data_planes.rs b/crates/control-plane-api/src/server/public/graphql/data_planes.rs index 56cd13d8f6f..690dcac5679 100644 --- a/crates/control-plane-api/src/server/public/graphql/data_planes.rs +++ b/crates/control-plane-api/src/server/public/graphql/data_planes.rs @@ -15,6 +15,106 @@ pub enum DataPlaneCloudProvider { Local, } +/// Controller-observed provisioning status of a configured private link. +#[derive(Debug, Clone, Copy, PartialEq, Eq, async_graphql::Enum)] +pub enum PrivateLinkProvisioningStatus { + /// Not yet provisioned for the current configuration. + Pending, + /// Provisioned; `details` describes the endpoint. + Provisioned, + /// Provisioning failed; see `error`. + Failed, +} + +impl PrivateLinkProvisioningStatus { + fn from_db(s: &str) -> async_graphql::Result { + match s { + "pending" => Ok(Self::Pending), + "provisioned" => Ok(Self::Provisioned), + "failed" => Ok(Self::Failed), + other => Err(async_graphql::Error::new(format!( + "unknown private link status {other:?}" + ))), + } + } +} + +/// A configured private link and its controller-observed provisioning status. +#[derive(Debug, Clone, SimpleObject)] +#[graphql(name = "PrivateLink")] +pub struct DataPlanePrivateLink { + /// Stable identifier of this private link. + pub id: models::Id, + /// The link configuration (AWS PrivateLink, Azure Private Link, or GCP PSC). + /// Its variant (`AWSPrivateLink`/`AzurePrivateLink`/`GCPPrivateServiceConnect`) + /// is the link's cloud provider. + pub config: models::PrivateLink, + /// Controller-observed provisioning status. + pub status: PrivateLinkProvisioningStatus, + /// Provider-specific provisioning details (DNS entries, IPs) once + /// provisioned; opaque JSON exported by the data-plane controller. + pub details: Option>, + /// Failure detail when `status` is `failed`. + pub error: Option, + /// When the controller last observed this link's status. + pub observed_at: Option>, +} + +/// Keys the request-scoped loader by data plane so sibling `privateLinks` +/// resolvers collapse into one batch query. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +struct DataPlanePrivateLinksKey(models::Id); + +impl async_graphql::dataloader::Loader for super::PgDataLoader { + type Value = Vec; + type Error = String; + + async fn load( + &self, + keys: &[DataPlanePrivateLinksKey], + ) -> Result, Self::Error> { + let data_plane_ids: Vec = keys.iter().map(|key| key.0).collect(); + let rows = sqlx::query!( + r#" + SELECT + data_plane_id as "data_plane_id: models::Id", + id as "id: models::Id", + config as "config!: sqlx::types::Json", + status, + details as "details: sqlx::types::Json", + error, + observed_at as "observed_at: chrono::DateTime" + FROM internal.data_plane_private_links + WHERE data_plane_id = ANY($1::flowid[]) + ORDER BY data_plane_id, created_at, id + "#, + &data_plane_ids as &[models::Id], + ) + .fetch_all(&self.0) + .await + .map_err(|err| format!("failed to fetch data plane private links: {err}"))?; + + let mut result: HashMap> = + HashMap::new(); + for row in rows { + let status = + PrivateLinkProvisioningStatus::from_db(&row.status).map_err(|err| err.message)?; + result + .entry(DataPlanePrivateLinksKey(row.data_plane_id)) + .or_default() + .push(DataPlanePrivateLink { + id: row.id, + config: row.config.0, + status, + details: row.details.map(|details| async_graphql::Json(details.0)), + error: row.error, + observed_at: row.observed_at, + }); + } + Ok(result) + } +} + /// A data plane where tasks execute and collections are stored. #[derive(Debug, Clone, SimpleObject)] #[graphql(complex)] @@ -46,13 +146,14 @@ pub struct DataPlane { pub azure_application_name: Option, /// Azure application client ID for this data-plane. pub azure_application_client_id: Option, - // The four private-networking fields below are gated behind - // `ViewDataPlanePrivateNetworking` and resolved by `ComplexObject` methods. - // They are stored as raw JSON and skipped from the derived object so the - // capability check lives with the field rather than the construction site; - // see the resolvers below. + // The private-networking fields below are gated behind + // `ViewDataPlanePrivateNetworking` and resolved by `ComplexObject` methods, + // so the capability check lives with the field rather than the construction + // site; see the resolvers below. `control_id` lets the `private_links` + // resolver query the `data_plane_private_links` table; the endpoint arrays + // are raw JSON exported by the controller. #[graphql(skip)] - raw_private_links: Vec, + control_id: models::Id, #[graphql(skip)] raw_aws_link_endpoints: Vec, #[graphql(skip)] @@ -63,15 +164,15 @@ pub struct DataPlane { #[ComplexObject] impl DataPlane { - /// Configured private link endpoints for this data-plane. Replacing this - /// list (via `updateDataPlanePrivateLinks`) triggers reconvergence by the - /// data-plane controller on its next poll. Returns an empty list to - /// callers that lack the `ViewDataPlanePrivateNetworking` capability on - /// this data plane. + /// Configured private links for this data-plane, each with its + /// controller-observed provisioning status. Mutating links (via + /// `addDataPlanePrivateLink` and friends) triggers reconvergence by the + /// data-plane controller. Returns an empty list to callers that lack the + /// `ViewDataPlanePrivateNetworking` capability on this data plane. async fn private_links( &self, ctx: &Context<'_>, - ) -> async_graphql::Result> { + ) -> async_graphql::Result> { if !super::may_access( ctx, &self.name, @@ -79,18 +180,11 @@ impl DataPlane { )? { return Ok(Vec::new()); } - self.raw_private_links - .iter() - .enumerate() - .map(|(idx, raw)| { - serde_json::from_value::(raw.clone()).map_err(|err| { - async_graphql::Error::new(format!( - "failed to parse private_links[{idx}] for data plane {}: {err}", - self.name, - )) - }) - }) - .collect() + let loader = ctx.data::>()?; + Ok(loader + .load_one(DataPlanePrivateLinksKey(self.control_id)) + .await? + .unwrap_or_default()) } /// AWS PrivateLink endpoint provisioning results, opaque JSON exported by @@ -176,7 +270,6 @@ async fn fetch_data_plane_details( dp.aws_iam_user_arn, dp.azure_application_name, dp.azure_application_client_id, - dp.private_links as "private_links!: Vec", dp.aws_link_endpoints as "aws_link_endpoints: Vec", dp.azure_link_endpoints as "azure_link_endpoints: Vec", dp.gcp_psc_endpoints as "gcp_psc_endpoints: Vec" @@ -199,7 +292,6 @@ async fn fetch_data_plane_details( aws_iam_user_arn: row.aws_iam_user_arn, azure_application_name: row.azure_application_name, azure_application_client_id: row.azure_application_client_id, - private_links: row.private_links, aws_link_endpoints: row.aws_link_endpoints.unwrap_or_default(), azure_link_endpoints: row.azure_link_endpoints.unwrap_or_default(), gcp_psc_endpoints: row.gcp_psc_endpoints.unwrap_or_default(), @@ -215,7 +307,6 @@ struct DataPlaneDetails { aws_iam_user_arn: Option, azure_application_name: Option, azure_application_client_id: Option, - private_links: Vec, aws_link_endpoints: Vec, azure_link_endpoints: Vec, gcp_psc_endpoints: Vec, @@ -453,7 +544,7 @@ impl DataPlanesQuery { azure_application_name: details.and_then(|d| d.azure_application_name.clone()), azure_application_client_id: details .and_then(|d| d.azure_application_client_id.clone()), - raw_private_links: details.map(|d| d.private_links.clone()).unwrap_or_default(), + control_id: dp.control_id, raw_aws_link_endpoints: details .map(|d| d.aws_link_endpoints.clone()) .unwrap_or_default(), @@ -537,41 +628,88 @@ impl DataPlanesQuery { #[derive(Debug, Default)] pub struct DataPlanesMutation; +/// Structural check: the name must sit under `ops/dp/private/` with at least +/// one path segment beyond it. Anything more specific (cluster suffix shape, +/// owning prefix shape) is the data plane's problem; an unknown but well-formed +/// name falls out as "not found" when no `data_planes` row matches. +fn require_private_dp_name(name: &str) -> async_graphql::Result<()> { + if name + .strip_prefix("ops/dp/private/") + .is_none_or(|rest| !rest.contains('/') || rest.starts_with('/')) + { + return Err(async_graphql::Error::new(format!( + "{name} is not a private data-plane name" + ))); + } + Ok(()) +} + +/// Maps a unique-violation on `(data_plane_id, service_identity)` to a clear +/// message; other database errors propagate unchanged. +fn map_link_db_error(err: sqlx::Error) -> async_graphql::Error { + if let sqlx::Error::Database(db) = &err { + if db.is_unique_violation() { + return async_graphql::Error::new( + "a private link with this service identity already exists on this data plane", + ); + } + } + async_graphql::Error::new(err.to_string()) +} + +/// Resolves an id-addressed private link, authorizes the caller to modify it, +/// and returns the owning data-plane name. A link that does not exist and a link +/// the caller may not modify both return the same "not found" error, so an +/// unauthorized caller cannot probe which link ids exist. This deliberately uses +/// the visibility gate ([`super::may_access`]) rather than the hard gate +/// ([`super::verify_authorization`]) so a denial is hidden as not-found instead +/// of surfacing as a distinguishable permission-denied that names the data plane. +async fn resolve_modifiable_link( + ctx: &Context<'_>, + id: models::Id, +) -> async_graphql::Result { + let env = ctx.data::()?; + let not_found = || async_graphql::Error::new(format!("private link '{id}' not found")); + + let Some(row) = sqlx::query!( + r#" + SELECT dp.data_plane_name + FROM internal.data_plane_private_links l + JOIN data_planes dp ON dp.id = l.data_plane_id + WHERE l.id = $1 + "#, + id as models::Id, + ) + .fetch_optional(&env.pg_pool) + .await? + else { + return Err(not_found()); + }; + + if !super::may_access( + ctx, + &row.data_plane_name, + models::authz::Capability::ModifyDataPlanePrivateNetworking, + )? { + return Err(not_found()); + } + + Ok(row.data_plane_name) +} + #[async_graphql::Object] impl DataPlanesMutation { - /// Replaces the configured private link endpoints on a private data plane. - /// - /// The provided list overwrites the entire `private_links` column; partial - /// updates are intentionally not supported. The data-plane controller - /// converges to the new configuration on its next poll. Returns the desired - /// private links state. The `*LinkEndpoints` provisioning results are not echoed here: - /// they lag this write until the controller converges, so callers needing them re-query `dataPlanes`. - /// - /// Requires the `ModifyDataPlanePrivateNetworking` capability on the - /// private data-plane name. - pub async fn update_data_plane_private_links( + /// Adds a private link to a private data plane. The data-plane controller + /// converges to provision it on its next poll; the returned link starts + /// `pending`. Requires `ModifyDataPlanePrivateNetworking` on the data plane. + pub async fn add_data_plane_private_link( &self, ctx: &Context<'_>, data_plane_name: String, - private_links: Vec, - ) -> async_graphql::Result> { + config: models::PrivateLink, + ) -> async_graphql::Result { let env = ctx.data::()?; - let claims = env.claims()?; - - // Structural check only: the name must sit under `ops/dp/private/` and - // have at least one path segment beyond it. Anything more specific - // (cluster suffix shape, owning prefix shape) is the data plane's - // problem, not the mutation's; an unknown name falls out as "not - // found" when the UPDATE matches zero rows. - if data_plane_name - .strip_prefix("ops/dp/private/") - .is_none_or(|rest| !rest.contains('/') || rest.starts_with('/')) - { - return Err(async_graphql::Error::new(format!( - "{data_plane_name} is not a private data-plane name" - ))); - } - + require_private_dp_name(&data_plane_name)?; super::verify_authorization( env, &data_plane_name, @@ -579,19 +717,24 @@ impl DataPlanesMutation { ) .await?; - let bound: Vec> = - private_links.iter().map(sqlx::types::Json).collect(); let row = sqlx::query!( - r#"UPDATE data_planes - SET private_links = $2, updated_at = now() - WHERE data_plane_name = $1 - RETURNING private_links as "private_links!: Vec" + r#" + INSERT INTO internal.data_plane_private_links (data_plane_id, config) + SELECT dp.id, $2 + FROM data_planes dp WHERE dp.data_plane_name = $1 + RETURNING + id as "id: models::Id", + status, + details as "details: sqlx::types::Json", + error, + observed_at as "observed_at: chrono::DateTime" "#, data_plane_name, - &bound as &[sqlx::types::Json<&models::PrivateLink>], + sqlx::types::Json(&config) as sqlx::types::Json<&models::PrivateLink>, ) .fetch_optional(&env.pg_pool) - .await?; + .await + .map_err(map_link_db_error)?; let Some(row) = row else { return Err(async_graphql::Error::new(format!( @@ -599,22 +742,89 @@ impl DataPlanesMutation { ))); }; - tracing::info!( - %data_plane_name, - link_count = row.private_links.len(), - %claims.sub, - "updated data plane private links", - ); + tracing::info!(%data_plane_name, link_id = %row.id, "added data plane private link"); - row.private_links - .into_iter() - .map(serde_json::from_value::) - .collect::, _>>() - .map_err(|err| { - async_graphql::Error::new(format!( - "stored private_links for {data_plane_name} did not round-trip: {err}" - )) - }) + Ok(DataPlanePrivateLink { + id: row.id, + config, + status: PrivateLinkProvisioningStatus::from_db(&row.status)?, + details: row.details.map(|d| async_graphql::Json(d.0)), + error: row.error, + observed_at: row.observed_at, + }) + } + + /// Replaces the configuration of an existing private link by id. A changed + /// config resets the observed status to `pending` and re-triggers convergence: + /// the desired-edit trigger clears the observation columns and bumps the + /// link's internal generation, so a converge already in flight against the + /// previous configuration cannot later stamp this link with a stale status. + /// Requires `ModifyDataPlanePrivateNetworking` on the owning data plane. + pub async fn update_data_plane_private_link( + &self, + ctx: &Context<'_>, + id: models::Id, + config: models::PrivateLink, + ) -> async_graphql::Result { + let env = ctx.data::()?; + resolve_modifiable_link(ctx, id).await?; + + // Only the desired config is set here. When it differs, the desired-edit + // trigger resets the observation and bumps generation in the same write; + // `RETURNING` reflects either the reset or the unchanged observation. + let row = sqlx::query!( + r#" + UPDATE internal.data_plane_private_links SET + config = $2 + WHERE id = $1 + RETURNING + status, + details as "details: sqlx::types::Json", + error, + observed_at as "observed_at: chrono::DateTime" + "#, + id as models::Id, + sqlx::types::Json(&config) as sqlx::types::Json<&models::PrivateLink>, + ) + .fetch_optional(&env.pg_pool) + .await + .map_err(map_link_db_error)? + // The row was authorized by `resolve_modifiable_link` above, but a + // concurrent remove (or a cascading data-plane teardown) can delete it + // before this UPDATE runs. Report the same existence-hiding not-found + // rather than leaking a raw "no rows returned" sqlx error. + .ok_or_else(|| async_graphql::Error::new(format!("private link '{id}' not found")))?; + + Ok(DataPlanePrivateLink { + id, + config, + status: PrivateLinkProvisioningStatus::from_db(&row.status)?, + details: row.details.map(|d| async_graphql::Json(d.0)), + error: row.error, + observed_at: row.observed_at, + }) + } + + /// Removes a private link by id. The controller tears down its endpoint on + /// the next converge. Requires `ModifyDataPlanePrivateNetworking` on the + /// owning data plane. Returns the removed link id. + pub async fn remove_data_plane_private_link( + &self, + ctx: &Context<'_>, + id: models::Id, + ) -> async_graphql::Result { + let env = ctx.data::()?; + let data_plane_name = resolve_modifiable_link(ctx, id).await?; + + _ = sqlx::query!( + "DELETE FROM internal.data_plane_private_links WHERE id = $1", + id as models::Id, + ) + .execute(&env.pg_pool) + .await?; + + tracing::info!(link_id = %id, %data_plane_name, "removed data plane private link"); + Ok(id) } } @@ -794,24 +1004,14 @@ mod tests { azureLinkEndpoints gcpPscEndpoints privateLinks { - __typename - ... on AWSPrivateLink { - region - azIds - serviceName - } - ... on AzurePrivateLink { - serviceName - location - dnsName - resourceType - } - ... on GCPPrivateServiceConnect { - serviceAttachment - region - dnsZoneName - dnsRecordNames - allPorts + id + status + details + config { + __typename + ... on AWSPrivateLink { region azIds serviceName } + ... on AzurePrivateLink { serviceName location dnsName resourceType } + ... on GCPPrivateServiceConnect { serviceAttachment region dnsZoneName dnsRecordNames allPorts } } } } @@ -827,12 +1027,10 @@ mod tests { insta::assert_json_snapshot!("data_planes_with_private_links", response); } - // A caller with only legacy `read` on the DP prefix can view the - // private-networking fields (the `Viewer` bundle carries - // `ViewDataPlanePrivateNetworking`, because `read` on a data-plane - // prefix already conveys deploy-level trust) but cannot mutate them: - // `ModifyDataPlanePrivateNetworking` only comes via the separately - // granted `ManageDataPlane` bundle. + // A malformed config must cause the `privateLinks` resolver to return a + // GraphQL error, rather than silently omit a row the caller may view. + // The migration validates rows present at backfill time, but support + // and other internal writers can edit this unvalidated JSONB column later. #[sqlx::test( migrations = "../../supabase/migrations", fixtures( @@ -840,21 +1038,16 @@ mod tests { scripts("data_planes", "alice", "private_links") ) )] - async fn test_read_grants_view_but_not_modify(pool: sqlx::PgPool) { + async fn test_graphql_data_planes_malformed_private_link(pool: sqlx::PgPool) { let _guard = test_server::init(); + // Retain the service identity while removing the other required AWS + // fields, producing a row that matches no `models::PrivateLink` variant. sqlx::query( - "INSERT INTO auth.users (id, email) VALUES \ - ('22222222-2222-2222-2222-222222222222', 'bob@example.test')", - ) - .execute(&pool) - .await - .unwrap(); - sqlx::query( - "INSERT INTO user_grants (user_id, object_role, capability) VALUES \ - ($1, 'ops/dp/private/aliceCo/', 'read')", + r#"UPDATE internal.data_plane_private_links + SET config = '{"service_name":"com.amazonaws.vpce.us-east-1.vpce-svc-malformed"}'::jsonb + WHERE id = '00:00:00:00:00:00:0a:01'"#, ) - .bind(uuid::Uuid::from_bytes([0x22; 16])) .execute(&pool) .await .unwrap(); @@ -862,8 +1055,7 @@ mod tests { let server = test_server::TestServer::start(pool.clone(), test_server::snapshot(pool, false).await) .await; - let bob_token = - server.make_access_token(uuid::Uuid::from_bytes([0x22; 16]), Some("bob@example.test")); + let token = server.make_access_token(uuid::Uuid::from_bytes([0x11; 16]), None); let response: serde_json::Value = server .graphql( @@ -874,55 +1066,22 @@ mod tests { edges { node { name - privateLinks { __typename } - awsLinkEndpoints - azureLinkEndpoints - gcpPscEndpoints + privateLinks { id } } } } } "# }), - Some(&bob_token), + Some(&token), ) .await; - let edges = response["data"]["dataPlanes"]["edges"] - .as_array() - .expect("should have edges"); - let private_dp = edges - .iter() - .find(|e| e["node"]["name"] == "ops/dp/private/aliceCo/aws-us-east-1-c1") - .expect("bob should see the private dp via his read grant"); - // The fixture populates three private links and one AWS provisioning - // result; bob's `read` is enough to view all of them. - assert_eq!( - private_dp["node"]["privateLinks"].as_array().unwrap().len(), - 3, - "read must grant view of private links: {private_dp}", - ); - assert_eq!( - private_dp["node"]["awsLinkEndpoints"] - .as_array() - .unwrap() - .len(), - 1, - "read must grant view of endpoint provisioning results: {private_dp}", - ); - - // Mutating requires `ModifyDataPlanePrivateNetworking`, which `read` - // does not carry. - let bob_denied: serde_json::Value = server - .graphql( - &update_mutation("ops/dp/private/aliceCo/aws-us-east-1-c1", VALID_AWS_INPUT), - Some(&bob_token), - ) - .await; - assert_eq!( - first_error_message(&bob_denied), - "PermissionDenied: bob@example.test is not authorized to access prefix or name 'ops/dp/private/aliceCo/aws-us-east-1-c1' with required capability ModifyDataPlanePrivateNetworking", + assert!( + response["data"].is_null(), + "expected fail-closed data: {response}" ); + insta::assert_json_snapshot!("data_planes_malformed_private_link", response); } // Existing tenants can still view their private data plane's private links @@ -993,7 +1152,7 @@ mod tests { // Modify is denied: ModifyDataPlanePrivateNetworking flowed only // through the now-cleared `manage_data_plane` bundle on the edge. let denied: serde_json::Value = server - .graphql(&update_mutation(dp, VALID_AWS_INPUT), Some(&alice_token)) + .graphql(&add_mutation(dp, VALID_AWS_INPUT), Some(&alice_token)) .await; assert_eq!( first_error_message(&denied), @@ -1001,62 +1160,11 @@ mod tests { ); } - // A malformed `private_links` row produces a field-level error that names - // the data plane and the failing index. Because `privateLinks` is declared - // `[PrivateLink!]!` (non-null), the error null-propagates up to the - // nullable root and the whole `data` field comes back as null; the error - // path locates the offending edge. - #[sqlx::test( - migrations = "../../supabase/migrations", - fixtures( - path = "../../../fixtures", - scripts("data_planes", "alice", "private_links") - ) - )] - async fn test_graphql_data_planes_malformed_private_link(pool: sqlx::PgPool) { - let _guard = test_server::init(); - - // Corrupt the private_links column for the private dp before snapshot. - sqlx::query( - r#"UPDATE data_planes - SET private_links = array['{"not":"a private link"}'::json] - WHERE data_plane_name = 'ops/dp/private/aliceCo/aws-us-east-1-c1'"#, - ) - .execute(&pool) - .await - .unwrap(); - - let server = - test_server::TestServer::start(pool.clone(), test_server::snapshot(pool, false).await) - .await; - - let token = server.make_access_token(uuid::Uuid::from_bytes([0x11; 16]), None); - - let response: serde_json::Value = server - .graphql( - &serde_json::json!({ - "query": r#" - query { - dataPlanes { - edges { - node { - name - privateLinks { __typename } - } - } - } - } - "# - }), - Some(&token), - ) - .await; - - insta::assert_json_snapshot!("data_planes_malformed_private_link", response); - } - - // ===== updateDataPlanePrivateLinks mutation tests ===== + // ===== per-link CRUD mutation tests ===== + // The `*_INPUT` constants are `PrivateLinkConfigInput` @oneOf values. The + // AWS one matches the fixture's existing AWS link (used to exercise the + // duplicate-identity guard); `NEW_AWS_INPUT` is a distinct link to add. const VALID_AWS_INPUT: &str = r#"{ "aws": { "region": "us-east-1", @@ -1064,44 +1172,79 @@ mod tests { "serviceName": "com.amazonaws.vpce.us-east-1.vpce-svc-abc123" } }"#; - const VALID_AZURE_INPUT: &str = r#"{ - "azure": { - "serviceName": "/subscriptions/x/resourceGroups/rg/providers/Microsoft.Network/privateLinkServices/svc", - "location": "eastus", - "dnsName": "privatelink.database.windows.net", - "resourceType": "" - } - }"#; - const VALID_GCP_INPUT: &str = r#"{ - "gcp": { - "serviceAttachment": "projects/p/regions/us-central1/serviceAttachments/sa", - "region": "us-central1", - "dnsZoneName": "z", - "dnsRecordNames": ["r1"], - "allPorts": true + const NEW_AWS_INPUT: &str = r#"{ + "aws": { + "region": "us-east-1", + "azIds": ["use1-az1"], + "serviceName": "com.amazonaws.vpce.us-east-1.vpce-svc-new999" } }"#; - fn update_mutation(name: &str, links_json: &str) -> serde_json::Value { - // The mutation echoes the stored links as the `PrivateLink` union, so - // the selection set spreads each variant's discriminating fields. + fn add_mutation(name: &str, config_json: &str) -> serde_json::Value { serde_json::json!({ "query": r#" - mutation($name: String!, $links: [PrivateLinkInput!]!) { - updateDataPlanePrivateLinks(dataPlaneName: $name, privateLinks: $links) { - __typename - ... on AWSPrivateLink { region serviceName } - ... on AzurePrivateLink { serviceName location } - ... on GCPPrivateServiceConnect { serviceAttachment region } + mutation($name: String!, $config: PrivateLinkConfigInput!) { + addDataPlanePrivateLink(dataPlaneName: $name, config: $config) { + id + status + config { + __typename + ... on AWSPrivateLink { serviceName } + ... on AzurePrivateLink { serviceName } + ... on GCPPrivateServiceConnect { serviceAttachment } + } } }"#, "variables": { "name": name, - "links": serde_json::from_str::(&format!("[{links_json}]")).unwrap(), + "config": serde_json::from_str::(config_json).unwrap(), + } + }) + } + + fn update_link_mutation(id: &str, config_json: &str) -> serde_json::Value { + serde_json::json!({ + "query": r#" + mutation($id: Id!, $config: PrivateLinkConfigInput!) { + updateDataPlanePrivateLink(id: $id, config: $config) { + id status config { __typename ... on AWSPrivateLink { serviceName } } + } + }"#, + "variables": { + "id": id, + "config": serde_json::from_str::(config_json).unwrap(), } }) } + fn remove_link_mutation(id: &str) -> serde_json::Value { + serde_json::json!({ + "query": r#" + mutation($id: Id!) { removeDataPlanePrivateLink(id: $id) }"#, + "variables": { "id": id } + }) + } + + /// Extracts the first error message from a GraphQL response, or panics + /// if the response did not return an error. + fn first_error_message(response: &serde_json::Value) -> &str { + response["errors"][0]["message"] + .as_str() + .unwrap_or_else(|| panic!("expected an error, got: {response}")) + } + + async fn count_links(pool: &sqlx::PgPool, dp: &str) -> i64 { + sqlx::query_scalar( + r#"SELECT count(*) FROM internal.data_plane_private_links l + JOIN data_planes dp ON dp.id = l.data_plane_id + WHERE dp.data_plane_name = $1"#, + ) + .bind(dp) + .fetch_one(pool) + .await + .unwrap() + } + #[sqlx::test( migrations = "../../supabase/migrations", fixtures( @@ -1109,89 +1252,108 @@ mod tests { scripts("data_planes", "alice", "private_links") ) )] - async fn test_update_private_links_happy_path(pool: sqlx::PgPool) { + async fn test_add_private_link(pool: sqlx::PgPool) { let _guard = test_server::init(); - let server = test_server::TestServer::start( pool.clone(), test_server::snapshot(pool.clone(), false).await, ) .await; let alice_token = server.make_access_token(uuid::Uuid::from_bytes([0x11; 16]), None); - let dp = "ops/dp/private/aliceCo/aws-us-east-1-c1"; - let links = format!("{VALID_AWS_INPUT},{VALID_AZURE_INPUT},{VALID_GCP_INPUT}"); - - let updated_at_before: chrono::DateTime = - sqlx::query_scalar("SELECT updated_at FROM data_planes WHERE data_plane_name = $1") - .bind(dp) - .fetch_one(&pool) - .await - .unwrap(); - let response: serde_json::Value = server - .graphql(&update_mutation(dp, &links), Some(&alice_token)) + // A new link is created `pending` (no endpoint provisioned yet) as a + // fourth row alongside the three from the fixture. + let added: serde_json::Value = server + .graphql(&add_mutation(dp, NEW_AWS_INPUT), Some(&alice_token)) .await; - // The mutation echoes the three submitted links in order, one per - // union variant. - let echoed = response["data"]["updateDataPlanePrivateLinks"] - .as_array() - .unwrap_or_else(|| panic!("expected echoed links, got: {response}")); - let typenames: Vec<&str> = echoed - .iter() - .map(|l| l["__typename"].as_str().unwrap()) - .collect(); + let link = &added["data"]["addDataPlanePrivateLink"]; assert_eq!( - typenames, - [ - "AWSPrivateLink", - "AzurePrivateLink", - "GCPPrivateServiceConnect" - ], + link["config"]["__typename"], "AWSPrivateLink", + "got: {added}" ); - assert_eq!(echoed[0]["region"], "us-east-1"); - - // Postgres `now()` is `transaction_timestamp()` at microsecond - // precision, so two distinct transactions return distinct values. - let updated_at_after: chrono::DateTime = - sqlx::query_scalar("SELECT updated_at FROM data_planes WHERE data_plane_name = $1") - .bind(dp) - .fetch_one(&pool) - .await - .unwrap(); - assert!( - updated_at_after > updated_at_before, - "updated_at must advance on a successful mutation" + assert_eq!(link["status"], "PENDING"); + assert_eq!( + link["config"]["serviceName"], + "com.amazonaws.vpce.us-east-1.vpce-svc-new999" ); + assert!(link["id"].is_string()); + assert_eq!(count_links(&pool, dp).await, 4); - // Calling again with a single AWS link replaces the entire array - // rather than merging. - let response: serde_json::Value = server - .graphql(&update_mutation(dp, VALID_AWS_INPUT), Some(&alice_token)) + // Adding a link whose service identity already exists on the data plane + // is rejected by the unique constraint. + let dup: serde_json::Value = server + .graphql(&add_mutation(dp, VALID_AWS_INPUT), Some(&alice_token)) .await; - let echoed = response["data"]["updateDataPlanePrivateLinks"] - .as_array() - .unwrap_or_else(|| panic!("expected echoed links, got: {response}")); - assert_eq!(echoed.len(), 1); - assert_eq!(echoed[0]["__typename"], "AWSPrivateLink"); + assert_eq!( + first_error_message(&dup), + "a private link with this service identity already exists on this data plane", + ); + assert_eq!(count_links(&pool, dp).await, 4); + } - // Confirm the second call replaced (rather than merged) the array. - let stored_count: i64 = sqlx::query_scalar( - "SELECT array_length(private_links, 1)::bigint FROM data_planes WHERE data_plane_name = $1", + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures( + path = "../../../fixtures", + scripts("data_planes", "alice", "private_links") ) - .bind(dp) - .fetch_one(&pool) - .await - .unwrap(); - assert_eq!(stored_count, 1); - } + )] + async fn test_update_and_remove_private_link(pool: sqlx::PgPool) { + let _guard = test_server::init(); + let server = test_server::TestServer::start( + pool.clone(), + test_server::snapshot(pool.clone(), false).await, + ) + .await; + let alice_token = server.make_access_token(uuid::Uuid::from_bytes([0x11; 16]), None); + let dp = "ops/dp/private/aliceCo/aws-us-east-1-c1"; - /// Extracts the first error message from a GraphQL response, or panics - /// if the response did not return an error. - fn first_error_message(response: &serde_json::Value) -> &str { - response["errors"][0]["message"] - .as_str() - .unwrap_or_else(|| panic!("expected an error, got: {response}")) + // The fixture's AWS link id; it starts `provisioned`. Replacing its + // config resets the observed status to `pending`. + let aws_id = "0000000000000a01"; + let updated: serde_json::Value = server + .graphql( + &update_link_mutation(aws_id, NEW_AWS_INPUT), + Some(&alice_token), + ) + .await; + let link = &updated["data"]["updateDataPlanePrivateLink"]; + assert_eq!(link["id"], aws_id, "got: {updated}"); + assert_eq!(link["status"], "PENDING"); + assert_eq!( + link["config"]["serviceName"], + "com.amazonaws.vpce.us-east-1.vpce-svc-new999" + ); + + // Editing a link that is already `pending` is allowed: it replaces the + // config and stays `pending` for the next converge. The desired-edit + // trigger bumps the link's generation on this write, which is what keeps + // a converge racing the earlier edit from stamping a stale status. + let reupdated: serde_json::Value = server + .graphql( + &update_link_mutation(aws_id, VALID_AWS_INPUT), + Some(&alice_token), + ) + .await; + let link = &reupdated["data"]["updateDataPlanePrivateLink"]; + assert_eq!(link["id"], aws_id, "got: {reupdated}"); + assert_eq!(link["status"], "PENDING"); + assert_eq!( + link["config"]["serviceName"], + "com.amazonaws.vpce.us-east-1.vpce-svc-abc123" + ); + + // Removing a link is allowed in any status, `pending` included; it + // returns the removed id and drops the row. + let removed: serde_json::Value = server + .graphql(&remove_link_mutation(aws_id), Some(&alice_token)) + .await; + assert_eq!( + removed["data"]["removeDataPlanePrivateLink"], aws_id, + "got: {removed}" + ); + assert_eq!(count_links(&pool, dp).await, 2); } #[sqlx::test( @@ -1201,10 +1363,10 @@ mod tests { scripts("data_planes", "alice", "private_links") ) )] - async fn test_update_private_links_authorization(pool: sqlx::PgPool) { + async fn test_private_link_mutation_authorization(pool: sqlx::PgPool) { let _guard = test_server::init(); - // Create a bob who has no grants on the private dp. + // bob has no grants on the private dp. sqlx::query( "INSERT INTO auth.users (id, email) VALUES \ ('22222222-2222-2222-2222-222222222222', 'bob@example.test')", @@ -1213,34 +1375,56 @@ mod tests { .await .unwrap(); - let server = - test_server::TestServer::start(pool.clone(), test_server::snapshot(pool, false).await) - .await; + let server = test_server::TestServer::start( + pool.clone(), + test_server::snapshot(pool.clone(), false).await, + ) + .await; let alice_token = server.make_access_token(uuid::Uuid::from_bytes([0x11; 16]), None); let bob_token = server.make_access_token(uuid::Uuid::from_bytes([0x22; 16]), Some("bob@example.test")); - let dp = "ops/dp/private/aliceCo/aws-us-east-1-c1"; - // Alice has read on the private dp via the aliceCo/ -> ops/dp/private/aliceCo/ - // role grant installed by the private_links fixture. + // Alice (read + manage_data_plane bundle) can add. let alice_ok: serde_json::Value = server - .graphql(&update_mutation(dp, VALID_AWS_INPUT), Some(&alice_token)) + .graphql(&add_mutation(dp, NEW_AWS_INPUT), Some(&alice_token)) .await; - let echoed = alice_ok["data"]["updateDataPlanePrivateLinks"] - .as_array() - .unwrap_or_else(|| panic!("alice with `read` should succeed: {alice_ok}")); - assert_eq!(echoed.len(), 1); - assert_eq!(echoed[0]["__typename"], "AWSPrivateLink"); + assert_eq!( + alice_ok["data"]["addDataPlanePrivateLink"]["config"]["__typename"], "AWSPrivateLink", + "got: {alice_ok}" + ); - // Bob has no grants and should be rejected. + // Bob is rejected for lacking ModifyDataPlanePrivateNetworking. The + // name-addressed `add` openly names the prefix, because the caller + // supplied the name and so reveals nothing they did not already know. let bob_denied: serde_json::Value = server - .graphql(&update_mutation(dp, VALID_AWS_INPUT), Some(&bob_token)) + .graphql(&add_mutation(dp, NEW_AWS_INPUT), Some(&bob_token)) .await; assert_eq!( first_error_message(&bob_denied), "PermissionDenied: bob@example.test is not authorized to access prefix or name 'ops/dp/private/aliceCo/aws-us-east-1-c1' with required capability ModifyDataPlanePrivateNetworking", ); + + // An id-addressed mutation on a link Bob may not modify must return the + // same "not found" as a missing id, never a permission error that would + // confirm the link (or its data plane) exists. `0000000000000a01` is the + // fixture's existing AWS link. + let aws_id = "0000000000000a01"; + for probe in [ + update_link_mutation(aws_id, NEW_AWS_INPUT), + remove_link_mutation(aws_id), + ] { + let response: serde_json::Value = server.graphql(&probe, Some(&bob_token)).await; + let message = first_error_message(&response); + assert!( + message.contains("not found") && !message.contains("PermissionDenied"), + "expected an existence-hiding not-found error, got: {response}" + ); + } + + // Bob's denied remove did not actually delete: Alice's added link plus + // the three from the fixture remain. + assert_eq!(count_links(&pool, dp).await, 4); } #[sqlx::test( @@ -1250,22 +1434,22 @@ mod tests { scripts("data_planes", "alice", "private_links") ) )] - async fn test_update_private_links_name_validation(pool: sqlx::PgPool) { + async fn test_add_private_link_name_validation(pool: sqlx::PgPool) { let _guard = test_server::init(); let server = test_server::TestServer::start(pool.clone(), test_server::snapshot(pool, false).await) .await; let alice_token = server.make_access_token(uuid::Uuid::from_bytes([0x11; 16]), None); - // Names outside `ops/dp/private//...` are rejected by the - // structural check before any auth or DB work. + // Names outside `ops/dp/private//...` are rejected before any + // auth or DB work. let cases: &[&str] = &[ "ops/dp/public/aws-us-west-2-c1", "ops/dp/private/aws-us-east-1-c1", ]; for name in cases { let response: serde_json::Value = server - .graphql(&update_mutation(name, VALID_AWS_INPUT), Some(&alice_token)) + .graphql(&add_mutation(name, NEW_AWS_INPUT), Some(&alice_token)) .await; assert_eq!( first_error_message(&response), @@ -1274,13 +1458,11 @@ mod tests { ); } - // A structurally-valid name that alice is authorized for (the - // fixture's aliceCo/ -> ops/dp/private/aliceCo/ role grant covers any - // sub-prefix) but which matches no data_planes row: the UPDATE - // affects zero rows and reports not-found. + // A well-formed name alice is authorized for but with no matching + // data_planes row reports not-found. let response: serde_json::Value = server .graphql( - &update_mutation("ops/dp/private/aliceCo/aws-us-east-2-c9", VALID_AWS_INPUT), + &add_mutation("ops/dp/private/aliceCo/aws-us-east-2-c9", NEW_AWS_INPUT), Some(&alice_token), ) .await; diff --git a/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__data_planes__tests__data_planes_malformed_private_link.snap b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__data_planes__tests__data_planes_malformed_private_link.snap index 7ab623c0ddc..14a287dc7d3 100644 --- a/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__data_planes__tests__data_planes_malformed_private_link.snap +++ b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__data_planes__tests__data_planes_malformed_private_link.snap @@ -12,7 +12,7 @@ expression: response "line": 7 } ], - "message": "failed to parse private_links[0] for data plane ops/dp/private/aliceCo/aws-us-east-1-c1: data did not match any variant of untagged enum PrivateLink", + "message": "failed to fetch data plane private links: error occurred while decoding column 2: data did not match any variant of untagged enum PrivateLink", "path": [ "dataPlanes", "edges", diff --git a/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__data_planes__tests__data_planes_with_private_links.snap b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__data_planes__tests__data_planes_with_private_links.snap index b13edcd151f..13eb383e893 100644 --- a/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__data_planes__tests__data_planes_with_private_links.snap +++ b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__data_planes__tests__data_planes_with_private_links.snap @@ -10,8 +10,13 @@ expression: response "node": { "awsLinkEndpoints": [ { - "endpoint_id": "vpce-0123456789abcdef0", - "state": "available" + "dns_entries": [ + { + "dns_name": "vpce-0123abc.vpce-svc-abc123.us-east-1.vpce.amazonaws.com", + "hosted_zone_id": "Z7HUB22EVRPK5" + } + ], + "service_name": "com.amazonaws.vpce.us-east-1.vpce-svc-abc123" } ], "azureLinkEndpoints": [], @@ -19,31 +24,54 @@ expression: response "name": "ops/dp/private/aliceCo/aws-us-east-1-c1", "privateLinks": [ { - "__typename": "AWSPrivateLink", - "azIds": [ - "use1-az1", - "use1-az2" - ], - "region": "us-east-1", - "serviceName": "com.amazonaws.vpce.us-east-1.vpce-svc-abc123" + "config": { + "__typename": "AWSPrivateLink", + "azIds": [ + "use1-az1", + "use1-az2" + ], + "region": "us-east-1", + "serviceName": "com.amazonaws.vpce.us-east-1.vpce-svc-abc123" + }, + "details": { + "dns_entries": [ + { + "dns_name": "vpce-0123abc.vpce-svc-abc123.us-east-1.vpce.amazonaws.com", + "hosted_zone_id": "Z7HUB22EVRPK5" + } + ], + "service_name": "com.amazonaws.vpce.us-east-1.vpce-svc-abc123" + }, + "id": "0000000000000a01", + "status": "PROVISIONED" }, { - "__typename": "AzurePrivateLink", - "dnsName": "privatelink.database.windows.net", - "location": "eastus", - "resourceType": null, - "serviceName": "/subscriptions/x/resourceGroups/rg/providers/Microsoft.Network/privateLinkServices/svc" + "config": { + "__typename": "AzurePrivateLink", + "dnsName": "privatelink.database.windows.net", + "location": "eastus", + "resourceType": null, + "serviceName": "/subscriptions/x/resourceGroups/rg/providers/Microsoft.Network/privateLinkServices/svc" + }, + "details": null, + "id": "0000000000000a02", + "status": "PENDING" }, { - "__typename": "GCPPrivateServiceConnect", - "allPorts": true, - "dnsRecordNames": [ - "r1", - "r2" - ], - "dnsZoneName": "z", - "region": "us-central1", - "serviceAttachment": "projects/p/regions/us-central1/serviceAttachments/sa" + "config": { + "__typename": "GCPPrivateServiceConnect", + "allPorts": true, + "dnsRecordNames": [ + "r1", + "r2" + ], + "dnsZoneName": "z", + "region": "us-central1", + "serviceAttachment": "projects/p/regions/us-central1/serviceAttachments/sa" + }, + "details": null, + "id": "0000000000000a03", + "status": "PENDING" } ] } diff --git a/crates/data-plane-controller/src/job/executor.rs b/crates/data-plane-controller/src/job/executor.rs index afdc4688a47..9e8be7e7dba 100644 --- a/crates/data-plane-controller/src/job/executor.rs +++ b/crates/data-plane-controller/src/job/executor.rs @@ -35,6 +35,12 @@ pub struct Outcome { pub publish_stack: Option, // KMS key used to encrypt HMAC keys pub kms_key: String, + // Private links pinned by (id, generation) at the `PulumiUp1` poll of this + // converge. The post-converge status write only lands on rows whose + // generation still matches, so a link edited mid-converge is skipped and + // settled by the follow-up converge queued when the per-poll refresh + // observes the edit. + pub pinned_links: Vec, } /// Type-erased function for dispatching work execution. @@ -186,11 +192,26 @@ impl Executor { let sleep = match state_ref.status { Status::Idle => self.on_idle(state_ref, inbox, releases, row_state).await?, status => { - // Refresh private_links from the current DB row on every poll, - // so that retries pick up changes made to the data_planes table. + // Refresh desired links on every poll. Generation comparison is + // required in addition to config comparison: A -> B -> A between + // polls has the same final config but still invalidates the + // observation and requires another converge. + let links_changed = state_ref.stack.config.model.private_links + != row_state.stack.config.model.private_links + || state_ref.pinned_links != row_state.pinned_links; state_ref.stack.config.model.private_links = row_state.stack.config.model.private_links; + if matches!(status, Status::PulumiUp1) { + // PulumiUp1 applies the row state read by this poll, so pin + // that exact version without queuing a redundant follow-up. + state_ref.pinned_links = row_state.pinned_links; + } else if links_changed { + // Later polls must retain the PulumiUp1 pins for attributing + // its exports, while remembering to converge the newer rows. + state_ref.pending_converge = true; + } + // For all non-Idle statuses, dispatch to service worker. let action = Action::from_status(status).context("cannot convert status to action")?; @@ -214,6 +235,7 @@ impl Executor { publish_exports: state_ref.publish_exports.take(), publish_stack, kms_key: self.controller_config.secrets_provider.clone(), + pinned_links: state_ref.pinned_links.clone(), }) } @@ -286,13 +308,16 @@ impl Executor { if state.last_pulumi_up + CONVERGE_INTERVAL < chrono::Utc::now() { state.pending_converge = true; } - // Changes to branch or stack configuration require a convergence pass. + // Changes to branch, stack configuration, or private-link generations + // require a convergence pass. Generation comparison catches edits that + // return to the same final config between polls. if state.deploy_branch != next.deploy_branch { state.deploy_branch = next.deploy_branch; state.pending_converge = true; } - if state.stack.config != next.stack.config { + if state.stack.config != next.stack.config || state.pinned_links != next.pinned_links { state.stack.config = next.stack.config; + state.pinned_links = next.pinned_links; state.pending_converge = true; } @@ -393,7 +418,6 @@ async fn fetch_row_state( logs_token, data_plane_name, data_plane_fqdn, - private_links AS "private_links: Vec>", pulumi_key AS "pulumi_key", pulumi_stack AS "pulumi_stack!" FROM data_planes @@ -411,7 +435,38 @@ async fn fetch_row_state( config.model.name = Some(row.data_plane_name); config.model.fqdn = Some(row.data_plane_fqdn); - config.model.private_links = row.private_links.into_iter().map(|link| link.0).collect(); + // Desired links come from `data_plane_private_links`. Each link's + // (id, generation) is pinned so the post-converge status write attributes + // link results to the exact configuration version this converge applied. + let link_rows = sqlx::query!( + r#" + SELECT + id AS "id: models::Id", + generation, + config AS "config!: sqlx::types::Json" + FROM internal.data_plane_private_links + WHERE data_plane_id = $1 + ORDER BY created_at, id + "#, + row.data_plane_id as models::Id, + ) + .fetch_all(pool) + .await + .context("failed to fetch data-plane private links")?; + + let mut private_links = Vec::with_capacity(link_rows.len()); + let mut pinned_links = Vec::with_capacity(link_rows.len()); + for link in link_rows { + pinned_links.push(stack::PinnedLink { + id: link.id, + generation: link.generation, + }); + private_links.push(stack::PrivateLinkEntry { + id: Some(link.id), + config: link.config.0, + }); + } + config.model.private_links = private_links; let stack = if let Some(key) = row.pulumi_key { stack::PulumiStack { @@ -442,6 +497,7 @@ async fn fetch_row_state( preview_branch: String::new(), pending_refresh: false, pending_converge: false, + pinned_links, publish_exports: None, publish_stack: None, }) @@ -545,6 +601,10 @@ impl automations::Outcome for Outcome { // These fields are already implied by other row columns. model.name = None; model.fqdn = None; + // Never read back: `fetch_row_state` unconditionally overwrites the + // model's links from `data_plane_private_links` rows, so persisting + // them here would only store a stale copy that drifts from the table. + model.private_links = Vec::new(); _ = sqlx::query!( r#" @@ -575,6 +635,7 @@ impl automations::Outcome for Outcome { bastion_tunnel_private_key, azure_application_name, azure_application_client_id, + link_results, dekaf_address, dekaf_registry_address, }) = self.publish_exports @@ -618,13 +679,284 @@ impl automations::Outcome for Outcome { .execute(&mut *txn) .await .context("failed to publish exports into data_planes row")?; + + write_private_link_statuses( + &mut *txn, + self.data_plane_id, + &self.pinned_links, + link_results.as_deref().unwrap_or_default(), + ) + .await?; } Ok(automations::Action::Sleep(self.sleep)) } } +/// Records each private link's observed status after a converge from the +/// per-link results est-dry-dock exports, matched to the links this converge +/// pinned by the control-plane id each result echoes back. +/// +/// A pinned link with no matching result is left untouched: results without an +/// id (a converge which pre-dated sending ids, such as the handoff converge of +/// a rolling controller deploy) cannot be attributed, and a link absent from +/// the results entirely was not observed. In both cases the row keeps its +/// prior status until a converge which observes it. +/// +/// Only rows whose generation still matches the value pinned when this converge +/// read its desired links are updated. A link edited mid-converge has a bumped +/// generation, so it is skipped here (this converge did not provision its +/// current config) and is settled by the follow-up converge queued when the +/// per-poll refresh observed the edit. +async fn write_private_link_statuses( + conn: &mut sqlx::PgConnection, + data_plane_id: models::Id, + pinned_links: &[stack::PinnedLink], + link_results: &[stack::LinkResult], +) -> anyhow::Result<()> { + let pinned_ids: Vec = pinned_links.iter().map(|l| l.id).collect(); + let pinned_generations: Vec = pinned_links.iter().map(|l| l.generation).collect(); + + let mut result_ids: Vec = Vec::new(); + let mut result_statuses: Vec = Vec::new(); + let mut result_errors: Vec> = Vec::new(); + let mut result_details: Vec> = Vec::new(); + + for result in link_results { + let Some(id) = result.id else { continue }; + + // Guard the status column's check constraint: a status this controller + // doesn't know is skipped (no observation) rather than failing the + // write, so a newer est-dry-dock cannot wedge an older controller. + if !matches!(result.status.as_str(), "provisioned" | "failed") { + tracing::warn!( + link_id = %id, + status = %result.status, + "skipping link result with unexpected status" + ); + continue; + } + result_ids.push(id); + result_statuses.push(result.status.clone()); + result_errors.push(result.error.clone()); + result_details.push(result.details.clone()); + } + + sqlx::query!( + r#" + WITH pinned AS ( + SELECT id, generation + FROM unnest($2::flowid[], $3::bigint[]) AS p(id, generation) + ), + results AS ( + SELECT id, status, error, detail + FROM unnest($4::flowid[], $5::text[], $6::text[], $7::jsonb[]) + AS r(id, status, error, detail) + ) + UPDATE internal.data_plane_private_links l SET + status = r.status, + details = r.detail, + error = r.error, + observed_at = now(), + updated_at = now() + FROM internal.data_plane_private_links l2 + JOIN pinned p ON p.id = l2.id AND p.generation = l2.generation + JOIN results r ON r.id = l2.id + WHERE l.id = l2.id + AND l2.data_plane_id = $1 + "#, + data_plane_id as models::Id, + pinned_ids as Vec, + pinned_generations as Vec, + result_ids as Vec, + &result_statuses, + result_errors as Vec>, + result_details as Vec>, + ) + .execute(&mut *conn) + .await + .context("failed to update private link statuses")?; + + Ok(()) +} + const IDLE_INTERVAL: std::time::Duration = std::time::Duration::from_secs(60); const POLL_AGAIN: std::time::Duration = std::time::Duration::ZERO; const REFRESH_INTERVAL: std::time::Duration = std::time::Duration::from_secs(2 * 60 * 60); const CONVERGE_INTERVAL: std::time::Duration = std::time::Duration::from_secs(4 * 60 * 60); + +#[cfg(test)] +mod tests { + use super::stack; + use super::stack::PinnedLink; + use super::write_private_link_statuses; + + async fn link_id(pool: &sqlx::PgPool, identity: &str) -> models::Id { + sqlx::query_scalar!( + r#"SELECT id as "id: models::Id" + FROM internal.data_plane_private_links WHERE service_identity = $1"#, + identity, + ) + .fetch_one(pool) + .await + .unwrap() + } + + async fn status_of(pool: &sqlx::PgPool, identity: &str) -> (String, Option) { + let row = sqlx::query!( + r#" + SELECT status, details as "details: sqlx::types::Json" + FROM internal.data_plane_private_links WHERE service_identity = $1 + "#, + identity, + ) + .fetch_one(pool) + .await + .unwrap(); + (row.status, row.details.map(|d| d.0)) + } + + async fn error_of(pool: &sqlx::PgPool, identity: &str) -> Option { + sqlx::query_scalar!( + r#"SELECT error + FROM internal.data_plane_private_links WHERE service_identity = $1"#, + identity, + ) + .fetch_one(pool) + .await + .unwrap() + } + + // Matches this converge's exported link results to the pinned links by id + // and skips any link whose generation no longer matches the pinned value. + // Covers a provisioned result (details recorded), a failed result (error + // recorded), no-observation for a pinned link absent from the results, + // skipped entries (a null id and an unknown status), and the generation + // guard (a row edited after the pin is skipped, and pinning its current + // generation processes it). + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "fixtures", scripts("private_link_statuses")) + )] + async fn write_private_link_statuses_records_and_guards_generation(pool: sqlx::PgPool) { + let data_plane_id: models::Id = sqlx::query_scalar!( + r#"SELECT id as "id: models::Id" FROM data_planes WHERE data_plane_name = $1"#, + "ops/dp/private/testCo/aws-1", + ) + .fetch_one(&pool) + .await + .unwrap(); + + // This converge reported svc-a provisioned, svc-g failed, and svc-edited + // provisioned; svc-orphan is absent from the results (no observation: + // its only entries are one with a null id, which cannot be attributed, + // and one with a status this controller does not know). + let results = vec![ + stack::LinkResult { + id: Some(link_id(&pool, "svc-a").await), + status: "provisioned".to_string(), + error: None, + details: Some(serde_json::json!({ + "service_name": "svc-a", + "dns_entries": [{"dns_name": "svc-a.example"}] + })), + }, + stack::LinkResult { + id: Some(link_id(&pool, "svc-g").await), + status: "failed".to_string(), + error: Some("Invalid PrivateLink Availability Zone ID".to_string()), + details: None, + }, + stack::LinkResult { + id: Some(link_id(&pool, "svc-edited").await), + status: "provisioned".to_string(), + error: None, + details: Some(serde_json::json!({"service_name": "svc-edited"})), + }, + stack::LinkResult { + id: None, + status: "provisioned".to_string(), + error: None, + details: Some(serde_json::json!({"service_name": "svc-orphan"})), + }, + stack::LinkResult { + id: Some(link_id(&pool, "svc-orphan").await), + status: "certainly-not-a-status".to_string(), + error: None, + details: None, + }, + ]; + + // Pin every link at generation 1. svc-edited is at generation 2 in the + // fixture (an edit landed after this converge read its desired state), so + // its pinned generation no longer matches the row. + let pinned = vec![ + PinnedLink { + id: link_id(&pool, "svc-a").await, + generation: 1, + }, + PinnedLink { + id: link_id(&pool, "svc-orphan").await, + generation: 1, + }, + PinnedLink { + id: link_id(&pool, "svc-edited").await, + generation: 1, + }, + PinnedLink { + id: link_id(&pool, "svc-g").await, + generation: 1, + }, + ]; + + let mut conn = pool.acquire().await.unwrap(); + write_private_link_statuses(&mut conn, data_plane_id, &pinned, &results) + .await + .unwrap(); + + // A provisioned result -> provisioned with its details. + let (status, details) = status_of(&pool, "svc-a").await; + assert_eq!(status, "provisioned"); + assert_eq!(details.unwrap()["service_name"], "svc-a"); + + // A failed result -> failed with its error, details cleared. + let (status, details) = status_of(&pool, "svc-g").await; + assert_eq!(status, "failed"); + assert!(details.is_none()); + assert_eq!( + error_of(&pool, "svc-g").await.as_deref(), + Some("Invalid PrivateLink Availability Zone ID"), + ); + + // Absent from the results -> no observation: prior status and details + // are retained. + let (status, details) = status_of(&pool, "svc-orphan").await; + assert_eq!(status, "provisioned"); + assert_eq!(details.unwrap()["service_name"], "svc-orphan"); + + // The pinned generation no longer matches the row, so its result is not + // applied and the row keeps its pre-edit observation. + let (status, details) = status_of(&pool, "svc-edited").await; + assert_eq!(status, "provisioned"); + assert_eq!(details.unwrap()["stale"], true); + + // Pinning svc-edited at its current generation (2) lets a later + // converge apply its result, refreshing status and details. + let results = vec![stack::LinkResult { + id: Some(link_id(&pool, "svc-edited").await), + status: "provisioned".to_string(), + error: None, + details: Some(serde_json::json!({"service_name": "svc-edited", "fresh": true})), + }]; + let pinned = vec![PinnedLink { + id: link_id(&pool, "svc-edited").await, + generation: 2, + }]; + write_private_link_statuses(&mut conn, data_plane_id, &pinned, &results) + .await + .unwrap(); + let (status, details) = status_of(&pool, "svc-edited").await; + assert_eq!(status, "provisioned"); + assert_eq!(details.unwrap()["fresh"], true); + } +} diff --git a/crates/data-plane-controller/src/job/fixtures/private_link_statuses.sql b/crates/data-plane-controller/src/job/fixtures/private_link_statuses.sql new file mode 100644 index 00000000000..a9d9f0ebc13 --- /dev/null +++ b/crates/data-plane-controller/src/job/fixtures/private_link_statuses.sql @@ -0,0 +1,78 @@ +-- Seeds a private data plane and four private links whose statuses and +-- generations exercise every branch of `write_private_link_statuses`: +-- svc-a aws, pending, gen 1 -> a provisioned result is recorded +-- svc-orphan aws, provisioned, gen 1 -> absent from the results, so the row +-- is untouched (no observation) +-- svc-edited aws, provisioned, gen 2 -> pinned at gen 1, so generation no +-- longer matches and the row is skipped +-- svc-g gcp, pending, gen 1 -> a failed result is recorded +-- Rows are inserted with explicit generations; the desired-edit trigger fires +-- only on update, so these inserts do not perturb them. +insert into public.data_planes ( + id, + data_plane_name, + data_plane_fqdn, + hmac_keys, + encrypted_hmac_keys, + broker_address, + reactor_address, + ops_logs_name, + ops_stats_name, + ops_l1_events_name, + ops_l1_inferred_name, + ops_l1_stats_name, + ops_l2_events_transform, + ops_l2_inferred_transform, + ops_l2_stats_transform, + enable_l2, + cidr_blocks, + aws_iam_user_arn, + gcp_service_account_email, + azure_application_name, + azure_application_client_id +) values ( + '555555555555', + 'ops/dp/private/testCo/aws-1', + 'dp.private.testCo', + '{c2VjcmV0}', + '{}', + 'broker.dp.private.testCo', + 'reactor.dp.private.testCo', + 'ops/tasks/private/testCo/logs', + 'ops/tasks/private/testCo/stats', + 'ops/rollups/L1/private/testCo/events', + 'ops/rollups/L1/private/testCo/inferred', + 'ops/rollups/L1/private/testCo/stats', + 'from.dp.private.testCo', + 'from.dp.private.testCo', + 'from.dp.private.testCo', + false, + '{10.20.0.0/16}', + 'arn:aws:iam::444555666:user/test', + 'test-gcp-private@estuary-test.iam.gserviceaccount.com', + 'estuary-test-app-private', + '55555555-5555-5555-5555-555555555555' +); + +insert into internal.data_plane_private_links + (id, data_plane_id, config, status, details, generation) values + ( + '00:00:00:00:00:00:0b:01', '555555555555', + '{"region":"us-east-1","az_ids":["a"],"service_name":"svc-a"}'::jsonb, + 'pending', null, 1 + ), + ( + '00:00:00:00:00:00:0b:02', '555555555555', + '{"region":"us-east-1","az_ids":["a"],"service_name":"svc-edited"}'::jsonb, + 'provisioned', '{"service_name":"svc-edited","stale":true}'::jsonb, 2 + ), + ( + '00:00:00:00:00:00:0b:04', '555555555555', + '{"service_attachment":"svc-g","region":"r","dns_zone_name":"z","dns_record_names":["n"]}'::jsonb, + 'pending', null, 1 + ), + ( + '00:00:00:00:00:00:0b:05', '555555555555', + '{"region":"us-east-1","az_ids":["a"],"service_name":"svc-orphan"}'::jsonb, + 'provisioned', '{"service_name":"svc-orphan"}'::jsonb, 1 + ); diff --git a/crates/data-plane-controller/src/service/worker.rs b/crates/data-plane-controller/src/service/worker.rs index b184ae33322..f3f00033912 100644 --- a/crates/data-plane-controller/src/service/worker.rs +++ b/crates/data-plane-controller/src/service/worker.rs @@ -285,7 +285,7 @@ impl Worker { &self, state: &mut stack::State, checkouts: &mut HashMap, - private_links: Vec, + private_links: Vec, ) -> anyhow::Result { state.stack.config.model.private_links = private_links; diff --git a/crates/data-plane-controller/src/shared/dry_run_fixture.json b/crates/data-plane-controller/src/shared/dry_run_fixture.json index 58ce50c87b5..f4743575c73 100644 --- a/crates/data-plane-controller/src/shared/dry_run_fixture.json +++ b/crates/data-plane-controller/src/shared/dry_run_fixture.json @@ -36,7 +36,36 @@ "ip_address": "10.128.0.100", "region": "us-central1", "dns_zone_name": "estuary-psc", - "dns_record_names": ["customer-db", "customer-api"] + "dns_record_names": [ + "customer-db", + "customer-api" + ] + } + ], + "link_results": [ + { + "id": "0000000000000b01", + "status": "provisioned", + "error": null, + "details": { + "service_name": "com.amazonaws.vpce.us-east-1.vpce-svc-foo", + "dns_entries": [ + { + "dns_name": "vpce-123-abc.vpce-svc-foo.us-east-1.vpce.amazonaws.com", + "hosted_zone_id": "ZZ1" + }, + { + "dns_name": "vpce-123-abc-us-east-1a.vpce-svc-foo.us-east-1.vpce.amazonaws.com", + "hosted_zone_id": "ZZ1" + } + ] + } + }, + { + "id": "0000000000000b02", + "status": "failed", + "error": "Invalid PrivateLink Availability Zone ID 'use1-az9' (valid values are ['use1-az1', 'use1-az2'])", + "details": null } ], "bastion_tunnel_private_key": "bastion_tunnel_private_key fixture", diff --git a/crates/data-plane-controller/src/shared/stack.rs b/crates/data-plane-controller/src/shared/stack.rs index c4ff8bcb81b..a9642ebb41f 100644 --- a/crates/data-plane-controller/src/shared/stack.rs +++ b/crates/data-plane-controller/src/shared/stack.rs @@ -28,6 +28,16 @@ pub enum Status { AwaitDNS2, } +/// A private link's identity and desired-config version, pinned when a converge +/// reads its desired state. The controller's post-converge status write matches +/// on both, so a link edited mid-converge (which bumps its generation) is left +/// for a follow-up converge rather than stamped with a stale status. +#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct PinnedLink { + pub id: models::Id, + pub generation: i64, +} + #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] pub struct State { // DataPlane which this controller manages. @@ -65,6 +75,13 @@ pub struct State { #[serde(default, skip_serializing_if = "is_false")] pub pending_converge: bool, + // Last-observed private-link (id, generation) pairs. At PulumiUp1 these are + // pinned for the current converge and retained through its status write, so + // a link edited mid-converge is skipped and settled by a follow-up. While + // idle they are the generation baseline for detecting A -> B -> A edits. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub pinned_links: Vec, + // When Some, updated Pulumi stack exports to be written back into the `data_planes` row. #[serde(default, skip_serializing_if = "Option::is_none")] pub publish_exports: Option, @@ -114,7 +131,7 @@ pub struct DataPlane { #[serde(default, skip_serializing_if = "Vec::is_empty")] pub allow_cidrs: Vec, #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub private_links: Vec, + pub private_links: Vec, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub extra_private_routes: Vec, // When true, the host-local podman `flow-connectors` network is created @@ -179,6 +196,20 @@ pub struct GCPBYOC { // references them through this module, so re-export them here. pub use models::{AWSPrivateLink, AzurePrivateLink, GCPPrivateServiceConnect, PrivateLink}; +/// An element of the stack model's `private_links`: the link's user-owned +/// configuration plus the id of its `data_plane_private_links` row, serialized +/// as a sibling `id` field of the flattened config. est-dry-dock treats the id +/// as opaque and echoes it back in its per-link `link_results` export, which is +/// how each result is attributed to its link's row. +/// `id` is None only for stack configs persisted before ids were sent. +#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct PrivateLinkEntry { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub id: Option, + #[serde(flatten)] + pub config: PrivateLink, +} + #[derive(Clone, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)] #[serde(rename_all = "lowercase")] pub enum Role { @@ -309,12 +340,16 @@ pub struct ControlExports { pub azure_link_endpoints: Vec, pub bastion_tunnel_private_key: Option, pub cidr_blocks: Vec, - // TODO(whb): Remove #[serde(default)] once est-dry-dock is deployed with gcp_psc_endpoints output. - #[serde(default)] pub gcp_psc_endpoints: Vec, pub gcp_service_account_email: String, pub hmac_keys: Vec, pub ssh_key: String, + // Per-link provisioning results, one entry per model `private_links` entry, + // addressed by the control-plane link id each entry echoes back. + // None for stack outputs produced by an est-dry-dock which pre-dates the + // export, in which case no per-link statuses are written. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub link_results: Option>, // Computed by the controller based on Dekaf deployment status. // Not exported by Pulumi, hence the default. #[serde(default, skip_serializing_if = "Option::is_none")] @@ -323,6 +358,28 @@ pub struct ControlExports { pub dekaf_registry_address: Option, } +/// A per-link provisioning result from est-dry-dock's `control.link_results` +/// export, addressed by the control-plane link id it echoes back. +#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] +pub struct LinkResult { + /// Id of the `data_plane_private_links` row this result addresses. + /// None when the model entry carried no id (outputs of a converge which + /// pre-dates sending ids); such entries cannot be attributed and are + /// skipped by the status write. + #[serde(default)] + pub id: Option, + /// `provisioned` or `failed`. `failed` is a per-link input validation + /// error caught by the Pulumi program, and is authoritative. `provisioned` + /// means the link's resources were registered and the program completed; + /// it is trusted because exports are only read after a successful + /// `pulumi up`. + pub status: String, + #[serde(default)] + pub error: Option, + #[serde(default)] + pub details: Option, +} + #[derive(Debug, serde::Deserialize, serde::Serialize)] pub struct PulumiStackResourceChanges { #[serde(default)] @@ -593,12 +650,15 @@ mod test { serde_json::from_value::(fixtures.get("aws_private_link").unwrap().clone()) .unwrap() .private_links[0], - PrivateLink::AWS(AWSPrivateLink { - az_ids: vec!["a".to_string(), "b".to_string()], - region: "us-west-2".to_string(), - service_name: "service".to_string(), - service_region: None, - }), + PrivateLinkEntry { + id: None, + config: PrivateLink::AWS(AWSPrivateLink { + az_ids: vec!["a".to_string(), "b".to_string()], + region: "us-west-2".to_string(), + service_name: "service".to_string(), + service_region: None, + }), + }, ); let azure_parsed = serde_json::from_value::( @@ -607,12 +667,15 @@ mod test { .unwrap(); assert_eq!( azure_parsed.private_links[0], - PrivateLink::Azure(AzurePrivateLink { - location: "eastus".to_string(), - service_name: "service".to_string(), - resource_type: Some("managedInstance".to_string()), - dns_name: Some("privatelink.database.windows.net".to_string()), - }), + PrivateLinkEntry { + id: None, + config: PrivateLink::Azure(AzurePrivateLink { + location: "eastus".to_string(), + service_name: "service".to_string(), + resource_type: Some("managedInstance".to_string()), + dns_name: Some("privatelink.database.windows.net".to_string()), + }), + }, ); assert_eq!( azure_parsed.azure_byoc, @@ -645,15 +708,18 @@ mod test { .unwrap(); assert_eq!( gcp_psc_parsed.private_links[0], - PrivateLink::GCP(GCPPrivateServiceConnect { - service_attachment: - "projects/customer-project/regions/us-central1/serviceAttachments/customer-db" - .to_string(), - region: "us-central1".to_string(), - dns_zone_name: "estuary-psc".to_string(), - dns_record_names: vec!["customer-db".to_string(), "customer-api".to_string()], - all_ports: true, - }), + PrivateLinkEntry { + id: None, + config: PrivateLink::GCP(GCPPrivateServiceConnect { + service_attachment: + "projects/customer-project/regions/us-central1/serviceAttachments/customer-db" + .to_string(), + region: "us-central1".to_string(), + dns_zone_name: "estuary-psc".to_string(), + dns_record_names: vec!["customer-db".to_string(), "customer-api".to_string()], + all_ports: true, + }), + }, ); assert_eq!( gcp_psc_parsed.gcp_byoc, @@ -663,6 +729,78 @@ mod test { ); } + // The wire shape est-dry-dock parses: `id` is a sibling of the flattened + // link config, omitted when None, and round-trips losslessly. + #[test] + fn private_link_entry_round_trip() { + let with_id = PrivateLinkEntry { + id: Some(models::Id::new([0, 0, 0, 0, 0, 0, 0xb, 0x1])), + config: PrivateLink::AWS(AWSPrivateLink { + az_ids: vec!["use1-az1".to_string()], + region: "us-east-1".to_string(), + service_name: "com.amazonaws.vpce.us-east-1.vpce-svc-abc".to_string(), + service_region: None, + }), + }; + let raw = serde_json::to_string(&with_id).unwrap(); + assert_eq!( + raw, + r#"{"id":"0000000000000b01","region":"us-east-1","az_ids":["use1-az1"],"service_name":"com.amazonaws.vpce.us-east-1.vpce-svc-abc"}"#, + ); + assert_eq!( + serde_json::from_str::(&raw).unwrap(), + with_id + ); + + // Entries without an id (persisted state which pre-dates ids, or the + // legacy element shape) parse with id None and serialize without it. + let without_id = r#"{"service_name":"svc","location":"eastus"}"#; + let parsed: PrivateLinkEntry = serde_json::from_str(without_id).unwrap(); + assert_eq!(parsed.id, None); + assert!(matches!(parsed.config, PrivateLink::Azure(_))); + assert_eq!(serde_json::to_string(&parsed).unwrap(), without_id); + } + + // Stack outputs from an est-dry-dock which pre-dates the `link_results` + // export parse with None rather than erroring, so a mixed-version fleet + // cannot fail exports parsing; per-entry `id`/`error`/`details` are + // likewise optional. + #[test] + fn control_exports_link_results_are_optional() { + let base = serde_json::json!({ + "aws_iam_user_arn": "arn:aws:iam::123456:user/data-planes/data-plane-abcd", + "aws_link_endpoints": [], + "azure_application_client_id": "12345678-1234-1234-1234-123456789abc", + "azure_application_name": "data-plane-123.dp.estuary-data.com", + "azure_link_endpoints": [], + "bastion_tunnel_private_key": null, + "cidr_blocks": [], + "gcp_psc_endpoints": [], + "gcp_service_account_email": "data-plane-abcd@project.iam.gserviceaccount.com", + "hmac_keys": [], + "ssh_key": "ssh_key fixture" + }); + + let parsed: ControlExports = serde_json::from_value(base.clone()).unwrap(); + assert_eq!(parsed.link_results.map(|r| r.len()), None); + + let mut with_null = base.clone(); + with_null["link_results"] = serde_json::Value::Null; + let parsed: ControlExports = serde_json::from_value(with_null).unwrap(); + assert_eq!(parsed.link_results.map(|r| r.len()), None); + + // A minimal entry: only `status` is required. + let mut with_minimal = base; + with_minimal["link_results"] = serde_json::json!([{"status": "provisioned"}]); + let parsed: ControlExports = serde_json::from_value(with_minimal).unwrap(); + let results = parsed.link_results.unwrap(); + assert_eq!(results.len(), 1); + assert_eq!(results[0].id, None); + assert_eq!(results[0].status, "provisioned"); + assert_eq!(results[0].error, None); + assert!(results[0].details.is_none()); + } + #[test] fn transition_checks() { let last: State = serde_json::from_str(include_str!("state_fixture.json")).unwrap(); diff --git a/crates/data-plane-controller/tests/snapshots/test_initial_convergence_and_release__test.snap b/crates/data-plane-controller/tests/snapshots/test_initial_convergence_and_release__test.snap index 478c4339f37..b8d975e108c 100644 --- a/crates/data-plane-controller/tests/snapshots/test_initial_convergence_and_release__test.snap +++ b/crates/data-plane-controller/tests/snapshots/test_initial_convergence_and_release__test.snap @@ -326,6 +326,32 @@ expression: trace.lock().unwrap().as_slice() "a2V5Mw==" ], "ssh_key": "", + "link_results": [ + { + "id": "0000000000000b01", + "status": "provisioned", + "error": null, + "details": { + "dns_entries": [ + { + "dns_name": "vpce-123-abc.vpce-svc-foo.us-east-1.vpce.amazonaws.com", + "hosted_zone_id": "ZZ1" + }, + { + "dns_name": "vpce-123-abc-us-east-1a.vpce-svc-foo.us-east-1.vpce.amazonaws.com", + "hosted_zone_id": "ZZ1" + } + ], + "service_name": "com.amazonaws.vpce.us-east-1.vpce-svc-foo" + } + }, + { + "id": "0000000000000b02", + "status": "failed", + "error": "Invalid PrivateLink Availability Zone ID 'use1-az9' (valid values are ['use1-az1', 'use1-az2'])", + "details": null + } + ], "dekaf_address": "tls://dekaf.test-plane.estuary.dev:9092", "dekaf_registry_address": "https://dekaf.test-plane.estuary.dev:443" } @@ -733,6 +759,32 @@ expression: trace.lock().unwrap().as_slice() "a2V5Mw==" ], "ssh_key": "", + "link_results": [ + { + "id": "0000000000000b01", + "status": "provisioned", + "error": null, + "details": { + "dns_entries": [ + { + "dns_name": "vpce-123-abc.vpce-svc-foo.us-east-1.vpce.amazonaws.com", + "hosted_zone_id": "ZZ1" + }, + { + "dns_name": "vpce-123-abc-us-east-1a.vpce-svc-foo.us-east-1.vpce.amazonaws.com", + "hosted_zone_id": "ZZ1" + } + ], + "service_name": "com.amazonaws.vpce.us-east-1.vpce-svc-foo" + } + }, + { + "id": "0000000000000b02", + "status": "failed", + "error": "Invalid PrivateLink Availability Zone ID 'use1-az9' (valid values are ['use1-az1', 'use1-az2'])", + "details": null + } + ], "dekaf_address": "tls://dekaf.test-plane.estuary.dev:9092", "dekaf_registry_address": "https://dekaf.test-plane.estuary.dev:443" } @@ -1161,6 +1213,32 @@ expression: trace.lock().unwrap().as_slice() "a2V5Mw==" ], "ssh_key": "", + "link_results": [ + { + "id": "0000000000000b01", + "status": "provisioned", + "error": null, + "details": { + "dns_entries": [ + { + "dns_name": "vpce-123-abc.vpce-svc-foo.us-east-1.vpce.amazonaws.com", + "hosted_zone_id": "ZZ1" + }, + { + "dns_name": "vpce-123-abc-us-east-1a.vpce-svc-foo.us-east-1.vpce.amazonaws.com", + "hosted_zone_id": "ZZ1" + } + ], + "service_name": "com.amazonaws.vpce.us-east-1.vpce-svc-foo" + } + }, + { + "id": "0000000000000b02", + "status": "failed", + "error": "Invalid PrivateLink Availability Zone ID 'use1-az9' (valid values are ['use1-az1', 'use1-az2'])", + "details": null + } + ], "dekaf_address": "tls://dekaf.test-plane.estuary.dev:9092", "dekaf_registry_address": "https://dekaf.test-plane.estuary.dev:443" } @@ -1581,6 +1659,32 @@ expression: trace.lock().unwrap().as_slice() "a2V5Mw==" ], "ssh_key": "", + "link_results": [ + { + "id": "0000000000000b01", + "status": "provisioned", + "error": null, + "details": { + "dns_entries": [ + { + "dns_name": "vpce-123-abc.vpce-svc-foo.us-east-1.vpce.amazonaws.com", + "hosted_zone_id": "ZZ1" + }, + { + "dns_name": "vpce-123-abc-us-east-1a.vpce-svc-foo.us-east-1.vpce.amazonaws.com", + "hosted_zone_id": "ZZ1" + } + ], + "service_name": "com.amazonaws.vpce.us-east-1.vpce-svc-foo" + } + }, + { + "id": "0000000000000b02", + "status": "failed", + "error": "Invalid PrivateLink Availability Zone ID 'use1-az9' (valid values are ['use1-az1', 'use1-az2'])", + "details": null + } + ], "dekaf_address": "tls://dekaf.test-plane.estuary.dev:9092", "dekaf_registry_address": "https://dekaf.test-plane.estuary.dev:443" } diff --git a/crates/data-plane-controller/tests/snapshots/test_preview__test.snap b/crates/data-plane-controller/tests/snapshots/test_preview__test.snap index b0337f490fc..a51848a4d4f 100644 --- a/crates/data-plane-controller/tests/snapshots/test_preview__test.snap +++ b/crates/data-plane-controller/tests/snapshots/test_preview__test.snap @@ -368,6 +368,32 @@ expression: trace.lock().unwrap().as_slice() "a2V5Mw==" ], "ssh_key": "", + "link_results": [ + { + "id": "0000000000000b01", + "status": "provisioned", + "error": null, + "details": { + "dns_entries": [ + { + "dns_name": "vpce-123-abc.vpce-svc-foo.us-east-1.vpce.amazonaws.com", + "hosted_zone_id": "ZZ1" + }, + { + "dns_name": "vpce-123-abc-us-east-1a.vpce-svc-foo.us-east-1.vpce.amazonaws.com", + "hosted_zone_id": "ZZ1" + } + ], + "service_name": "com.amazonaws.vpce.us-east-1.vpce-svc-foo" + } + }, + { + "id": "0000000000000b02", + "status": "failed", + "error": "Invalid PrivateLink Availability Zone ID 'use1-az9' (valid values are ['use1-az1', 'use1-az2'])", + "details": null + } + ], "dekaf_address": "tls://dekaf.test-plane.estuary.dev:9092", "dekaf_registry_address": "https://dekaf.test-plane.estuary.dev:443" } diff --git a/crates/data-plane-controller/tests/snapshots/test_private_links__private_links.snap b/crates/data-plane-controller/tests/snapshots/test_private_links__private_links.snap index 2750b8e9bfb..5d90538c110 100644 --- a/crates/data-plane-controller/tests/snapshots/test_private_links__private_links.snap +++ b/crates/data-plane-controller/tests/snapshots/test_private_links__private_links.snap @@ -65,6 +65,7 @@ expression: trace.lock().unwrap().as_slice() ], "private_links": [ { + "id": "0000000000000b01", "region": "us-west-2", "az_ids": [ "a", @@ -336,6 +337,32 @@ expression: trace.lock().unwrap().as_slice() "a2V5Mw==" ], "ssh_key": "", + "link_results": [ + { + "id": "0000000000000b01", + "status": "provisioned", + "error": null, + "details": { + "dns_entries": [ + { + "dns_name": "vpce-123-abc.vpce-svc-foo.us-east-1.vpce.amazonaws.com", + "hosted_zone_id": "ZZ1" + }, + { + "dns_name": "vpce-123-abc-us-east-1a.vpce-svc-foo.us-east-1.vpce.amazonaws.com", + "hosted_zone_id": "ZZ1" + } + ], + "service_name": "com.amazonaws.vpce.us-east-1.vpce-svc-foo" + } + }, + { + "id": "0000000000000b02", + "status": "failed", + "error": "Invalid PrivateLink Availability Zone ID 'use1-az9' (valid values are ['use1-az1', 'use1-az2'])", + "details": null + } + ], "dekaf_address": "tls://dekaf.test-plane.estuary.dev:9092", "dekaf_registry_address": "https://dekaf.test-plane.estuary.dev:443" } @@ -417,6 +444,7 @@ expression: trace.lock().unwrap().as_slice() ], "private_links": [ { + "id": "0000000000000b01", "region": "us-west-2", "az_ids": [ "a", @@ -649,6 +677,32 @@ expression: trace.lock().unwrap().as_slice() "a2V5Mw==" ], "ssh_key": "", + "link_results": [ + { + "id": "0000000000000b01", + "status": "provisioned", + "error": null, + "details": { + "dns_entries": [ + { + "dns_name": "vpce-123-abc.vpce-svc-foo.us-east-1.vpce.amazonaws.com", + "hosted_zone_id": "ZZ1" + }, + { + "dns_name": "vpce-123-abc-us-east-1a.vpce-svc-foo.us-east-1.vpce.amazonaws.com", + "hosted_zone_id": "ZZ1" + } + ], + "service_name": "com.amazonaws.vpce.us-east-1.vpce-svc-foo" + } + }, + { + "id": "0000000000000b02", + "status": "failed", + "error": "Invalid PrivateLink Availability Zone ID 'use1-az9' (valid values are ['use1-az1', 'use1-az2'])", + "details": null + } + ], "dekaf_address": "tls://dekaf.test-plane.estuary.dev:9092", "dekaf_registry_address": "https://dekaf.test-plane.estuary.dev:443" } @@ -730,6 +784,7 @@ expression: trace.lock().unwrap().as_slice() ], "private_links": [ { + "id": "0000000000000b01", "region": "us-west-2", "az_ids": [ "a", @@ -738,6 +793,7 @@ expression: trace.lock().unwrap().as_slice() "service_name": "service" }, { + "id": "0000000000000b02", "region": "us-west-2", "az_ids": [ "b", diff --git a/crates/data-plane-controller/tests/test_private_links.rs b/crates/data-plane-controller/tests/test_private_links.rs index f901a02261a..d2c7a70e29f 100644 --- a/crates/data-plane-controller/tests/test_private_links.rs +++ b/crates/data-plane-controller/tests/test_private_links.rs @@ -8,10 +8,7 @@ use std::{ mod util; use util::{TraceEntry, initial_state, mock_dispatch_fn}; -#[tokio::test] -async fn test_private_links() { - let trace = Arc::new(Mutex::new(Vec::new())); - +fn test_executor(trace: Arc>>) -> Executor { let controller_config = ControllerConfig { dns_ttl: std::time::Duration::ZERO, dry_dock_remote: "git@github.com:estuary/est-dry-dock.git".to_string(), @@ -20,19 +17,26 @@ async fn test_private_links() { state_backend: "file:///tmp/pulumi-test-state".parse().unwrap(), dry_run: false, }; + Executor::new_with_dispatch(controller_config, mock_dispatch_fn(trace)) +} - let executor = Executor::new_with_dispatch(controller_config, mock_dispatch_fn(trace.clone())); +#[tokio::test] +async fn test_private_links() { + let trace = Arc::new(Mutex::new(Vec::new())); + let executor = test_executor(trace.clone()); let mut state: Option = None; let mut inbox: VecDeque<(models::Id, Option)> = VecDeque::new(); let mut row_state = initial_state(); - row_state.stack.config.model.private_links = - vec![stack::PrivateLink::AWS(stack::AWSPrivateLink { + row_state.stack.config.model.private_links = vec![stack::PrivateLinkEntry { + id: Some(models::Id::new([0, 0, 0, 0, 0, 0, 0xb, 0x1])), + config: stack::PrivateLink::AWS(stack::AWSPrivateLink { az_ids: vec!["a".to_string(), "b".to_string()], region: "us-west-2".to_string(), service_name: "service".to_string(), service_region: None, - })]; + }), + }]; inbox.push_back(( models::Id::zero(), @@ -73,12 +77,15 @@ async fn test_private_links() { .config .model .private_links - .push(stack::PrivateLink::AWS(stack::AWSPrivateLink { - az_ids: vec!["b".to_string(), "c".to_string()], - region: "us-west-2".to_string(), - service_name: "service-2".to_string(), - service_region: None, - })); + .push(stack::PrivateLinkEntry { + id: Some(models::Id::new([0, 0, 0, 0, 0, 0, 0xb, 0x2])), + config: stack::PrivateLink::AWS(stack::AWSPrivateLink { + az_ids: vec!["b".to_string(), "c".to_string()], + region: "us-west-2".to_string(), + service_name: "service-2".to_string(), + service_region: None, + }), + }); inbox.push_back((models::Id::zero(), Some(Message::Converge))); @@ -112,3 +119,72 @@ async fn test_private_links() { insta::assert_json_snapshot!(trace.lock().unwrap().as_slice()); } + +#[tokio::test] +async fn generation_only_change_queues_converge() { + let link_id = models::Id::new([0, 0, 0, 0, 0, 0, 0xb, 0x1]); + let pinned = |generation| stack::PinnedLink { + id: link_id, + generation, + }; + + // While idle, generations are the change-detection baseline. The plane is + // disabled so the queued converge remains observable instead of starting. + let trace = Arc::new(Mutex::new(Vec::new())); + let executor = test_executor(trace); + let mut idle = initial_state(); + idle.disabled = true; + idle.stack.encrypted_key = "encrypted".to_string(); + idle.last_pulumi_up = chrono::Utc::now(); + idle.last_refresh = chrono::Utc::now(); + idle.pinned_links = vec![pinned(1)]; + let mut row_state = idle.clone(); + row_state.pinned_links = vec![pinned(2)]; + + let mut state = Some(idle); + let mut inbox = VecDeque::new(); + let outcome = executor + .on_poll( + models::Id::new([42; 8]), + &mut state, + &mut inbox, + Vec::new(), + row_state, + ) + .await + .unwrap(); + let state = state.unwrap(); + assert!(matches!(outcome.status, stack::Status::Idle)); + assert!(state.pending_converge); + assert_eq!(state.pinned_links, vec![pinned(2)]); + + // During a converge, the same change queues a follow-up but preserves the + // PulumiUp1 pins needed to reject this converge's now-stale result. + let trace = Arc::new(Mutex::new(Vec::new())); + let executor = test_executor(trace); + let mut active = initial_state(); + active.stack.encrypted_key = "encrypted".to_string(); + active.status = stack::Status::AwaitDNS1; + active.last_pulumi_up = chrono::Utc::now(); + active.pinned_links = vec![pinned(1)]; + let mut row_state = active.clone(); + row_state.status = stack::Status::Idle; + row_state.pinned_links = vec![pinned(2)]; + + let mut state = Some(active); + let mut inbox = VecDeque::new(); + let outcome = executor + .on_poll( + models::Id::new([42; 8]), + &mut state, + &mut inbox, + Vec::new(), + row_state, + ) + .await + .unwrap(); + let state = state.unwrap(); + assert!(matches!(outcome.status, stack::Status::Ansible)); + assert!(state.pending_converge); + assert_eq!(state.pinned_links, vec![pinned(1)]); +} diff --git a/crates/data-plane-controller/tests/util.rs b/crates/data-plane-controller/tests/util.rs index 0d28063ec79..d9d2ffc301d 100644 --- a/crates/data-plane-controller/tests/util.rs +++ b/crates/data-plane-controller/tests/util.rs @@ -175,6 +175,7 @@ pub fn initial_state() -> stack::State { preview_branch: String::new(), pending_refresh: false, pending_converge: false, // Start false, should be set by diffing + pinned_links: Vec::new(), publish_exports: None, publish_stack: None, } diff --git a/crates/flow-client/control-plane-api.graphql b/crates/flow-client/control-plane-api.graphql index 493721246ca..4f8a9dd1828 100644 --- a/crates/flow-client/control-plane-api.graphql +++ b/crates/flow-client/control-plane-api.graphql @@ -731,11 +731,11 @@ type DataPlane { """ azureApplicationClientId: String """ - Configured private link endpoints for this data-plane. Replacing this - list (via `updateDataPlanePrivateLinks`) triggers reconvergence by the - data-plane controller on its next poll. Returns an empty list to - callers that lack the `ViewDataPlanePrivateNetworking` capability on - this data plane. + Configured private links for this data-plane, each with its + controller-observed provisioning status. Mutating links (via + `addDataPlanePrivateLink` and friends) triggers reconvergence by the + data-plane controller. Returns an empty list to callers that lack the + `ViewDataPlanePrivateNetworking` capability on this data plane. """ privateLinks: [PrivateLink!]! """ @@ -1284,18 +1284,26 @@ type MutationRoot { """ deleteInviteLink(token: UUID!): Boolean! """ - Replaces the configured private link endpoints on a private data plane. - - The provided list overwrites the entire `private_links` column; partial - updates are intentionally not supported. The data-plane controller - converges to the new configuration on its next poll. Returns the desired - private links state. The `*LinkEndpoints` provisioning results are not echoed here: - they lag this write until the controller converges, so callers needing them re-query `dataPlanes`. - - Requires the `ModifyDataPlanePrivateNetworking` capability on the - private data-plane name. + Adds a private link to a private data plane. The data-plane controller + converges to provision it on its next poll; the returned link starts + `pending`. Requires `ModifyDataPlanePrivateNetworking` on the data plane. + """ + addDataPlanePrivateLink(dataPlaneName: String!, config: PrivateLinkConfigInput!): PrivateLink! + """ + Replaces the configuration of an existing private link by id. A changed + config resets the observed status to `pending` and re-triggers convergence: + the desired-edit trigger clears the observation columns and bumps the + link's internal generation, so a converge already in flight against the + previous configuration cannot later stamp this link with a stale status. + Requires `ModifyDataPlanePrivateNetworking` on the owning data plane. + """ + updateDataPlanePrivateLink(id: Id!, config: PrivateLinkConfigInput!): PrivateLink! """ - updateDataPlanePrivateLinks(dataPlaneName: String!, privateLinks: [PrivateLinkInput!]!): [PrivateLink!]! + Removes a private link by id. The controller tears down its endpoint on + the next converge. Requires `ModifyDataPlanePrivateNetworking` on the + owning data plane. Returns the removed link id. + """ + removeDataPlanePrivateLink(id: Id!): Id! """ Create a refresh token for the authenticated user. """ @@ -1438,22 +1446,73 @@ input PrefixesBy { minCapability: Capability! } +""" +A configured private link and its controller-observed provisioning status. +""" +type PrivateLink { + """ + Stable identifier of this private link. + """ + id: Id! + """ + The link configuration (AWS PrivateLink, Azure Private Link, or GCP PSC). + Its variant (`AWSPrivateLink`/`AzurePrivateLink`/`GCPPrivateServiceConnect`) + is the link's cloud provider. + """ + config: PrivateLinkConfig! + """ + Controller-observed provisioning status. + """ + status: PrivateLinkProvisioningStatus! + """ + Provider-specific provisioning details (DNS entries, IPs) once + provisioned; opaque JSON exported by the data-plane controller. + """ + details: JSON + """ + Failure detail when `status` is `failed`. + """ + error: String + """ + When the controller last observed this link's status. + """ + observedAt: DateTime +} + """ Private link configuration for a customer-owned data plane: AWS PrivateLink, Azure Private Link, or GCP Private Service Connect. """ -union PrivateLink = AWSPrivateLink | AzurePrivateLink | GCPPrivateServiceConnect +union PrivateLinkConfig = AWSPrivateLink | AzurePrivateLink | GCPPrivateServiceConnect """ Private link configuration for a customer-owned data plane: AWS PrivateLink, Azure Private Link, or GCP Private Service Connect. """ -input PrivateLinkInput @oneOf { +input PrivateLinkConfigInput @oneOf { aws: AWSPrivateLinkInput azure: AzurePrivateLinkInput gcp: GCPPrivateServiceConnectInput } +""" +Controller-observed provisioning status of a configured private link. +""" +enum PrivateLinkProvisioningStatus { + """ + Not yet provisioned for the current configuration. + """ + PENDING + """ + Provisioned; `details` describes the endpoint. + """ + PROVISIONED + """ + Provisioning failed; see `error`. + """ + FAILED +} + """ Filter connectors by their protocol (capture or materialization). """ diff --git a/crates/models/src/private_links.rs b/crates/models/src/private_links.rs index bf5ae139070..49b6f162fda 100644 --- a/crates/models/src/private_links.rs +++ b/crates/models/src/private_links.rs @@ -10,7 +10,7 @@ use serde::{Deserialize, Serialize}; #[cfg_attr( feature = "async-graphql", derive(async_graphql::Union, async_graphql::OneofObject), - graphql(name = "PrivateLink", input_name = "PrivateLinkInput") + graphql(name = "PrivateLinkConfig", input_name = "PrivateLinkConfigInput") )] #[serde(untagged)] pub enum PrivateLink { diff --git a/supabase/migrations/20260618120000_data_plane_private_links.sql b/supabase/migrations/20260618120000_data_plane_private_links.sql new file mode 100644 index 00000000000..25c23b4c36d --- /dev/null +++ b/supabase/migrations/20260618120000_data_plane_private_links.sql @@ -0,0 +1,116 @@ +-- Model private links as first-class rows with a stable identity and +-- data-plane-controller-owned observed status, replacing the flat +-- `data_planes.private_links` JSON array as the source of truth. The agent-api +-- resolvers and the controller both read this table directly, and the controller +-- writes each link's observed status back to it. There is no wake trigger: the +-- controller re-reads desired links from this table on its poll loop and +-- detects changes by config and generation diff. +-- +-- The legacy `data_planes.private_links` and `*_link_endpoints` columns are left +-- in place. A controller still running the pre-cutover binary during the rolling +-- deploy keeps reading `private_links`, so the column must survive the rollout; +-- nothing projects table changes back into it, so it is frozen at its +-- pre-migration value. A follow-up migration drops it and the endpoint columns +-- and recreates `data_planes_overview` once the fleet is fully cut over. Private +-- link edits are paused for the duration of the rolling deploy so the frozen +-- column and the table cannot disagree while both old and new controllers run. + +begin; + +create table internal.data_plane_private_links ( + id public.flowid primary key not null default internal.id_generator(), + data_plane_id public.flowid not null + references public.data_planes (id) on delete cascade, + -- The polymorphic link configuration: the same element shape as the legacy + -- `data_planes.private_links` array; round-trips `models::PrivateLink`. + config jsonb not null, + -- Cloud provider of the link, stored so consumers need not parse `config` + -- to learn the variant. It is generated so it cannot drift from the config + -- it describes. AWS and Azure links both key on `service_name`, so the + -- provider disambiguates them. + provider text generated always as + (case + when (config ->> 'service_attachment') is not null then 'gcp' + when (config ->> 'az_ids') is not null then 'aws' + else 'azure' + end) stored, + -- Monotonic version of the desired configuration, bumped by the + -- desired-edit trigger below whenever `config` changes. The + -- controller pins each link's `(id, generation)` when it reads desired state + -- for a converge and lands that converge's observed status only on rows whose + -- generation still matches, so an edit racing a converge cannot be stamped + -- with a status computed from the pre-edit configuration. + generation bigint not null default 1, + -- The provider's service identifier, used to enforce uniqueness. + -- Identities are only meaningful per provider (AWS and Azure both key on + -- `service_name`), and a same-provider duplicate produces colliding Pulumi + -- resource names in est-dry-dock, wedging the converge; uniqueness is + -- therefore scoped to (data_plane_id, provider, service_identity). + service_identity text generated always as + (coalesce(config ->> 'service_name', config ->> 'service_attachment')) stored not null, + -- Observed state, written by the data-plane controller from est-dry-dock's + -- per-link `link_results` export, addressed by each row's id. + status text not null default 'pending' check (status in ('pending', 'provisioned', 'failed')), + details jsonb, + error text, + observed_at timestamptz, + created_at timestamptz not null default now(), + updated_at timestamptz not null default now(), + constraint data_plane_private_links_unique_provider_identity + unique (data_plane_id, provider, service_identity) +); + +comment on table internal.data_plane_private_links is + 'Per-link private networking configuration (desired) and controller-observed status for a data plane.'; + +-- No separate data_plane_id index: the unique constraint's index leads with +-- data_plane_id and serves those lookups. + +-- No RLS and no grants. The table lives in `internal`, which PostgREST does not +-- expose (`config.toml` lists only `public`), so it is never reachable over the +-- REST API. Every reader connects as `postgres`: the agent-api resolvers, which +-- gate on the View/Modify private-networking capabilities, and the data-plane +-- controller. A row-level policy would only bind a PostgREST caller, and none +-- can reach an `internal` table. This mirrors `internal.invite_links`. + +-- Backfill one row per element of every existing `private_links` array. The +-- column already holds the source data, so column and table start consistent. +-- The generated identity's NOT NULL constraint and the unique constraint abort +-- the migration if the legacy data cannot satisfy the table's invariants. +insert into internal.data_plane_private_links (data_plane_id, config) +select + dp.id, + elem::jsonb +from public.data_planes dp, + lateral unnest(dp.private_links) as elem; + +-- Any change to the user-owned desired config invalidates the observed status: +-- bump the generation and clear the observation columns in the same write. The +-- invariant then holds for every writer, not just the API mutations but also +-- the hand edits support performs directly against this table. Scoped to +-- `update of config` so the controller's post-converge status write +-- (which sets only status/details/observed_at/updated_at) does not fire it and +-- does not bump the generation it just pinned. +-- Runs as invoker (not `security definer`): it only rewrites fields of the row +-- already being updated. +create function internal.on_data_plane_private_links_desired_edit() returns trigger + language plpgsql + set search_path to '' + as $$ +begin + new.generation := old.generation + 1; + new.status := 'pending'; + new.details := null; + new.error := null; + new.observed_at := null; + new.updated_at := now(); + return new; +end; +$$; + +create trigger data_plane_private_links_desired_edit + before update of config on internal.data_plane_private_links + for each row when (old.config is distinct from new.config) + execute function internal.on_data_plane_private_links_desired_edit(); + +commit; diff --git a/supabase/tests/data_plane_private_links.test.sql b/supabase/tests/data_plane_private_links.test.sql new file mode 100644 index 00000000000..ffef7ec0a6e --- /dev/null +++ b/supabase/tests/data_plane_private_links.test.sql @@ -0,0 +1,85 @@ +-- Exercises the `data_plane_private_links_desired_edit` trigger: a +-- config update bumps the generation and clears the observation +-- columns, for any writer. A controller-owned status update +-- (status/details/observed_at only) does not fire it: it must not bump the +-- generation the controller just pinned. Table changes also never project into +-- the parent data plane's legacy `private_links` column, which stays frozen +-- until it is dropped. + +create function tests.test_data_plane_private_links_trigger() +returns setof text as $$ +declare + v_dp_id flowid := '666666666666'; +begin + + insert into public.data_planes ( + id, data_plane_name, data_plane_fqdn, hmac_keys, encrypted_hmac_keys, + broker_address, reactor_address, ops_logs_name, ops_stats_name, + ops_l1_events_name, ops_l1_inferred_name, ops_l1_stats_name, + ops_l2_events_transform, ops_l2_inferred_transform, ops_l2_stats_transform, + enable_l2, cidr_blocks, aws_iam_user_arn, gcp_service_account_email, + azure_application_name, azure_application_client_id + ) values ( + v_dp_id, 'ops/dp/private/triggerCo/aws-1', 'dp.private.triggerCo', + '{c2VjcmV0}', '{}', 'broker.dp.private.triggerCo', 'reactor.dp.private.triggerCo', + 'ops/tasks/private/triggerCo/logs', 'ops/tasks/private/triggerCo/stats', + 'ops/rollups/L1/private/triggerCo/events', 'ops/rollups/L1/private/triggerCo/inferred', + 'ops/rollups/L1/private/triggerCo/stats', 'from.dp.private.triggerCo', + 'from.dp.private.triggerCo', 'from.dp.private.triggerCo', false, + '{10.30.0.0/16}', 'arn:aws:iam::444555666:user/test', + 'test-gcp@estuary-test.iam.gserviceaccount.com', 'estuary-test-app', + '66666666-6666-6666-6666-666666666666' + ); + + insert into internal.data_plane_private_links (id, data_plane_id, config) values + ('00:00:00:00:00:00:0d:01', v_dp_id, + '{"region":"us-east-1","az_ids":["a"],"service_name":"svc-x"}'::jsonb); + + return query select is( + (select array_length(private_links, 1) from public.data_planes where id = v_dp_id), + null, 'insert does not project into the legacy private_links column'); + return query select is( + (select generation from internal.data_plane_private_links where data_plane_id = v_dp_id), + 1::bigint, 'a freshly inserted link starts at generation 1'); + return query select is( + (select provider from internal.data_plane_private_links where data_plane_id = v_dp_id), + 'aws', 'provider is generated from config'); + + -- A controller-owned status update is outside the trigger's update-of scope, + -- so it does not bump the generation the controller pinned for it. + update internal.data_plane_private_links + set status = 'failed', details = '{}'::jsonb, error = 'boom', observed_at = now() + where data_plane_id = v_dp_id; + + return query select is( + (select generation from internal.data_plane_private_links where data_plane_id = v_dp_id), + 1::bigint, 'a status-only update does not bump the generation'); + + -- Assigning the existing config is a no-op: it must not invalidate a healthy + -- observation or schedule unnecessary infrastructure work. + update internal.data_plane_private_links + set config = config + where data_plane_id = v_dp_id; + + return query select ok( + (select generation = 1 and status = 'failed' and error = 'boom' + and details is not null and observed_at is not null + from internal.data_plane_private_links where data_plane_id = v_dp_id), + 'an identical config update preserves generation and observation'); + + -- An actual config change bumps the generation and clears the observed + -- status set just above. + update internal.data_plane_private_links + set config = config || '{"region":"us-west-2"}'::jsonb + where data_plane_id = v_dp_id; + + return query select is( + (select generation from internal.data_plane_private_links where data_plane_id = v_dp_id), + 2::bigint, 'config update bumps the generation'); + return query select ok( + (select status = 'pending' and details is null and error is null and observed_at is null + from internal.data_plane_private_links where data_plane_id = v_dp_id), + 'config update resets the observed status columns'); + +end +$$ language plpgsql;