flow: x-schema-name annotation for 2nd-to-last resource path

crates/agent/src/connector_tags.rs

@@ -163,6 +163,15 @@ impl TagHandler {
             tracing::warn!(image = %image_composed, "capture connector spec omits resource_path_pointers");
         }
 
+        // Validate that there is an x-collection-name annotation in the resource config schema
+        // of materialization connectors
+        if proto_type == RuntimeProtocol::Materialize {
+            if let Err(err) = crate::resource_configs::pointer_for_schema(resource_config_schema.get()) {
+                tracing::warn!(image = %image_composed, error = %err, "resource schema does not have x-collection-name annotation");
+                return Ok((row.tag_id, JobStatus::SpecFailed));
+            }
+        }
+
         // The tag fields may not be updated if the resource_path_pointers have
         // changed. If that happens, then we bail without making any changes
         // other than to job_status.

crates/agent/src/controllers/materialization.rs

@@ -5,10 +5,11 @@ use super::{
 use crate::{
     controllers::publication_status::PublicationStatus,
     publications::{PublicationResult, RejectedField},
+    resource_configs::ResourceSpecPointers,
 };
 use anyhow::Context;
 use itertools::Itertools;
-use models::{ModelDef, OnIncompatibleSchemaChange};
+use models::{ModelDef, OnIncompatibleSchemaChange, SourceCapture};
 use proto_flow::materialize::response::validated::constraint::Type as ConstraintType;
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
@@ -58,11 +59,11 @@ impl MaterializationStatus {
             }
         }
 
-        if let Some(source_capture_name) = &model.source_capture {
+        if let Some(source_capture) = &model.source_capture {
             // If the source capture has been deleted, we will have already handled that as a
             // part of `handle_deleted_dependencies`.
             if let Some(source_capture_model) =
-                dependencies.live.captures.get_by_key(source_capture_name)
+                dependencies.live.captures.get_by_key(&source_capture.capture_name())
             {
                 if self.source_capture.is_none() {
                     self.source_capture = Some(SourceCaptureStatus::default());
@@ -264,10 +265,10 @@ fn handle_deleted_dependencies(
     if drafted
         .source_capture
         .as_ref()
-        .map(|sc| deleted.contains(sc.as_str()))
+        .map(|sc| deleted.contains(sc.capture_name().as_str()))
         .unwrap_or(false)
     {
-        let capture_name = drafted.source_capture.take().unwrap();
+        let capture_name = drafted.source_capture.take().unwrap().capture_name();
         source_capture.take();
         descriptions.push(format!(
             r#"removed sourceCapture: "{capture_name}" because the capture was deleted"#
@@ -335,7 +336,7 @@ impl SourceCaptureStatus {
             .get_connector_spec(config.image.clone())
             .await
             .context("failed to fetch connector spec")?;
-        let collection_name_pointer = crate::resource_configs::pointer_for_schema(
+        let resource_spec_pointers = crate::resource_configs::pointer_for_schema(
             connector_spec.resource_config_schema.get(),
         )?;
 
@@ -370,7 +371,8 @@ impl SourceCaptureStatus {
         draft_row.is_touch = false;
 
         update_linked_materialization(
-            collection_name_pointer,
+            model.source_capture.as_ref().unwrap(),
+            resource_spec_pointers,
             &self.add_bindings,
             draft_row.model.as_mut().unwrap(),
         )?;
@@ -398,15 +400,17 @@ fn get_bindings_to_add(
 }
 
 fn update_linked_materialization(
-    resource_collection_name_ptr: doc::Pointer,
+    source_capture: &SourceCapture,
+    resource_spec_pointers: ResourceSpecPointers,
     bindings_to_add: &BTreeSet<models::Collection>,
     materialization: &mut models::MaterializationDef,
 ) -> anyhow::Result<()> {
     for collection_name in bindings_to_add {
         let mut resource_spec = serde_json::json!({});
         crate::resource_configs::update_materialization_resource_spec(
+            source_capture,
             &mut resource_spec,
-            &resource_collection_name_ptr,
+            &resource_spec_pointers,
             &collection_name,
         )?;
 

crates/agent/src/controllers/mod.rs

@@ -457,6 +457,7 @@ mod test {
     use std::collections::{BTreeSet, VecDeque};
 
     use chrono::TimeZone;
+    use models::Capture;
 
     use super::*;
     use crate::controllers::materialization::SourceCaptureStatus;

crates/agent/src/controllers/snapshots/agent__controllers__test__materialization-status-round-trip.snap

@@ -1,6 +1,6 @@
 ---
 source: crates/agent/src/controllers/mod.rs
-expression: "StatusSnapshot { starting: status, json: as_json, parsed: round_tripped }"
+expression: "StatusSnapshot { starting: status, json: as_json, parsed: round_tripped, }"
 ---
 StatusSnapshot {
     starting: Materialization(

crates/agent/src/integration_tests/dependencies_and_activations.rs

@@ -75,7 +75,7 @@ async fn test_dependencies_and_controllers() {
                 "sourceCapture": "owls/capture",
                 "endpoint": {
                     "connector": {
-                        "image": "ghcr.io/estuary/materialize-postgres:dev",
+                        "image": "materialize/test:test",
                         "config": {}
                     }
                 },

crates/agent/src/integration_tests/harness.rs

@@ -168,7 +168,28 @@ impl TestHarness {
                     'materialization',
                     'http://test.test/',
                     '{"type": "object"}',
-                    '{"type": "object", "properties": {"id": {"type": "string", "x-collection-name": true}}}',
+                    '{"type": "object", "properties": {"id": {"type": "string", "x-collection-name": true}, "schema": {"type": "string", "x-schema-name": true}, "delta": {"type": "boolean", "x-delta-updates": true}}}',
+                    '{/id}',
+                    '{"type": "success"}'
+                ) on conflict do nothing
+            ),
+            materialize_tag_no_annotations as (
+                insert into connector_tags (
+                    connector_id,
+                    image_tag,
+                    protocol,
+                    documentation_url,
+                    endpoint_spec_schema,
+                    resource_spec_schema,
+                    resource_path_pointers,
+                    job_status
+                ) values (
+                    (select id from materialize_image),
+                    ':test-no-annotation',
+                    'materialization',
+                    'http://test.test/',
+                    '{"type": "object"}',
+                    '{"type": "object", "properties": {"id": {"type": "string", "x-collection-name": true}, "schema": {"type": "string"}, "delta": {"type": "boolean"}}}',
                     '{/id}',
                     '{"type": "success"}'
                 ) on conflict do nothing